feat: add JWT authentication and gate the quiz flow

backend/.env.example

@@ -2,3 +2,4 @@ MONGO_URI=your_mongodb_atlas_uri
 GEMINI_API_KEY=your_gemini_api_key
 GEMINI_MODEL=gemini-3.6-flash
 GITHUB_TOKEN=
+JWT_SECRET=change_me_to_a_long_random_string

backend/app/api/v1/endpoints/auth.py

@@ -0,0 +1,29 @@
+"""
+Thin HTTP layer for authentication. No business logic here — only
+request/response translation and HTTP error mapping. Logic lives in
+app/services/auth_service.py.
+"""
+from fastapi import APIRouter, HTTPException
+
+from app.schemas.auth import LoginRequest, RegisterRequest, TokenResponse
+from app.services import auth_service
+
+router = APIRouter()
+
+
+@router.post("/register", response_model=TokenResponse, status_code=201)
+async def register(req: RegisterRequest):
+    try:
+        return await auth_service.register_user(req.email, req.password, req.role)
+    except ValueError:
+        # 409 rather than 400: the request was well-formed, the address is taken.
+        raise HTTPException(409, "That email is already registered.")
+
+
+@router.post("/login", response_model=TokenResponse)
+async def login(req: LoginRequest):
+    try:
+        return await auth_service.authenticate_user(req.email, req.password)
+    except LookupError:
+        # Deliberately identical for unknown email and wrong password.
+        raise HTTPException(401, "Incorrect email or password.")

backend/app/api/v1/endpoints/quiz.py

@@ -3,11 +3,16 @@
 request/response translation and HTTP error mapping. Logic lives in
 app/services/quiz_service.py.
 
