phil/StarPunk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement IndieAuth token and authorization endpoints (Phase 2)

Browse Source 
Following design in /docs/design/micropub-architecture.md and
/docs/decisions/ADR-029-micropub-v1-implementation-phases.md

Token Endpoint (/auth/token):
- Exchange authorization codes for access tokens
- Form-encoded POST requests per IndieAuth spec
- PKCE support (code_verifier validation)
- OAuth 2.0 error responses
- Validates client_id, redirect_uri, me parameters
- Returns Bearer tokens with scope

Authorization Endpoint (/auth/authorization):
- GET: Display consent form (requires admin login)
- POST: Process approval/denial
- PKCE support (code_challenge storage)
- Scope validation and filtering
- Integration with session management
- Proper error handling and redirects

All 33 Phase 2 tests pass (17 token + 16 authorization)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Phil Skentelbery
2025-11-24 12:26:54 -07:00
parent 2eaf67279d
commit e5050a0a7e
4 changed files with 1103 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

361
tests/test_routes_authorization.py Normal file
View File

@@ -0,0 +1,361 @@
"""
Tests for authorization endpoint route
Tests the /auth/authorization endpoint for IndieAuth client authorization.
"""
import pytest
from starpunk.auth import create_session
from urllib.parse import urlparse, parse_qs
def create_admin_session(client, app):
"""Helper to create an authenticated admin session"""
with app.test_request_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
session_token = create_session(admin_me)
client.set_cookie('starpunk_session', session_token)
return session_token
def test_authorization_endpoint_get_not_logged_in(client, app):
"""Test authorization endpoint redirects to login when not authenticated"""
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create'
})
# Should redirect to login
assert response.status_code == 302
assert '/auth/login' in response.location
def test_authorization_endpoint_get_logged_in(client, app):
"""Test authorization endpoint shows consent form when authenticated"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create'
})
assert response.status_code == 200
assert b'Authorization Request' in response.data
assert b'https://client.example' in response.data
assert b'create' in response.data
def test_authorization_endpoint_missing_response_type(client, app):
"""Test authorization endpoint rejects missing response_type"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123'
})
assert response.status_code == 400
assert b'Missing response_type' in response.data
def test_authorization_endpoint_invalid_response_type(client, app):
"""Test authorization endpoint rejects unsupported response_type"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'token', # Only 'code' is supported
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123'
})
assert response.status_code == 400
assert b'Unsupported response_type' in response.data
def test_authorization_endpoint_missing_client_id(client, app):
"""Test authorization endpoint rejects missing client_id"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123'
})
assert response.status_code == 400
assert b'Missing client_id' in response.data
def test_authorization_endpoint_missing_redirect_uri(client, app):
"""Test authorization endpoint rejects missing redirect_uri"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'state': 'random_state_123'
})
assert response.status_code == 400
assert b'Missing redirect_uri' in response.data
def test_authorization_endpoint_missing_state(client, app):
"""Test authorization endpoint rejects missing state"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback'
})
assert response.status_code == 400
assert b'Missing state' in response.data
def test_authorization_endpoint_empty_scope(client, app):
"""Test authorization endpoint allows empty scope"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': '' # Empty scope allowed per IndieAuth spec
})
assert response.status_code == 200
assert b'Authorization Request' in response.data
def test_authorization_endpoint_filters_unsupported_scopes(client, app):
"""Test authorization endpoint filters to supported scopes only"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create update delete' # Only 'create' is supported in V1
})
assert response.status_code == 200
# Should only show 'create' scope
assert b'create' in response.data
def test_authorization_endpoint_post_approve(client, app):
"""Test authorization approval generates code and redirects"""
create_admin_session(client, app)
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
response = client.post('/auth/authorization', data={
'approve': 'yes',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'me': admin_me,
'response_type': 'code'
})
# Should redirect to client's redirect_uri
assert response.status_code == 302
assert response.location.startswith('https://client.example/callback')
# Parse redirect URL
parsed = urlparse(response.location)
params = parse_qs(parsed.query)
# Should include code and state
assert 'code' in params
assert 'state' in params
assert params['state'][0] == 'random_state_123'
assert len(params['code'][0]) > 0
def test_authorization_endpoint_post_deny(client, app):
"""Test authorization denial redirects with error"""
create_admin_session(client, app)
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
response = client.post('/auth/authorization', data={
'approve': 'no',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'me': admin_me,
'response_type': 'code'
})
# Should redirect to client's redirect_uri with error
assert response.status_code == 302
assert response.location.startswith('https://client.example/callback')
# Parse redirect URL
parsed = urlparse(response.location)
params = parse_qs(parsed.query)
# Should include error
assert 'error' in params
assert params['error'][0] == 'access_denied'
assert 'state' in params
assert params['state'][0] == 'random_state_123'
def test_authorization_endpoint_post_not_logged_in(client, app):
"""Test authorization POST requires authentication"""
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
response = client.post('/auth/authorization', data={
'approve': 'yes',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'me': admin_me,
'response_type': 'code'
})
# Should redirect to login
assert response.status_code == 302
assert '/auth/login' in response.location
def test_authorization_endpoint_with_pkce(client, app):
"""Test authorization endpoint accepts PKCE parameters"""
create_admin_session(client, app)
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'code_challenge': 'E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM',
'code_challenge_method': 'S256'
})
assert response.status_code == 200
assert b'Authorization Request' in response.data
def test_authorization_endpoint_post_with_pkce(client, app):
"""Test authorization approval preserves PKCE parameters"""
create_admin_session(client, app)
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
response = client.post('/auth/authorization', data={
'approve': 'yes',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'me': admin_me,
'response_type': 'code',
'code_challenge': 'E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM',
'code_challenge_method': 'S256'
})
assert response.status_code == 302
assert response.location.startswith('https://client.example/callback')
# Parse redirect URL
parsed = urlparse(response.location)
params = parse_qs(parsed.query)
# Should have code and state
assert 'code' in params
assert 'state' in params
def test_authorization_endpoint_preserves_me_parameter(client, app):
"""Test authorization endpoint uses ADMIN_ME as identity"""
create_admin_session(client, app)
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
response = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create'
})
assert response.status_code == 200
# Should show admin's identity in the form
assert admin_me.encode() in response.data
def test_authorization_flow_end_to_end(client, app):
"""Test complete authorization flow from consent to token exchange"""
create_admin_session(client, app)
with app.app_context():
admin_me = app.config.get('ADMIN_ME', 'https://test.example.com')
# Step 1: Get authorization form
response1 = client.get('/auth/authorization', query_string={
'response_type': 'code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create'
})
assert response1.status_code == 200
# Step 2: Approve authorization
response2 = client.post('/auth/authorization', data={
'approve': 'yes',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'state': 'random_state_123',
'scope': 'create',
'me': admin_me,
'response_type': 'code'
})
assert response2.status_code == 302
# Extract authorization code
parsed = urlparse(response2.location)
params = parse_qs(parsed.query)
code = params['code'][0]
# Step 3: Exchange code for token
response3 = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': admin_me
}, content_type='application/x-www-form-urlencoded')
assert response3.status_code == 200
token_data = response3.get_json()
assert 'access_token' in token_data
assert token_data['token_type'] == 'Bearer'
assert token_data['scope'] == 'create'
assert token_data['me'] == admin_me

