From 5e24e593e38f142519c8fc0aac67d58aad3e7b19 Mon Sep 17 00:00:00 2001 From: Krish Date: Sat, 22 Aug 2026 22:18:56 +0530 Subject: [PATCH] fix: enforce quiz ownership on submit and grade Authentication established who the caller was but never checked that the quiz was theirs. Any logged-in user who knew or guessed a quiz_id could answer and grade somebody else's attempt, which made the attribution the token establishes worthless - a comprehension score is a claim about a specific person, and this let one person's answers land on another person's record. Ownership is now a precondition of loading the attempt rather than a check each caller performs. _load_owned_attempt refuses anything that is missing or belongs to someone else, and start_followup and grade_quiz both go through it. user_id is a required argument on both, not an optional one. When the signatures changed, five existing tests failed with TypeError rather than quietly passing, which is the point: an endpoint that forgets to pass the caller cannot compile by accident into a bypass. Three decisions worth recording. A quiz belonging to somebody else answers 404, not 403. A 403 would confirm the quiz exists and turn the endpoint into an oracle for discovering valid ids. Tests assert the missing-quiz and foreign-quiz responses are byte-identical, the same reasoning already applied to login not revealing whether an account exists. Refusal happens before any model call. Two tests assert generate_followup_question and grade_answers are never reached on a rejected request, so an attacker cannot burn the daily Gemini quota on quizzes they have no access to. Attempts with no owner are unreachable by anyone. Any attempt created before authentication existed has user_id None and now 404s for every caller. That is correct rather than unfortunate: an unattributed attempt proves nothing about anyone. Verified live with two registered accounts against Atlas: A generates a quiz, B is refused 404 on both submit and followup, B gets the same 404 for an id that does not exist at all, and A still submits their own quiz successfully. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PET9qKZXhgjEbZK7MReYQj --- backend/app/api/v1/endpoints/quiz.py | 10 ++++- backend/app/models/quiz.py | 5 ++- backend/app/services/quiz_service.py | 33 ++++++++++---- backend/tests/test_auth_api.py | 66 ++++++++++++++++++++++++++++ backend/tests/test_quiz_service.py | 65 ++++++++++++++++++++++++--- 5 files changed, 163 insertions(+), 16 deletions(-) diff --git a/backend/app/api/v1/endpoints/quiz.py b/backend/app/api/v1/endpoints/quiz.py index 0f63cd6..d86e1c0 100644 --- a/backend/app/api/v1/endpoints/quiz.py +++ b/backend/app/api/v1/endpoints/quiz.py @@ -40,9 +40,11 @@ async def submit(req: QuizSubmitRequest, user: dict = Depends(get_current_user)) """Records answers and returns the adaptive follow-up. Does not grade.""" try: return await quiz_service.start_followup( - req.quiz_id, [a.model_dump() for a in req.answers] + req.quiz_id, [a.model_dump() for a in req.answers], user["user_id"] ) except LookupError: + # Also the answer when the quiz exists but belongs to somebody else, so + # this cannot be used to discover which quiz ids are real. raise HTTPException(404, "Quiz not found") except ValueError: raise HTTPException(400, "No answers submitted") @@ -52,8 +54,12 @@ async def submit(req: QuizSubmitRequest, user: dict = Depends(get_current_user)) async def followup(req: FollowUpRequest, user: dict = Depends(get_current_user)): """Grades the original answers together with the follow-up defence.""" try: - result = await quiz_service.grade_quiz(req.quiz_id, req.answer, req.seconds_left) + result = await quiz_service.grade_quiz( + req.quiz_id, req.answer, user["user_id"], req.seconds_left + ) except LookupError: + # Also the answer when the quiz exists but belongs to somebody else, so + # this cannot be used to discover which quiz ids are real. raise HTTPException(404, "Quiz not found") return { diff --git a/backend/app/models/quiz.py b/backend/app/models/quiz.py index 1e17bfa..b6b4b3e 100644 --- a/backend/app/models/quiz.py +++ b/backend/app/models/quiz.py @@ -7,7 +7,10 @@ { "_id": str (uuid4), "repo_url": str, - "user_id": str | None, + "user_id": str, # owner; set from the access token, never from the request body. + # Every read after creation goes through + # quiz_service._load_owned_attempt, which refuses an attempt + # belonging to anyone else. "questions": [ {"id": str, "question": str, "file_reference": str | None, "category": "problem" | "logic" | "stack" | "usage" | None} diff --git a/backend/app/services/quiz_service.py b/backend/app/services/quiz_service.py index cbd67b7..86dd30d 100644 --- a/backend/app/services/quiz_service.py +++ b/backend/app/services/quiz_service.py @@ -76,16 +76,33 @@ def suspicion(a: dict) -> float: return max(timed, key=suspicion) -async def start_followup(quiz_id: str, answers: list[dict]) -> dict: +async def _load_owned_attempt(quiz_id: str, user_id: str) -> dict: + """ + Fetch an attempt, but only for the person it belongs to. + + Ownership is enforced here rather than in each caller so it cannot be + forgotten by one of them. A quiz that exists but belongs to someone else + raises the same LookupError as one that does not exist, so the endpoint + answers 404 either way and cannot be used to discover which quiz ids are real. + + Attempts created before authentication existed have no owner and are therefore + unreachable, which is correct: an unattributed attempt proves nothing about + anyone. + """ + attempt = await quiz_repository.get_attempt(quiz_id) + if not attempt or attempt.get("user_id") != user_id: + raise LookupError("quiz_not_found") + return attempt + + +async def start_followup(quiz_id: str, answers: list[dict], user_id: str) -> dict: """ Record the answers and open the follow-up round. Deliberately does not grade yet — grading before the follow-up would let a candidate bank a score and abandon the round they cannot pass. """ - attempt = await quiz_repository.get_attempt(quiz_id) - if not attempt: - raise LookupError("quiz_not_found") + attempt = await _load_owned_attempt(quiz_id, user_id) suspect = pick_suspect_answer(answers) if suspect is None: @@ -114,11 +131,11 @@ async def start_followup(quiz_id: str, answers: list[dict]) -> dict: } -async def grade_quiz(quiz_id: str, followup_answer: str, seconds_left: float | None = None) -> dict: +async def grade_quiz( + quiz_id: str, followup_answer: str, user_id: str, seconds_left: float | None = None +) -> dict: """Final grading — original answers plus the follow-up defence.""" - attempt = await quiz_repository.get_attempt(quiz_id) - if not attempt: - raise LookupError("quiz_not_found") + attempt = await _load_owned_attempt(quiz_id, user_id) followup = dict(attempt.get("followup") or {}) followup["answer"] = followup_answer diff --git a/backend/tests/test_auth_api.py b/backend/tests/test_auth_api.py index a88394f..90e4617 100644 --- a/backend/tests/test_auth_api.py +++ b/backend/tests/test_auth_api.py @@ -155,3 +155,69 @@ def test_generate_attributes_the_quiz_to_the_token_not_the_body(monkeypatch, que assert resp.status_code == 200 assert spy.call_args.args[1] == "real-user" + + +# --- ownership at the HTTP boundary --------------------------------------- + + +def test_another_user_cannot_submit_against_your_quiz(monkeypatch): + """ + Being logged in is not enough. Before this, any authenticated caller who knew + or guessed a quiz_id could answer somebody else's quiz, which made the + attribution the token establishes meaningless. + """ + attempt = {"_id": "quiz-1", "user_id": "owner", "questions": [{"id": "a", "question": "q"}]} + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + gen = AsyncMock(return_value="follow-up?") + monkeypatch.setattr(quiz_service.gemini_client, "generate_followup_question", gen) + monkeypatch.setattr(quiz_service.quiz_repository, "update_followup", AsyncMock()) + + body = {"quiz_id": "quiz-1", "answers": [{"question_id": "a", "answer": "b"}]} + intruder = client.post("/api/v1/quiz/submit", json=body, headers=auth_header("intruder")) + + assert intruder.status_code == 404 + gen.assert_not_called(), "must refuse before spending an API call" + + +def test_another_user_cannot_grade_your_quiz(monkeypatch): + attempt = {"_id": "quiz-1", "user_id": "owner", "questions": [], "answers": [], "followup": {}} + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + grade = AsyncMock() + monkeypatch.setattr(quiz_service.gemini_client, "grade_answers", grade) + + resp = client.post("/api/v1/quiz/followup", + json={"quiz_id": "quiz-1", "answer": "defence"}, + headers=auth_header("intruder")) + + assert resp.status_code == 404 + grade.assert_not_called() + + +def test_a_foreign_quiz_looks_exactly_like_a_missing_one(monkeypatch): + """Otherwise the 404/403 split would leak which quiz ids exist.""" + body = {"quiz_id": "quiz-1", "answers": [{"question_id": "a", "answer": "b"}]} + + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=None)) + missing = client.post("/api/v1/quiz/submit", json=body, headers=auth_header("intruder")) + + attempt = {"_id": "quiz-1", "user_id": "owner", "questions": [{"id": "a", "question": "q"}]} + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + foreign = client.post("/api/v1/quiz/submit", json=body, headers=auth_header("intruder")) + + assert missing.status_code == foreign.status_code == 404 + assert missing.json() == foreign.json() + + +def test_the_owner_can_still_submit(monkeypatch): + attempt = {"_id": "quiz-1", "user_id": "owner", "questions": [{"id": "a", "question": "q"}]} + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + monkeypatch.setattr(quiz_service.gemini_client, "generate_followup_question", + AsyncMock(return_value="follow-up?")) + monkeypatch.setattr(quiz_service.quiz_repository, "update_followup", AsyncMock()) + + resp = client.post("/api/v1/quiz/submit", + json={"quiz_id": "quiz-1", "answers": [{"question_id": "a", "answer": "b"}]}, + headers=auth_header("owner")) + + assert resp.status_code == 200 + assert resp.json()["followup"]["targets_question_id"] == "a" diff --git a/backend/tests/test_quiz_service.py b/backend/tests/test_quiz_service.py index fab3eef..4d70773 100644 --- a/backend/tests/test_quiz_service.py +++ b/backend/tests/test_quiz_service.py @@ -112,7 +112,7 @@ async def test_start_followup_targets_the_flagged_answer(monkeypatch, attempt): monkeypatch.setattr(quiz_service.gemini_client, "generate_followup_question", gen) monkeypatch.setattr(quiz_service.quiz_repository, "update_followup", AsyncMock()) - out = await quiz_service.start_followup("quiz-1", attempt["answers"]) + out = await quiz_service.start_followup("quiz-1", attempt["answers"], "u1") assert out["followup"]["targets_question_id"] == "q2" # the pasted answer's text is what the model was asked to push on @@ -127,7 +127,7 @@ async def test_start_followup_does_not_grade(monkeypatch, attempt): graded = AsyncMock() monkeypatch.setattr(quiz_service.gemini_client, "grade_answers", graded) - out = await quiz_service.start_followup("quiz-1", attempt["answers"]) + out = await quiz_service.start_followup("quiz-1", attempt["answers"], "u1") graded.assert_not_called() assert "score" not in out @@ -136,7 +136,7 @@ async def test_start_followup_does_not_grade(monkeypatch, attempt): async def test_start_followup_unknown_quiz(monkeypatch): monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=None)) with pytest.raises(LookupError): - await quiz_service.start_followup("nope", [answer("q1")]) + await quiz_service.start_followup("nope", [answer("q1")], "u1") # --- final grading --------------------------------------------------------- @@ -147,7 +147,7 @@ async def test_grade_quiz_passes_the_followup_defence_to_the_grader(monkeypatch, monkeypatch.setattr(quiz_service.gemini_client, "grade_answers", grade) monkeypatch.setattr(quiz_service.quiz_repository, "update_result", AsyncMock()) - result = await quiz_service.grade_quiz("quiz-1", "my defence", seconds_left=9.0) + result = await quiz_service.grade_quiz("quiz-1", "my defence", "u1", seconds_left=9.0) assert result["overall_score"] == 88 sent = grade.call_args.kwargs["followup"] @@ -158,4 +158,59 @@ async def test_grade_quiz_passes_the_followup_defence_to_the_grader(monkeypatch, async def test_grade_quiz_unknown_quiz(monkeypatch): monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=None)) with pytest.raises(LookupError): - await quiz_service.grade_quiz("nope", "answer") + await quiz_service.grade_quiz("nope", "answer", "u1") + + +# --- ownership ------------------------------------------------------------ + + +async def test_start_followup_refuses_another_users_quiz(monkeypatch, attempt): + """The quiz exists and the caller is authenticated — but it is not theirs.""" + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + gen = AsyncMock() + monkeypatch.setattr(quiz_service.gemini_client, "generate_followup_question", gen) + + with pytest.raises(LookupError): + await quiz_service.start_followup("quiz-1", attempt["answers"], "someone-else") + gen.assert_not_called(), "must refuse before spending an API call" + + +async def test_grade_quiz_refuses_another_users_quiz(monkeypatch, attempt): + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + grade = AsyncMock() + monkeypatch.setattr(quiz_service.gemini_client, "grade_answers", grade) + + with pytest.raises(LookupError): + await quiz_service.grade_quiz("quiz-1", "defence", "someone-else") + grade.assert_not_called() + + +async def test_a_missing_quiz_and_someone_elses_are_indistinguishable(monkeypatch, attempt): + """Both raise LookupError, so the endpoint answers 404 either way.""" + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=None)) + with pytest.raises(LookupError) as missing: + await quiz_service.grade_quiz("nope", "d", "u1") + + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + with pytest.raises(LookupError) as foreign: + await quiz_service.grade_quiz("quiz-1", "d", "someone-else") + + assert str(missing.value) == str(foreign.value) + + +async def test_unattributed_attempts_are_unreachable(monkeypatch, attempt): + """Attempts predating auth have no owner and prove nothing about anyone.""" + attempt["user_id"] = None + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + with pytest.raises(LookupError): + await quiz_service.grade_quiz("quiz-1", "d", "u1") + + +async def test_owner_still_gets_through(monkeypatch, attempt): + monkeypatch.setattr(quiz_service.quiz_repository, "get_attempt", AsyncMock(return_value=attempt)) + monkeypatch.setattr(quiz_service.gemini_client, "grade_answers", + AsyncMock(return_value={"overall_score": 70, "breakdown": []})) + monkeypatch.setattr(quiz_service.quiz_repository, "update_result", AsyncMock()) + + result = await quiz_service.grade_quiz("quiz-1", "defence", "u1") + assert result["overall_score"] == 70