+Every route here requires an access token. The quiz is a record of what a
+specific person understood, so an unattributed attempt is not meaningful.
+
 Flow: generate -> submit (opens follow-up) -> followup (final grade).
 Grading deliberately happens only after the follow-up, so a candidate
 cannot bank a score and walk away from the round they cannot pass.
 """
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
+
+from app.core.dependencies import get_current_user
 
 from app.schemas.quiz import (
     FollowUpRequest,
@@ -23,15 +28,15 @@
 
 
 @router.post("/generate", response_model=QuizGenerateResponse)
-async def generate(req: QuizGenerateRequest):
+async def generate(req: QuizGenerateRequest, user: dict = Depends(get_current_user)):
     try:
-        return await quiz_service.create_quiz(req.repo_url, req.user_id)
+        return await quiz_service.create_quiz(req.repo_url, user["user_id"])
     except ValueError:
         raise HTTPException(400, "Couldn't read source files from that repo — make sure it's public.")
 
 
 @router.post("/submit", response_model=QuizSubmitResponse)
-async def submit(req: QuizSubmitRequest):
+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(
@@ -44,7 +49,7 @@ async def submit(req: QuizSubmitRequest):
 
 
 @router.post("/followup", response_model=QuizResultResponse)
-async def followup(req: FollowUpRequest):
+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)

backend/app/api/v1/router.py

@@ -1,8 +1,9 @@
 """Aggregates every v1 route. main.py only ever imports this one router."""
 from fastapi import APIRouter
 
-from app.api.v1.endpoints import jobs, quiz
+from app.api.v1.endpoints import auth, jobs, quiz
 
 api_router = APIRouter()
+api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
 api_router.include_router(quiz.router, prefix="/quiz", tags=["quiz"])
 api_router.include_router(jobs.router, prefix="/jobs", tags=["jobs"])

backend/app/core/config.py

@@ -11,6 +11,11 @@ class Settings(BaseSettings):
     gemini_api_key: str = ""
     gemini_model: str = "gemini-3.6-flash"  # verify current model name in Google AI Studio
     github_token: str = ""  # optional, raises GitHub API rate limits from 60/hr to 5000/hr
+    # Signing key for access tokens. The default is a placeholder so the app still
+    # boots in dev and CI; anything issuing real tokens must override it.
+    jwt_secret: str = "dev-only-insecure-secret-change-me"
+    jwt_algorithm: str = "HS256"
+    jwt_expire_minutes: int = 60 * 12
 
     class Config:
         env_file = ".env"

backend/app/core/dependencies.py

@@ -0,0 +1,31 @@
+"""
+FastAPI dependencies shared across endpoints.
+
+Kept out of security.py so that module stays importable by tests and scripts
+without dragging in FastAPI request handling.
+"""
+from fastapi import Depends, HTTPException
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from app.core.security import decode_access_token
+
+# auto_error=False so a missing header reaches us as None and we can answer with a
+# consistent 401 rather than letting the shape of the failure vary.
+_bearer = HTTPBearer(auto_error=False)
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
+) -> dict:
+    """Resolves the caller from the Authorization header, or 401s."""
+    if credentials is None or not credentials.credentials:
+        raise HTTPException(401, "Not authenticated")
+    try:
+        claims = decode_access_token(credentials.credentials)
+    except ValueError:
+        raise HTTPException(401, "Invalid or expired token")
+
+    user_id = claims.get("sub")
+    if not user_id:
+        raise HTTPException(401, "Invalid or expired token")
+    return {"user_id": user_id, "role": claims.get("role")}

backend/app/core/security.py

@@ -0,0 +1,70 @@
+"""
+Password hashing and access tokens.
+
+Nothing else in the app should import passlib or jwt directly — go through here so
+the algorithm, expiry, and claim shape live in one place.
+"""
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+import jwt
+from passlib.context import CryptContext
+
+from app.core.config import settings
+
+# bcrypt is pinned below 5.x in requirements: passlib 1.7.4's bcrypt backend fails
+# its own self-test against bcrypt 5, which breaks hashing entirely rather than
+# just warning.
+_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# passlib probes bcrypt.__about__, which 4.1+ removed. It catches the failure and
+# carries on, but logs the traceback on every import. Silence the cosmetic noise
+# without hiding anything that actually matters.
+logging.getLogger("passlib.handlers.bcrypt").setLevel(logging.CRITICAL)
+
+# bcrypt silently uses only the first 72 bytes. Rejecting longer input is better
+# than accepting a password whose tail never mattered.
+MAX_PASSWORD_BYTES = 72
+
+
+def hash_password(password: str) -> str:
+    if len(password.encode("utf-8")) > MAX_PASSWORD_BYTES:
+        raise ValueError("password_too_long")
+    return _pwd.hash(password)
+
+
+def verify_password(password: str, hashed: str) -> bool:
+    if len(password.encode("utf-8")) > MAX_PASSWORD_BYTES:
+        return False
+    try:
+        return _pwd.verify(password, hashed)
+    except ValueError:
+        # Malformed or truncated hash in storage — treat as a failed login rather
+        # than a 500, so a bad row cannot be used to probe the endpoint.
+        return False
+
+
+def create_access_token(user_id: str, role: str, expires_minutes: Optional[int] = None) -> str:
+    now = datetime.now(timezone.utc)
+    minutes = settings.jwt_expire_minutes if expires_minutes is None else expires_minutes
+    payload = {
+        "sub": user_id,
+        "role": role,
+        "iat": now,
+        "exp": now + timedelta(minutes=minutes),
+    }
+    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
+
+
+def decode_access_token(token: str) -> dict:
+    """
+    Returns the claims, or raises ValueError for anything untrustworthy.
+
+    Callers get one failure mode rather than jwt's several, so no endpoint can
+    accidentally treat an expired token differently from a forged one.
+    """
+    try:
+        return jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
+    except jwt.PyJWTError as exc:
+        raise ValueError("invalid_token") from exc

backend/app/models/user.py

@@ -0,0 +1,23 @@
+"""
+Documents this app's shape, MongoDB doesn't enforce it. Kept here so anyone
+(human or agentic IDE) touching user_repository.py knows exactly what a users
+document looks like.
+
+users collection:
+{
+  "_id": str (uuid4),
+  "email": str,            # stored lowercased; unique, see note below
+  "hashed_password": str,  # bcrypt, never the plaintext
+  "role": "candidate" | "employer",
+  "created_at": datetime (UTC),
+}
+
+The password is never stored or logged in any other form, and no query in
+user_repository.py returns it except the one login needs.
+
+Uniqueness note: email uniqueness is currently enforced by a read-before-write in
+auth_service.register_user, which is racy under concurrent signups. A unique index
+on `email` is the real fix:
+
+    db.users.create_index("email", unique=True)
+"""

backend/app/repositories/user_repository.py

@@ -0,0 +1,27 @@
+"""
+Only file allowed to query the users collection directly.
+Services call these functions instead of touching Mongo themselves —
+swap storage engines later by editing only this file.
+"""
+from typing import Optional
+
+from app.db.mongodb import get_collection
+
+collection = get_collection("users")
+
+
+def _normalise_email(email: str) -> str:
+    """Emails are matched case-insensitively, so they are stored folded."""
+    return email.strip().lower()
+
+
+async def create_user(doc: dict) -> None:
+    await collection.insert_one(doc)
+
+
+async def get_user_by_email(email: str) -> Optional[dict]:
+    return await collection.find_one({"email": _normalise_email(email)})
+
+
+async def get_user_by_id(user_id: str) -> Optional[dict]:
+    return await collection.find_one({"_id": user_id})

backend/app/schemas/auth.py

@@ -0,0 +1,25 @@
+"""Request/response DTOs for the auth endpoints — what crosses the wire."""
+from typing import Literal
+
+from pydantic import BaseModel, EmailStr, Field
+
+Role = Literal["candidate", "employer"]
+
+
+class RegisterRequest(BaseModel):
+    email: EmailStr
+    # bcrypt only considers the first 72 bytes, so longer input is rejected rather
+    # than silently truncated at the boundary.
+    password: str = Field(min_length=8, max_length=72)
+    role: Role = "candidate"
+
+
+class LoginRequest(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    role: Role

backend/app/schemas/quiz.py

@@ -10,7 +10,8 @@
 
 class QuizGenerateRequest(BaseModel):
     repo_url: str
-    user_id: Optional[str] = None
+    # user_id is deliberately absent: it comes from the access token, so a caller
+    # cannot attribute a quiz attempt to somebody else by editing the body.
 
 
 class QuizQuestion(BaseModel):

backend/app/services/auth_service.py

@@ -0,0 +1,54 @@
+"""
+Business logic for authentication.
+
+Endpoints stay thin: they translate these outcomes into status codes and never
+inspect password material themselves.
+"""
+import uuid
+from datetime import datetime, timezone
+
+from app.core.security import create_access_token, hash_password, verify_password
+from app.repositories import user_repository
+
+
+async def register_user(email: str, password: str, role: str) -> dict:
+    """Creates the account and returns a token, so signup does not need a second round trip."""
+    normalised = email.strip().lower()
+
+    if await user_repository.get_user_by_email(normalised):
+        raise ValueError("email_taken")
+
+    user_id = str(uuid.uuid4())
+    await user_repository.create_user(
+        {
+            "_id": user_id,
+            "email": normalised,
+            "hashed_password": hash_password(password),
+            "role": role,
+            "created_at": datetime.now(timezone.utc),
+        }
+    )
+    return {
+        "access_token": create_access_token(user_id, role),
+        "token_type": "bearer",
+        "role": role,
+    }
+
+
+async def authenticate_user(email: str, password: str) -> dict:
+    """
+    Verifies credentials and issues a token.
+
+    Raises the same LookupError whether the email is unknown or the password is
+    wrong, so the endpoint cannot become an oracle for which accounts exist.
+    """
+    user = await user_repository.get_user_by_email(email)
+    if not user or not verify_password(password, user.get("hashed_password", "")):
+        raise LookupError("bad_credentials")
+
+    role = user.get("role", "candidate")
+    return {
+        "access_token": create_access_token(user["_id"], role),
+        "token_type": "bearer",
+        "role": role,
+    }

backend/requirements.txt

@@ -3,7 +3,10 @@ uvicorn[standard]
 motor
 httpx
 google-generativeai
-pydantic
+pydantic[email]
 pydantic-settings
+passlib[bcrypt]
+bcrypt<5  # passlib 1.7.4's bcrypt backend fails its self-test against bcrypt 5.x
+pyjwt
 pytest
 pytest-asyncio