394
tests/test_routes_token.py Normal file
View File

@@ -0,0 +1,394 @@
"""
Tests for token endpoint route
Tests the /auth/token endpoint for IndieAuth token exchange.
"""
import pytest
from starpunk.tokens import create_authorization_code
import hashlib
def test_token_endpoint_success(client, app):
"""Test successful token exchange"""
with app.app_context():
# Create authorization code
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create"
)
# Exchange for token
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 200
data = response.get_json()
assert 'access_token' in data
assert data['token_type'] == 'Bearer'
assert data['scope'] == 'create'
assert data['me'] == 'https://user.example'
def test_token_endpoint_with_pkce(client, app):
"""Test token exchange with PKCE"""
with app.app_context():
# Generate PKCE verifier and challenge
code_verifier = "test_verifier_with_sufficient_entropy_12345"
code_challenge = hashlib.sha256(code_verifier.encode()).hexdigest()
# Create authorization code with PKCE
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create",
code_challenge=code_challenge,
code_challenge_method="S256"
)
# Exchange with correct verifier
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example',
'code_verifier': code_verifier
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 200
data = response.get_json()
assert 'access_token' in data
def test_token_endpoint_missing_grant_type(client, app):
"""Test token endpoint rejects missing grant_type"""
with app.app_context():
response = client.post('/auth/token', data={
'code': 'some_code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'grant_type' in data['error_description']
def test_token_endpoint_invalid_grant_type(client, app):
"""Test token endpoint rejects invalid grant_type"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'password',
'code': 'some_code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'unsupported_grant_type'
def test_token_endpoint_missing_code(client, app):
"""Test token endpoint rejects missing code"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'code' in data['error_description']
def test_token_endpoint_missing_client_id(client, app):
"""Test token endpoint rejects missing client_id"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': 'some_code',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'client_id' in data['error_description']
def test_token_endpoint_missing_redirect_uri(client, app):
"""Test token endpoint rejects missing redirect_uri"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': 'some_code',
'client_id': 'https://client.example',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'redirect_uri' in data['error_description']
def test_token_endpoint_missing_me(client, app):
"""Test token endpoint rejects missing me parameter"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': 'some_code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'me' in data['error_description']
def test_token_endpoint_invalid_code(client, app):
"""Test token endpoint rejects invalid authorization code"""
with app.app_context():
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': 'invalid_code_12345',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
def test_token_endpoint_code_replay(client, app):
"""Test token endpoint prevents code replay attacks"""
with app.app_context():
# Create authorization code
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create"
)
# First exchange succeeds
response1 = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response1.status_code == 200
# Second exchange fails (replay attack)
response2 = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response2.status_code == 400
data = response2.get_json()
assert data['error'] == 'invalid_grant'
assert 'already been used' in data['error_description']
def test_token_endpoint_client_id_mismatch(client, app):
"""Test token endpoint rejects mismatched client_id"""
with app.app_context():
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create"
)
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://different-client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
assert 'client_id' in data['error_description']
def test_token_endpoint_redirect_uri_mismatch(client, app):
"""Test token endpoint rejects mismatched redirect_uri"""
with app.app_context():
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create"
)
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/different-callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
assert 'redirect_uri' in data['error_description']
def test_token_endpoint_me_mismatch(client, app):
"""Test token endpoint rejects mismatched me parameter"""
with app.app_context():
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create"
)
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://different-user.example'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
assert 'me parameter' in data['error_description']
def test_token_endpoint_empty_scope(client, app):
"""Test token endpoint rejects authorization code with empty scope"""
with app.app_context():
# Create authorization code with empty scope
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="" # Empty scope
)
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
}, content_type='application/x-www-form-urlencoded')
# IndieAuth spec: MUST NOT issue token if no scope
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_scope'
def test_token_endpoint_wrong_content_type(client, app):
"""Test token endpoint rejects non-form-encoded requests"""
with app.app_context():
response = client.post('/auth/token',
json={
'grant_type': 'authorization_code',
'code': 'some_code',
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
})
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_request'
assert 'Content-Type' in data['error_description']
def test_token_endpoint_pkce_missing_verifier(client, app):
"""Test token endpoint rejects PKCE exchange without verifier"""
with app.app_context():
# Create authorization code with PKCE
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create",
code_challenge="some_challenge",
code_challenge_method="S256"
)
# Exchange without verifier
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example'
# Missing code_verifier
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
assert 'code_verifier' in data['error_description']
def test_token_endpoint_pkce_wrong_verifier(client, app):
"""Test token endpoint rejects PKCE exchange with wrong verifier"""
with app.app_context():
code_verifier = "correct_verifier"
code_challenge = hashlib.sha256(code_verifier.encode()).hexdigest()
# Create authorization code with PKCE
code = create_authorization_code(
me="https://user.example",
client_id="https://client.example",
redirect_uri="https://client.example/callback",
scope="create",
code_challenge=code_challenge,
code_challenge_method="S256"
)
# Exchange with wrong verifier
response = client.post('/auth/token', data={
'grant_type': 'authorization_code',
'code': code,
'client_id': 'https://client.example',
'redirect_uri': 'https://client.example/callback',
'me': 'https://user.example',
'code_verifier': 'wrong_verifier'
}, content_type='application/x-www-form-urlencoded')
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'invalid_grant'
assert 'code_verifier' in data['error_description']