add uncommitted plans and CLAUDE.md
This commit is contained in:
parent
6b4cbdc152
commit
fb133f9cba
10 changed files with 5241 additions and 0 deletions
851
docs/plans/2026-02-18-consent-screen-plan.md
Normal file
851
docs/plans/2026-02-18-consent-screen-plan.md
Normal file
|
|
@ -0,0 +1,851 @@
|
|||
# Consent Screen Implementation Plan
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** Add a consent screen to the OIDC authorization flow so users
|
||||
approve scopes before tokens are issued. Consent is saved per user+client
|
||||
and only re-shown when requested scopes change.
|
||||
|
||||
**Architecture:** New `user_consents` SQLite table stores approved scopes
|
||||
per user+client. A consent check runs after authentication in the
|
||||
authorization flow. If consent is needed, the user sees a scope approval
|
||||
page. The manage-app client bypasses consent. Partial consent (unchecking
|
||||
optional scopes) is supported.
|
||||
|
||||
**Tech Stack:** SQLite migration, aiosqlite repository, Pydantic model,
|
||||
Jinja2 template, existing FastAPI routing patterns.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Add Consent model and SQLite migration
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/porchlight/models.py`
|
||||
- Create: `src/porchlight/store/sqlite/migrations/002_user_consents.sql`
|
||||
- Modify: `src/porchlight/store/protocols.py`
|
||||
- Modify: `src/porchlight/store/sqlite/repositories.py`
|
||||
- Test: `tests/test_store/test_sqlite_consent_repo.py`
|
||||
|
||||
**Step 1: Write the failing tests**
|
||||
|
||||
Add `tests/test_store/test_sqlite_consent_repo.py`:
|
||||
|
||||
```python
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from porchlight.models import Consent
|
||||
from porchlight.store.protocols import ConsentRepository
|
||||
from porchlight.store.sqlite.repositories import SQLiteConsentRepository
|
||||
|
||||
|
||||
async def test_implements_protocol(db_connection) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
assert isinstance(repo, ConsentRepository)
|
||||
|
||||
|
||||
async def test_set_and_get_consent(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
await repo.set_consent(sample_user.userid, "test-rp", ["openid", "profile"])
|
||||
|
||||
consent = await repo.get_consent(sample_user.userid, "test-rp")
|
||||
assert consent is not None
|
||||
assert consent.userid == sample_user.userid
|
||||
assert consent.client_id == "test-rp"
|
||||
assert consent.scopes == ["openid", "profile"]
|
||||
assert isinstance(consent.created_at, datetime)
|
||||
assert isinstance(consent.updated_at, datetime)
|
||||
|
||||
|
||||
async def test_get_consent_not_found(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
consent = await repo.get_consent(sample_user.userid, "nonexistent")
|
||||
assert consent is None
|
||||
|
||||
|
||||
async def test_set_consent_upserts(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
await repo.set_consent(sample_user.userid, "test-rp", ["openid"])
|
||||
await repo.set_consent(sample_user.userid, "test-rp", ["openid", "profile", "email"])
|
||||
|
||||
consent = await repo.get_consent(sample_user.userid, "test-rp")
|
||||
assert consent is not None
|
||||
assert consent.scopes == ["openid", "profile", "email"]
|
||||
|
||||
|
||||
async def test_delete_consent(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
await repo.set_consent(sample_user.userid, "test-rp", ["openid"])
|
||||
|
||||
result = await repo.delete_consent(sample_user.userid, "test-rp")
|
||||
assert result is True
|
||||
|
||||
consent = await repo.get_consent(sample_user.userid, "test-rp")
|
||||
assert consent is None
|
||||
|
||||
|
||||
async def test_delete_consent_not_found(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
result = await repo.delete_consent(sample_user.userid, "nonexistent")
|
||||
assert result is False
|
||||
|
||||
|
||||
async def test_list_consents(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
await repo.set_consent(sample_user.userid, "rp-a", ["openid"])
|
||||
await repo.set_consent(sample_user.userid, "rp-b", ["openid", "profile"])
|
||||
|
||||
consents = await repo.list_consents(sample_user.userid)
|
||||
assert len(consents) == 2
|
||||
client_ids = {c.client_id for c in consents}
|
||||
assert client_ids == {"rp-a", "rp-b"}
|
||||
|
||||
|
||||
async def test_list_consents_empty(db_connection, sample_user) -> None:
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
consents = await repo.list_consents(sample_user.userid)
|
||||
assert consents == []
|
||||
|
||||
|
||||
async def test_consent_deleted_on_user_cascade(db_connection, sample_user) -> None:
|
||||
"""Consent records are deleted when the user is deleted (CASCADE)."""
|
||||
from porchlight.store.sqlite.repositories import SQLiteUserRepository
|
||||
|
||||
repo = SQLiteConsentRepository(db_connection)
|
||||
user_repo = SQLiteUserRepository(db_connection)
|
||||
|
||||
await repo.set_consent(sample_user.userid, "test-rp", ["openid"])
|
||||
await user_repo.delete(sample_user.userid)
|
||||
|
||||
consent = await repo.get_consent(sample_user.userid, "test-rp")
|
||||
assert consent is None
|
||||
```
|
||||
|
||||
This test file uses fixtures `db_connection` and `sample_user` that should
|
||||
already exist in `tests/conftest.py`. Check `tests/conftest.py` for the
|
||||
exact fixture names — they may be named differently (e.g. `db` instead of
|
||||
`db_connection`). Adapt the test to match.
|
||||
|
||||
**Step 2: Run tests to verify they fail**
|
||||
|
||||
Run: `uv run python -m pytest tests/test_store/test_sqlite_consent_repo.py -v`
|
||||
Expected: FAIL — `Consent` model not defined, `ConsentRepository` not
|
||||
defined, `SQLiteConsentRepository` not defined.
|
||||
|
||||
**Step 3: Write the implementation**
|
||||
|
||||
Add to `src/porchlight/models.py`:
|
||||
|
||||
```python
|
||||
class Consent(BaseModel):
|
||||
userid: str
|
||||
client_id: str
|
||||
scopes: list[str]
|
||||
created_at: datetime = Field(default_factory=_utcnow)
|
||||
updated_at: datetime = Field(default_factory=_utcnow)
|
||||
```
|
||||
|
||||
Create `src/porchlight/store/sqlite/migrations/002_user_consents.sql`:
|
||||
|
||||
```sql
|
||||
CREATE TABLE user_consents (
|
||||
userid TEXT NOT NULL REFERENCES users(userid) ON DELETE CASCADE,
|
||||
client_id TEXT NOT NULL,
|
||||
scopes TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (userid, client_id)
|
||||
);
|
||||
```
|
||||
|
||||
Add to `src/porchlight/store/protocols.py`:
|
||||
|
||||
```python
|
||||
from porchlight.models import Consent # add to existing imports
|
||||
|
||||
@runtime_checkable
|
||||
class ConsentRepository(Protocol):
|
||||
async def get_consent(self, userid: str, client_id: str) -> Consent | None: ...
|
||||
|
||||
async def set_consent(self, userid: str, client_id: str, scopes: list[str]) -> None: ...
|
||||
|
||||
async def delete_consent(self, userid: str, client_id: str) -> bool: ...
|
||||
|
||||
async def list_consents(self, userid: str) -> list[Consent]: ...
|
||||
```
|
||||
|
||||
Add to `src/porchlight/store/sqlite/repositories.py`:
|
||||
|
||||
```python
|
||||
import json # add to existing imports
|
||||
from porchlight.models import Consent # add to existing imports
|
||||
|
||||
|
||||
class SQLiteConsentRepository:
|
||||
def __init__(self, db: aiosqlite.Connection) -> None:
|
||||
self._db = db
|
||||
|
||||
async def get_consent(self, userid: str, client_id: str) -> Consent | None:
|
||||
async with self._db.execute(
|
||||
"SELECT * FROM user_consents WHERE userid = ? AND client_id = ?",
|
||||
(userid, client_id),
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return Consent(
|
||||
userid=row["userid"],
|
||||
client_id=row["client_id"],
|
||||
scopes=json.loads(row["scopes"]),
|
||||
created_at=datetime.fromisoformat(row["created_at"]),
|
||||
updated_at=datetime.fromisoformat(row["updated_at"]),
|
||||
)
|
||||
|
||||
async def set_consent(self, userid: str, client_id: str, scopes: list[str]) -> None:
|
||||
now = datetime.now(UTC).isoformat()
|
||||
await self._db.execute(
|
||||
"""
|
||||
INSERT INTO user_consents (userid, client_id, scopes, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT (userid, client_id)
|
||||
DO UPDATE SET scopes = excluded.scopes, updated_at = excluded.updated_at
|
||||
""",
|
||||
(userid, client_id, json.dumps(scopes), now, now),
|
||||
)
|
||||
await self._db.commit()
|
||||
|
||||
async def delete_consent(self, userid: str, client_id: str) -> bool:
|
||||
cursor = await self._db.execute(
|
||||
"DELETE FROM user_consents WHERE userid = ? AND client_id = ?",
|
||||
(userid, client_id),
|
||||
)
|
||||
await self._db.commit()
|
||||
return cursor.rowcount > 0
|
||||
|
||||
async def list_consents(self, userid: str) -> list[Consent]:
|
||||
async with self._db.execute(
|
||||
"SELECT * FROM user_consents WHERE userid = ? ORDER BY client_id",
|
||||
(userid,),
|
||||
) as cursor:
|
||||
rows = await cursor.fetchall()
|
||||
return [
|
||||
Consent(
|
||||
userid=row["userid"],
|
||||
client_id=row["client_id"],
|
||||
scopes=json.loads(row["scopes"]),
|
||||
created_at=datetime.fromisoformat(row["created_at"]),
|
||||
updated_at=datetime.fromisoformat(row["updated_at"]),
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
```
|
||||
|
||||
**Step 4: Run tests to verify they pass**
|
||||
|
||||
Run: `uv run python -m pytest tests/test_store/test_sqlite_consent_repo.py -v`
|
||||
Expected: All PASS.
|
||||
|
||||
**Step 5: Run full test suite**
|
||||
|
||||
Run: `uv run python -m pytest -v`
|
||||
Expected: All tests PASS (existing + new).
|
||||
|
||||
**Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add src/porchlight/models.py src/porchlight/store/protocols.py \
|
||||
src/porchlight/store/sqlite/migrations/002_user_consents.sql \
|
||||
src/porchlight/store/sqlite/repositories.py \
|
||||
tests/test_store/test_sqlite_consent_repo.py
|
||||
git commit -m "feat: add Consent model, migration, and repository"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Add consent check to authorization flow
|
||||
|
||||
**Files:**
|
||||
- Modify: `src/porchlight/app.py:36-38` (add consent_repo to app.state)
|
||||
- Modify: `src/porchlight/oidc/endpoints.py:45-97,100-149`
|
||||
- Test: `tests/test_oidc/test_consent_flow.py`
|
||||
|
||||
**Step 1: Write the failing tests**
|
||||
|
||||
Add `tests/test_oidc/test_consent_flow.py`:
|
||||
|
||||
```python
|
||||
import secrets
|
||||
from datetime import UTC, datetime
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from argon2 import PasswordHasher
|
||||
from httpx import AsyncClient
|
||||
|
||||
from porchlight.authn.password import PasswordService
|
||||
from porchlight.models import PasswordCredential, User
|
||||
|
||||
|
||||
async def test_authorization_shows_consent_for_new_client(client: AsyncClient) -> None:
|
||||
"""First-time authorization for an RP should redirect to /consent."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
|
||||
# Login
|
||||
await client.post(
|
||||
"/login/password",
|
||||
data={"username": "consentuser", "password": "testpass"},
|
||||
headers={"HX-Request": "true"},
|
||||
)
|
||||
|
||||
# Authorization request
|
||||
res = await client.get(
|
||||
"/authorization",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "consent-rp",
|
||||
"redirect_uri": "http://localhost:9000/callback",
|
||||
"scope": "openid profile",
|
||||
"state": "teststate",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
assert "/consent" in res.headers["location"]
|
||||
|
||||
|
||||
async def test_consent_page_renders(client: AsyncClient) -> None:
|
||||
"""GET /consent should render the consent form."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
await _login_and_start_auth(client)
|
||||
|
||||
res = await client.get("/consent")
|
||||
assert res.status_code == 200
|
||||
assert "consent-rp" in res.text
|
||||
assert "profile" in res.text.lower()
|
||||
|
||||
|
||||
async def test_consent_allow_redirects_with_code(client: AsyncClient) -> None:
|
||||
"""Approving consent should complete the authorization flow."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
await _login_and_start_auth(client)
|
||||
|
||||
res = await client.post(
|
||||
"/consent",
|
||||
data={"action": "allow", "scope": ["openid", "profile"]},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
location = res.headers["location"]
|
||||
parsed = urlparse(location)
|
||||
params = parse_qs(parsed.query)
|
||||
assert "code" in params
|
||||
|
||||
|
||||
async def test_consent_deny_redirects_with_error(client: AsyncClient) -> None:
|
||||
"""Denying consent should redirect with access_denied error."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
await _login_and_start_auth(client)
|
||||
|
||||
res = await client.post(
|
||||
"/consent",
|
||||
data={"action": "deny"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
location = res.headers["location"]
|
||||
parsed = urlparse(location)
|
||||
params = parse_qs(parsed.query)
|
||||
assert params["error"] == ["access_denied"]
|
||||
|
||||
|
||||
async def test_saved_consent_skips_consent_screen(client: AsyncClient) -> None:
|
||||
"""Second authorization with same scopes should skip consent."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
|
||||
# First flow: login, authorize, consent
|
||||
await _login_and_start_auth(client)
|
||||
await client.post(
|
||||
"/consent",
|
||||
data={"action": "allow", "scope": ["openid", "profile"]},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
# Second flow: same scopes, should skip consent
|
||||
res = await client.get(
|
||||
"/authorization",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "consent-rp",
|
||||
"redirect_uri": "http://localhost:9000/callback",
|
||||
"scope": "openid profile",
|
||||
"state": "teststate2",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
location = res.headers["location"]
|
||||
# Should redirect directly to callback, not /consent
|
||||
assert "callback" in location
|
||||
assert "code" in location
|
||||
|
||||
|
||||
async def test_new_scopes_reshows_consent(client: AsyncClient) -> None:
|
||||
"""If RP requests new scopes, consent screen should reappear."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
|
||||
# First flow: consent to openid only
|
||||
await _login_and_start_auth(client, scope="openid")
|
||||
await client.post(
|
||||
"/consent",
|
||||
data={"action": "allow", "scope": ["openid"]},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
# Second flow: request openid + profile (new scope)
|
||||
res = await client.get(
|
||||
"/authorization",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "consent-rp",
|
||||
"redirect_uri": "http://localhost:9000/callback",
|
||||
"scope": "openid profile",
|
||||
"state": "teststate2",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
assert "/consent" in res.headers["location"]
|
||||
|
||||
|
||||
async def test_manage_app_skips_consent(client: AsyncClient) -> None:
|
||||
"""The manage-app client should bypass consent entirely."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
settings = app.state.settings
|
||||
await _create_test_user(app)
|
||||
|
||||
await client.post(
|
||||
"/login/password",
|
||||
data={"username": "consentuser", "password": "testpass"},
|
||||
headers={"HX-Request": "true"},
|
||||
)
|
||||
|
||||
manage_cdb = app.state.oidc_server.context.cdb[settings.manage_client_id]
|
||||
redirect_uri = manage_cdb["redirect_uris"][0][0]
|
||||
|
||||
res = await client.get(
|
||||
"/authorization",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": settings.manage_client_id,
|
||||
"redirect_uri": redirect_uri,
|
||||
"scope": "openid profile email",
|
||||
"state": "teststate",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
location = res.headers["location"]
|
||||
# Should redirect directly to callback, not /consent
|
||||
assert "code" in location
|
||||
assert "/consent" not in location
|
||||
|
||||
|
||||
async def test_partial_consent_filters_scopes(client: AsyncClient) -> None:
|
||||
"""User can approve only some scopes (partial consent)."""
|
||||
app = client._transport.app # type: ignore[union-attr]
|
||||
_register_test_rp(app)
|
||||
await _create_test_user(app)
|
||||
|
||||
# Request openid + profile + email, approve only openid + profile
|
||||
await _login_and_start_auth(client, scope="openid profile email")
|
||||
res = await client.post(
|
||||
"/consent",
|
||||
data={"action": "allow", "scope": ["openid", "profile"]},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert res.status_code == 303
|
||||
location = res.headers["location"]
|
||||
assert "code" in location
|
||||
|
||||
# Verify consent was saved with only the approved scopes
|
||||
consent_repo = app.state.consent_repo
|
||||
consent = await consent_repo.get_consent("lusab-consent", "consent-rp")
|
||||
assert consent is not None
|
||||
assert set(consent.scopes) == {"openid", "profile"}
|
||||
|
||||
|
||||
# -- Test helpers --
|
||||
|
||||
def _register_test_rp(app) -> None:
|
||||
oidc_server = app.state.oidc_server
|
||||
if "consent-rp" in oidc_server.context.cdb:
|
||||
return
|
||||
client_id = "consent-rp"
|
||||
client_secret = "consent-secret-0123456789abcdef"
|
||||
oidc_server.context.cdb[client_id] = {
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"redirect_uris": [("http://localhost:9000/callback", {})],
|
||||
"response_types_supported": ["code"],
|
||||
"token_endpoint_auth_method": "client_secret_basic",
|
||||
"scope": ["openid", "profile", "email"],
|
||||
"allowed_scopes": ["openid", "profile", "email"],
|
||||
"client_salt": secrets.token_hex(8),
|
||||
}
|
||||
oidc_server.keyjar.add_symmetric(client_id, client_secret)
|
||||
|
||||
|
||||
async def _create_test_user(app) -> None:
|
||||
user_repo = app.state.user_repo
|
||||
existing = await user_repo.get_by_username("consentuser")
|
||||
if existing:
|
||||
return
|
||||
user = User(
|
||||
userid="lusab-consent",
|
||||
username="consentuser",
|
||||
email="consent@example.com",
|
||||
created_at=datetime.now(UTC),
|
||||
updated_at=datetime.now(UTC),
|
||||
)
|
||||
await user_repo.create(user)
|
||||
svc = PasswordService(hasher=PasswordHasher(time_cost=1, memory_cost=8192))
|
||||
cred_repo = app.state.credential_repo
|
||||
await cred_repo.create_password(
|
||||
PasswordCredential(user_id=user.userid, password_hash=svc.hash("testpass"))
|
||||
)
|
||||
|
||||
|
||||
async def _login_and_start_auth(client: AsyncClient, scope: str = "openid profile") -> None:
|
||||
await client.post(
|
||||
"/login/password",
|
||||
data={"username": "consentuser", "password": "testpass"},
|
||||
headers={"HX-Request": "true"},
|
||||
)
|
||||
await client.get(
|
||||
"/authorization",
|
||||
params={
|
||||
"response_type": "code",
|
||||
"client_id": "consent-rp",
|
||||
"redirect_uri": "http://localhost:9000/callback",
|
||||
"scope": scope,
|
||||
"state": "teststate",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
```
|
||||
|
||||
**Step 2: Run tests to verify they fail**
|
||||
|
||||
Run: `uv run python -m pytest tests/test_oidc/test_consent_flow.py -v`
|
||||
Expected: FAIL — consent_repo not on app.state, no `/consent` route.
|
||||
|
||||
**Step 3: Write the implementation**
|
||||
|
||||
**3a. Wire up consent_repo in `src/porchlight/app.py`**
|
||||
|
||||
Add after `app.state.magic_link_repo` (around line 38):
|
||||
|
||||
```python
|
||||
from porchlight.store.sqlite.repositories import SQLiteConsentRepository # add to imports
|
||||
|
||||
app.state.consent_repo = SQLiteConsentRepository(db)
|
||||
```
|
||||
|
||||
**3b. Add consent routes and modify authorization flow in `src/porchlight/oidc/endpoints.py`**
|
||||
|
||||
Add scope descriptions constant:
|
||||
|
||||
```python
|
||||
SCOPE_DESCRIPTIONS: dict[str, str] = {
|
||||
"openid": "Sign you in (required)",
|
||||
"profile": "Your name and profile information",
|
||||
"email": "Your email address",
|
||||
"phone": "Your phone number",
|
||||
}
|
||||
```
|
||||
|
||||
Modify `authorization()` (line 65-66) — when user is authenticated, instead
|
||||
of calling `_complete_authorization()` directly, call a new
|
||||
`_check_consent_or_complete()` helper:
|
||||
|
||||
```python
|
||||
if userid and username:
|
||||
return await _check_consent_or_complete(
|
||||
request, oidc_server, endpoint, parsed, userid, username, query_params
|
||||
)
|
||||
```
|
||||
|
||||
Similarly modify `authorization_complete()` (line 97) to call
|
||||
`_check_consent_or_complete()` instead of `_complete_authorization()`.
|
||||
|
||||
Add the consent check helper:
|
||||
|
||||
```python
|
||||
async def _check_consent_or_complete(
|
||||
request: Request,
|
||||
oidc_server: object,
|
||||
endpoint: object,
|
||||
parsed: object,
|
||||
userid: str,
|
||||
username: str,
|
||||
auth_params: dict,
|
||||
) -> Response:
|
||||
"""Check if consent is needed; if so redirect to /consent, otherwise complete."""
|
||||
settings = request.app.state.settings
|
||||
client_id = auth_params.get("client_id", "")
|
||||
|
||||
# Manage-app bypasses consent
|
||||
if client_id == settings.manage_client_id:
|
||||
return await _complete_authorization(request, oidc_server, endpoint, parsed, userid, username)
|
||||
|
||||
# Check stored consent
|
||||
consent_repo = request.app.state.consent_repo
|
||||
requested_scopes = auth_params.get("scope", "openid").split()
|
||||
stored_consent = await consent_repo.get_consent(userid, client_id)
|
||||
|
||||
if stored_consent and set(requested_scopes) <= set(stored_consent.scopes):
|
||||
# All requested scopes already approved
|
||||
return await _complete_authorization(request, oidc_server, endpoint, parsed, userid, username)
|
||||
|
||||
# Consent needed — store auth state and redirect
|
||||
request.session["consent_auth_request"] = auth_params
|
||||
return RedirectResponse("/consent", status_code=303)
|
||||
```
|
||||
|
||||
Add the consent page routes:
|
||||
|
||||
```python
|
||||
@router.get("/consent")
|
||||
async def consent_page(request: Request) -> Response:
|
||||
"""Show the consent form."""
|
||||
auth_params = request.session.get("consent_auth_request")
|
||||
if auth_params is None:
|
||||
return HTMLResponse("<h1>Error</h1><p>No pending consent request</p>", status_code=400)
|
||||
|
||||
userid = request.session.get("userid")
|
||||
if not userid:
|
||||
return RedirectResponse("/login", status_code=303)
|
||||
|
||||
client_id = auth_params.get("client_id", "")
|
||||
requested_scopes = auth_params.get("scope", "openid").split()
|
||||
|
||||
scope_info = [
|
||||
{"name": s, "description": SCOPE_DESCRIPTIONS.get(s, s), "required": s == "openid"}
|
||||
for s in requested_scopes
|
||||
]
|
||||
|
||||
templates = request.app.state.templates
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"consent.html",
|
||||
{"client_id": client_id, "scopes": scope_info},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/consent")
|
||||
async def consent_submit(request: Request) -> Response:
|
||||
"""Handle consent form submission."""
|
||||
auth_params = request.session.pop("consent_auth_request", None)
|
||||
if auth_params is None:
|
||||
return HTMLResponse("<h1>Error</h1><p>No pending consent request</p>", status_code=400)
|
||||
|
||||
userid = request.session.get("userid")
|
||||
username = request.session.get("username")
|
||||
if not userid or not username:
|
||||
return RedirectResponse("/login", status_code=303)
|
||||
|
||||
form = await request.form()
|
||||
action = form.get("action")
|
||||
client_id = auth_params.get("client_id", "")
|
||||
redirect_uri = auth_params.get("redirect_uri", "")
|
||||
state = auth_params.get("state", "")
|
||||
|
||||
if action == "deny":
|
||||
params = urlencode({"error": "access_denied", "state": state})
|
||||
return RedirectResponse(f"{redirect_uri}?{params}", status_code=303)
|
||||
|
||||
# Allow — collect approved scopes
|
||||
approved_scopes = form.getlist("scope")
|
||||
if "openid" not in approved_scopes:
|
||||
approved_scopes = ["openid"] + list(approved_scopes)
|
||||
|
||||
# Save consent
|
||||
consent_repo = request.app.state.consent_repo
|
||||
await consent_repo.set_consent(userid, client_id, list(approved_scopes))
|
||||
|
||||
# Filter auth request scopes to only approved
|
||||
auth_params["scope"] = " ".join(approved_scopes)
|
||||
|
||||
# Re-parse and complete
|
||||
oidc_server = request.app.state.oidc_server
|
||||
endpoint = oidc_server.get_endpoint("authorization")
|
||||
|
||||
try:
|
||||
parsed = endpoint.parse_request(auth_params)
|
||||
except Exception as exc:
|
||||
return HTMLResponse(f"<h1>Error</h1><p>{exc}</p>", status_code=400)
|
||||
|
||||
return await _complete_authorization(request, oidc_server, endpoint, parsed, userid, username)
|
||||
```
|
||||
|
||||
**Step 4: Run tests to verify they pass**
|
||||
|
||||
Run: `uv run python -m pytest tests/test_oidc/test_consent_flow.py -v`
|
||||
Expected: All PASS.
|
||||
|
||||
**Step 5: Run full test suite**
|
||||
|
||||
Run: `uv run python -m pytest -v`
|
||||
Expected: All tests PASS. Note: `test_full_authorization_code_flow` in
|
||||
`tests/test_oidc/test_e2e_flow.py` will need updating since it now hits the
|
||||
consent screen. Either pre-seed consent in the test setup, or add the consent
|
||||
step to the test flow.
|
||||
|
||||
**Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add src/porchlight/app.py src/porchlight/oidc/endpoints.py \
|
||||
tests/test_oidc/test_consent_flow.py
|
||||
git commit -m "feat: add consent check to authorization flow"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Add consent page template
|
||||
|
||||
**Files:**
|
||||
- Create: `src/porchlight/templates/consent.html`
|
||||
- Modify: `src/porchlight/static/style.css` (if needed)
|
||||
|
||||
**Step 1: Create the consent template**
|
||||
|
||||
Create `src/porchlight/templates/consent.html`:
|
||||
|
||||
```html
|
||||
{% extends "base.html" %}
|
||||
|
||||
{% block title %}Authorize — Porchlight{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<div class="card consent-card">
|
||||
<h1>Authorize {{ client_id }}</h1>
|
||||
<p>This application is requesting access to your account.</p>
|
||||
|
||||
<form method="post" action="/consent">
|
||||
<fieldset>
|
||||
<legend>Permissions requested</legend>
|
||||
<ul class="scope-list" role="list">
|
||||
{% for scope in scopes %}
|
||||
<li>
|
||||
<label>
|
||||
<input type="checkbox" name="scope" value="{{ scope.name }}"
|
||||
{% if scope.required %}checked disabled{% else %}checked{% endif %}>
|
||||
{{ scope.description }}
|
||||
</label>
|
||||
{% if scope.required %}
|
||||
<input type="hidden" name="scope" value="{{ scope.name }}">
|
||||
{% endif %}
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
</fieldset>
|
||||
|
||||
<div class="consent-actions">
|
||||
<button type="submit" name="action" value="allow" class="btn btn-primary">Allow</button>
|
||||
<button type="submit" name="action" value="deny" class="btn btn-secondary">Deny</button>
|
||||
</div>
|
||||
</form>
|
||||
</div>
|
||||
{% endblock %}
|
||||
```
|
||||
|
||||
Note: The `openid` checkbox is `checked disabled` so the user sees it but
|
||||
can't uncheck it. A hidden input ensures the value is still submitted.
|
||||
|
||||
**Step 2: Verify consent page renders**
|
||||
|
||||
Run: `uv run python -m pytest tests/test_oidc/test_consent_flow.py::test_consent_page_renders -v`
|
||||
Expected: PASS.
|
||||
|
||||
**Step 3: Commit**
|
||||
|
||||
```bash
|
||||
git add src/porchlight/templates/consent.html
|
||||
git commit -m "feat: add consent page template"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Update existing E2E test and add consent E2E test
|
||||
|
||||
**Files:**
|
||||
- Modify: `tests/test_oidc/test_e2e_flow.py`
|
||||
|
||||
The existing `test_full_authorization_code_flow` test directly calls
|
||||
`/authorization/complete` which now redirects to `/consent` instead of the RP
|
||||
callback. This test needs to handle the consent step.
|
||||
|
||||
**Step 1: Update the existing E2E flow test**
|
||||
|
||||
In `tests/test_oidc/test_e2e_flow.py`, after step 2 (login) and step 3
|
||||
(complete authorization), add the consent step. The test should:
|
||||
|
||||
1. Follow the redirect to `/consent` when `/authorization/complete`
|
||||
returns 303 to `/consent`
|
||||
2. POST to `/consent` with `action=allow` and the requested scopes
|
||||
3. Follow the redirect to the RP callback
|
||||
|
||||
Alternatively, pre-seed consent in the test setup using the consent repo:
|
||||
|
||||
```python
|
||||
consent_repo = app.state.consent_repo
|
||||
await consent_repo.set_consent(user.userid, client_id, ["openid", "profile", "email"])
|
||||
```
|
||||
|
||||
This approach is simpler and keeps the existing test focused on the OIDC
|
||||
token flow rather than the consent UI.
|
||||
|
||||
**Step 2: Run full test suite**
|
||||
|
||||
Run: `uv run python -m pytest -v`
|
||||
Expected: All PASS.
|
||||
|
||||
**Step 3: Commit**
|
||||
|
||||
```bash
|
||||
git add tests/test_oidc/test_e2e_flow.py
|
||||
git commit -m "test: update E2E flow test to handle consent"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Quality check
|
||||
|
||||
**Step 1: Run formatter and linter**
|
||||
|
||||
Run: `uv run ruff format src/ tests/ && uv run ruff check src/ tests/ --fix`
|
||||
|
||||
**Step 2: Run type checker**
|
||||
|
||||
Run: `uv run ty check src/`
|
||||
|
||||
**Step 3: Run full test suite**
|
||||
|
||||
Run: `uv run python -m pytest -v`
|
||||
|
||||
**Step 4: Fix any issues and commit**
|
||||
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "refactor: fix lint and type check issues"
|
||||
```
|
||||
Loading…
Add table
Add a link
Reference in a new issue