diff --git a/routers/exam/templates.py b/routers/exam/templates.py index 5b19af6..48ad039 100644 --- a/routers/exam/templates.py +++ b/routers/exam/templates.py @@ -137,6 +137,22 @@ def _lookup_exam_storage_loc(exam_id: str) -> Optional[str]: return None +def _signed_url_value(result: Any) -> str: + """Normalise supabase-py signed URL responses across v1/v2 shapes.""" + if isinstance(result, str): + return result + if isinstance(result, dict): + value = result.get("signedURL") or result.get("signedUrl") or result.get("signed_url") + if value: + return str(value) + data = getattr(result, "data", None) + if isinstance(data, dict): + value = data.get("signedURL") or data.get("signedUrl") or data.get("signed_url") + if value: + return str(value) + raise ValueError("Storage service did not return a signed URL") + + async def _parse_create_template_request(request: Request) -> tuple[CreateTemplateRequest, Optional[UploadFile]]: content_type = request.headers.get("content-type", "") if "multipart/form-data" in content_type: @@ -608,12 +624,13 @@ async def create_template( @router.get("/catalogue") -async def list_catalogue_papers() -> Dict[str, Any]: - """Lightweight exam-board paper catalogue for the create dialog.""" +async def list_catalogue_papers( + ctx: ExamContext = Depends(get_exam_context), +) -> Dict[str, Any]: + """Lightweight authenticated exam-board metadata catalogue for the create dialog.""" try: - sb = SupabaseServiceRoleClient().supabase res = ( - sb.table("eb_exams") + ctx.supabase.table("eb_exams") .select("id, exam_code, spec_code, paper_code, tier, session, type_code, storage_loc") .eq("type_code", "QP") .order("exam_code") @@ -624,6 +641,50 @@ async def list_catalogue_papers() -> Dict[str, Any]: raise HTTPException(status_code=502, detail=f"Could not load catalogue papers: {exc}") +@router.get("/catalogue/{exam_id}/signed-url") +async def get_catalogue_paper_signed_url( + exam_id: str, + expires_in: int = 300, + ctx: ExamContext = Depends(get_exam_context), +) -> Dict[str, Any]: + """Return a short-lived signed URL for an authenticated user's catalogue PDF access. + + The storage operation uses service role as a scoped backend exception for signing only; + raw cc.examboards object reads remain denied by storage.objects RLS. + """ + expires_in = max(60, min(int(expires_in or 300), 3600)) + try: + row = _first( + ctx.supabase.table("eb_exams") + .select("id, exam_code, storage_loc") + .eq("id", exam_id) + .eq("type_code", "QP") + .limit(1) + .execute() + ) + if not row or not row.get("storage_loc"): + raise HTTPException(status_code=404, detail="Catalogue paper not found") + try: + bucket, path = _parse_storage_loc(row["storage_loc"]) + except ValueError: + raise HTTPException(status_code=404, detail="Catalogue paper not found") + if bucket != "cc.examboards": + raise HTTPException(status_code=404, detail="Catalogue paper not found") + signed_url = _signed_url_value(StorageAdmin().create_signed_url(bucket, path, expires_in)) + return { + "exam_id": row["id"], + "exam_code": row.get("exam_code"), + "bucket": bucket, + "path": path, + "expires_in": expires_in, + "signed_url": signed_url, + } + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=502, detail=f"Could not sign catalogue paper URL: {exc}") + + @router.get("/templates") async def list_templates( include_archived: bool = False, diff --git a/tests/test_exam_templates.py b/tests/test_exam_templates.py index 598e7b6..6ac1676 100644 --- a/tests/test_exam_templates.py +++ b/tests/test_exam_templates.py @@ -143,6 +143,9 @@ class _FakeStorageAdmin: def download_file(self, bucket_id, file_path): return b"%PDF-1.7 fake" + def create_signed_url(self, bucket_id, file_path, expires_in=3600): + return {"signedURL": f"https://storage.test/{bucket_id}/{file_path}?token=fake&expires_in={expires_in}"} + class _FakeServiceRoleClient: def __init__(self, store): @@ -171,6 +174,65 @@ def test_requires_auth_when_not_overridden(): assert resp.status_code in (401, 403) # unauthenticated, not processed +def test_catalogue_requires_auth_when_not_overridden(): + app = FastAPI() + app.include_router(router, prefix="/api/exam") + resp = TestClient(app).get("/api/exam/catalogue") + assert resp.status_code in (401, 403) + + +def test_list_catalogue_papers_uses_as_user_metadata(): + store = { + "eb_exams": [ + {"id": "e1", "exam_code": "AQA-1", "type_code": "QP", "storage_loc": "cc.examboards/aqa/p.pdf"}, + {"id": "e2", "exam_code": "AQA-MS", "type_code": "MS", "storage_loc": "cc.examboards/aqa/ms.pdf"}, + ] + } + client, _ = make_client(store=store) + resp = client.get("/api/exam/catalogue") + assert resp.status_code == 200 + assert [p["id"] for p in resp.json()["papers"]] == ["e1"] + + +def test_catalogue_signed_url_requires_auth_and_signs_examboard_pdf(monkeypatch): + store = { + "eb_exams": [ + {"id": "e1", "exam_code": "AQA-1", "type_code": "QP", "storage_loc": "cc.examboards/aqa/physics/qp.pdf"}, + ] + } + client, _ = make_client(store=store) + monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin) + resp = client.get("/api/exam/catalogue/e1/signed-url?expires_in=120") + assert resp.status_code == 200 + body = resp.json() + assert body["bucket"] == "cc.examboards" + assert body["path"] == "aqa/physics/qp.pdf" + assert body["expires_in"] == 120 + assert "token=fake" in body["signed_url"] + + +def test_catalogue_signed_url_rejects_non_examboard_storage(monkeypatch): + store = { + "eb_exams": [ + {"id": "e1", "exam_code": "AQA-1", "type_code": "QP", "storage_loc": "cc.public/aqa/physics/qp.pdf"}, + ] + } + client, _ = make_client(store=store) + monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin) + assert client.get("/api/exam/catalogue/e1/signed-url").status_code == 404 + + +def test_catalogue_signed_url_rejects_non_catalogue_doc_type(monkeypatch): + store = { + "eb_exams": [ + {"id": "e1", "exam_code": "AQA-MS", "type_code": "MS", "storage_loc": "cc.examboards/aqa/physics/ms.pdf"}, + ] + } + client, _ = make_client(store=store) + monkeypatch.setattr(templates_mod, "StorageAdmin", _FakeStorageAdmin) + assert client.get("/api/exam/catalogue/e1/signed-url").status_code == 404 + + def test_create_template_sets_owner_and_institute(): client, store = make_client() resp = client.post("/api/exam/templates", json={"title": "AQA Physics 1H", "subject": "Physics"})