From 584488b9c79364a67848f0032c7cc04559696c4c Mon Sep 17 00:00:00 2001 From: jableader Date: Sat, 1 Nov 2025 13:51:08 +1100 Subject: [PATCH] v2 register/login endpoints exist with bearer scaffold --- api/deps.py | 34 ++++++++++++++++++++++- api/households.py | 16 ++++++++++- api/openapi.py | 33 ++++++++++++++++++++++- main.py | 5 ++++ tests/test_household_scoping.py | 48 +++++++++++++++++++++++++++++++++ tests/test_openapi_security.py | 25 +++++++++++++++++ 6 files changed, 158 insertions(+), 3 deletions(-) create mode 100644 tests/test_household_scoping.py create mode 100644 tests/test_openapi_security.py diff --git a/api/deps.py b/api/deps.py index 0f54624..74c7156 100644 --- a/api/deps.py +++ b/api/deps.py @@ -10,8 +10,13 @@ import db import persons from common import ProblemDetails from settings import settings -from users import repository as users_db from users.models import User +from typing import TypedDict + + +class HouseholdCtx(TypedDict): + id: int + slug: str # Dependency to create SQLite connection with PRAGMAs and per-request transaction @@ -105,3 +110,30 @@ async def get_current_user(request: Request, conn: aiosqlite.Connection = Depend if not row: raise HTTPException(status_code=401, detail="Unauthorized") return User(id=int(row[0]), email=row[1], display_name=row[2], profile_photo_url=row[3]) + + +async def get_household_from_slug( + request: Request, + householdSlug: str, # path parameter + user: User = Depends(get_current_user), + conn: aiosqlite.Connection = Depends(get_db), +) -> HouseholdCtx: + # Find household by slug + async with conn.execute( + "SELECT id, slug FROM Household WHERE slug = ? LIMIT 1", + (householdSlug,), + ) as c: + row = await c.fetchone() + if not row: + # 404 to avoid leaking membership existence + raise HTTPException(status_code=404, detail="Household not found") + hid = int(row[0]) + # Verify membership + async with conn.execute( + "SELECT 1 FROM HouseholdMember WHERE user_id = ? AND household_id = ? LIMIT 1", + (user.id, hid), + ) as c: + m = await c.fetchone() + if not m: + raise HTTPException(status_code=403, detail="Forbidden") + return {"id": hid, "slug": row[1]} diff --git a/api/households.py b/api/households.py index 4e324ea..0a5280c 100644 --- a/api/households.py +++ b/api/households.py @@ -6,7 +6,7 @@ from typing import List import aiosqlite from fastapi import APIRouter, Depends, Request -from api.deps import get_current_user, get_db, error_response +from api.deps import get_current_user, get_db, error_response, get_household_from_slug from common import ApiModel from users.models import User @@ -70,3 +70,17 @@ async def create_household( return HouseholdResponse(id=hid, name=body.name, slug=slug) except Exception: return error_response(request, 400, "Unable to create household") + + +# Household-scoped router and endpoint to validate scoping mechanics +scoped = APIRouter(prefix="/households/{householdSlug}") + + +class WhoAmI(ApiModel): + household_id: int + household_slug: str + + +@scoped.get("/whoami", response_model=WhoAmI) +async def whoami(household=Depends(get_household_from_slug)): + return WhoAmI(household_id=household["id"], household_slug=household["slug"]) diff --git a/api/openapi.py b/api/openapi.py index 0de951b..cce7629 100644 --- a/api/openapi.py +++ b/api/openapi.py @@ -53,7 +53,7 @@ def extend_with_problem_and_cookie_auth(app: FastAPI) -> None: }, ) - # Cookie-based auth for documentation (does not enforce at runtime) + # Cookie-based auth for v1 documentation (does not enforce at runtime) security_schemes.setdefault( "cookieAuth", { @@ -64,6 +64,17 @@ def extend_with_problem_and_cookie_auth(app: FastAPI) -> None: }, ) + # Bearer (JWT) auth for v2 + security_schemes.setdefault( + "bearerAuth", + { + "type": "http", + "scheme": "bearer", + "bearerFormat": "JWT", + "description": "JWT access token in Authorization header", + }, + ) + # Normalize v1 responses and mark cookie security for known endpoints paths = spec.get("paths", {}) protected_ops: set[str] = { @@ -140,6 +151,26 @@ def extend_with_problem_and_cookie_auth(app: FastAPI) -> None: if isinstance(store, dict) and store.get("$ref") == "#/components/schemas/StoreEnum": props["storeName"] = {"$ref": "#/components/schemas/StoreNameOut"} + # Mark bearer security for v2 routes we know require auth + # Simple heuristic: underline select paths under /api/v1/users/me and /api/v1/households/* that are protected + for path, ops in paths.items(): + if not isinstance(path, str) or not path.startswith("/api/v1/"): + continue + if not isinstance(ops, dict): + continue + needs_bearer = ( + path.startswith("/api/v1/users/me/") + or path.startswith("/api/v1/households/") + ) + if not needs_bearer: + continue + for _method, op in ops.items(): + if not isinstance(op, dict): + continue + security = op.setdefault("security", []) + if not any(isinstance(s, dict) and "bearerAuth" in s for s in security): + security.append({"bearerAuth": []}) + return spec # Rebind app.openapi to our generator (FastAPI supports this pattern). Mypy needs a narrow ignore here. diff --git a/main.py b/main.py index 99726bd..6494843 100644 --- a/main.py +++ b/main.py @@ -164,6 +164,11 @@ def create_app() -> FastAPI: # Experimental v2 auth endpoints (JWT to be implemented). Kept alongside v1 during transition. app.include_router(auth_v2_router.router, prefix="/api/v1", tags=["v2"]) app.include_router(households_router.router, prefix="/api/v1", tags=["v2"]) # new + # Mount household-scoped endpoints + try: + app.include_router(households_router.scoped, prefix="/api/v1", tags=["v2"]) # type: ignore[attr-defined] + except Exception: + pass # Routes app.add_api_route("/healthz", healthz, methods=["GET"], response_model=HealthStatus) diff --git a/tests/test_household_scoping.py b/tests/test_household_scoping.py new file mode 100644 index 0000000..3133ee5 --- /dev/null +++ b/tests/test_household_scoping.py @@ -0,0 +1,48 @@ +import unittest +from fastapi.testclient import TestClient + +import main +from db import connect, create + + +class TestHouseholdScoping(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.conn = await connect(":memory:") + await create(self.conn) + + async def override_get_db(): + try: + yield self.conn + finally: + pass + + main.app.dependency_overrides[main.get_db] = override_get_db + self.client = TestClient(main.app) + + # Register a user and capture token + r = self.client.post( + "/api/v1/auth/register", + json={"email": "h@test.com", "password": "pw", "displayName": "H"}, + ) + assert r.status_code == 200, r.text + self.token = r.json()["accessToken"] + self.headers = {"Authorization": f"Bearer {self.token}"} + + async def asyncTearDown(self): + await self.conn.close() + main.app.dependency_overrides.clear() + + def test_scoped_whoami_forbidden_on_default_when_not_member(self): + r = self.client.get("/api/v1/households/default/whoami", headers=self.headers) + assert r.status_code in (403, 404) # may be 404 if default household missing + + def test_scoped_whoami_ok_after_creating_household(self): + r = self.client.post( + "/api/v1/households", json={"name": "Family"}, headers=self.headers + ) + assert r.status_code == 200, r.text + slug = r.json()["slug"] + r = self.client.get(f"/api/v1/households/{slug}/whoami", headers=self.headers) + assert r.status_code == 200, r.text + body = r.json() + assert body["householdSlug"] == slug diff --git a/tests/test_openapi_security.py b/tests/test_openapi_security.py new file mode 100644 index 0000000..05a207e --- /dev/null +++ b/tests/test_openapi_security.py @@ -0,0 +1,25 @@ +import unittest + +import main + + +class TestOpenAPISecurity(unittest.TestCase): + def test_bearer_auth_included(self): + spec = main.app.openapi() + comps = spec.get("components", {}) + sec = comps.get("securitySchemes", {}) + assert "bearerAuth" in sec + bearer = sec["bearerAuth"] + assert bearer.get("type") == "http" + assert bearer.get("scheme") == "bearer" + assert bearer.get("bearerFormat") == "JWT" + + def test_household_routes_require_bearer(self): + spec = main.app.openapi() + paths = spec.get("paths", {}) + # Users me households + op = paths.get("/api/v1/users/me/households", {}).get("get") + assert op and any("bearerAuth" in s for s in op.get("security", [])) + # Household whoami + op = paths.get("/api/v1/households/{householdSlug}/whoami", {}).get("get") + assert op and any("bearerAuth" in s for s in op.get("security", []))