851 lines
27 KiB
Markdown
851 lines
27 KiB
Markdown
# 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"
|
|
```
|