diff --git a/backend/README.md b/backend/README.md index 82ea6cc..f53f2da 100644 --- a/backend/README.md +++ b/backend/README.md @@ -6,9 +6,22 @@ FastAPI application designed to run on the NVIDIA Jetson Orin Nano, with local d ## Components -- **FastAPI REST API**: Health-check and username/PIN login endpoints. -- **Authentication**: Bcrypt-hashed PINs and active sessions. -- **Database**: SQLite with idempotent SQL migrations. +- **FastAPI REST API**: + - Health checks (`/health`) + - Authentication (`/login`, `/logout`) + - Student self-service lifecycle (`/signup`, `/users/me`, `/users/me/pin`, `/users/me/username`) + - Staff management (`/users`, `/users/{id}/reset-pin`, `/users/{id}`, `/users/{id}/recover`) + - Privileged system audit trail (`/audit-logs`) +- **Security & Access Control**: + - Role-based access control (RBAC) with `student`, `teacher`, and `admin` roles. + - Forced PIN rotation enforcement and policy validation (4–8 numeric digits). + - In-memory rate limiting: credential lockout limiter (consecutive failure backoff) and global sliding window limiters (signup / rate-throttling). + - Bearer session token lifecycle with active session deactivation. + - Zero-credential logging guards and anti-oracle check ordering. +- **Database & Migrations**: + - SQLite database with foreign keys and index optimization. + - Numbered, idempotent schema migrations (`001` through `006`). + - Append-only `audit_logs` table tracking privileged user and account mutations. The following items are placeholders for planned work and are not fully implemented in the current backend: @@ -17,6 +30,28 @@ The following items are placeholders for planned work and are not fully implemen - **Math Validation**: Deterministic SymPy engine and containment guardrail. - **ASR Service**: Meta Omnilingual ASR 300M CTC int8 (`sherpa-onnx`) integration for K'iche' speech-to-text. +--- + +## API Summary + +| Method | Path | Auth / Role | Description | +| :--- | :--- | :--- | :--- | +| `GET` | `/health` | Public | System and SQLite health check | +| `POST` | `/signup` | Public (Rate-limited) | Student self-registration | +| `POST` | `/login` | Public (Rate-limited) | Authenticate with username and PIN; returns session token | +| `POST` | `/logout` | Bearer Token | Invalidate current caller session | +| `GET` | `/users/me` | Bearer Token | Return profile of authenticated user | +| `PATCH` | `/users/me/pin` | Bearer Token | Self-service PIN change (clears forced rotation flag) | +| `PATCH` | `/users/me/username` | Bearer Token | Self-service username change | +| `GET` | `/users` | Teacher, Admin | List active accounts, or deleted accounts with `?include_deleted=true` | +| `POST` | `/users` | Teacher, Admin | Staff user creation (Teachers: student/teacher; Admins: any role) | +| `POST` | `/users/{id}/reset-pin` | Teacher, Admin | Issue 6-digit temp PIN, invalidate sessions, and require PIN rotation | +| `DELETE` | `/users/{id}` | Teacher, Admin | Soft-delete user, anonymize username, preserve telemetry, last-admin guard | +| `POST` | `/users/{id}/recover` | Teacher, Admin | Restore soft-deleted account under new username with temporary PIN | +| `GET` | `/audit-logs` | Admin Only | View up to 500 append-only audit trail records | + +--- + ## Environment Setup Ensure you are using Python 3.10 or newer. @@ -74,12 +109,18 @@ python -m uvicorn src.main:app --reload The interactive API documentation is available at . #### Running Tests & Linters: -Run pytest with code coverage +Run pytest with full code coverage: ```bash python -m pytest ``` -Run Ruff linter and formatter checks manually + +Run a specific test subpackage: +```bash +python -m pytest tests/api/staff/ -o addopts="--strict-markers" ``` + +Run Ruff linter and formatter checks manually: +```bash ruff check . ruff format --check . ``` @@ -102,40 +143,90 @@ python -m uvicorn src.main:app --host 0.0.0.0 --port 8000 The database `tutorbox.db` will be initialized and migrated automatically on startup. Use the `DATABASE_PATH` environment variable to configure a custom SQLite file location. +--- + ## Project Structure Generated environments, caches, and build artifacts are omitted from this overview. ```text -. +backend/ ├── migrations/ │ ├── 001_initial_schema.sql │ ├── 002_add_user_role.sql -│ └── 003_add_lookup_indexes.sql +│ ├── 003_add_lookup_indexes.sql +│ ├── 004_add_must_change_pin.sql +│ ├── 005_add_users_deleted_at.sql +│ └── 006_add_audit_logs.sql ├── src/ │ ├── __init__.py │ ├── main.py │ ├── api/ │ │ ├── __init__.py -│ │ ├── auth.py -│ │ └── health.py +│ │ ├── auth/ +│ │ │ ├── __init__.py +│ │ │ ├── login.py +│ │ │ └── logout.py +│ │ ├── health.py +│ │ ├── staff/ +│ │ │ ├── __init__.py +│ │ │ ├── audit.py +│ │ │ ├── lifecycle.py +│ │ │ ├── reset_pin.py +│ │ │ └── users.py +│ │ └── users/ +│ │ ├── __init__.py +│ │ ├── credentials.py +│ │ ├── profile.py +│ │ └── signup.py │ ├── db/ │ │ ├── __init__.py +│ │ ├── audit.py │ │ ├── database.py │ │ └── migrations.py │ └── security/ │ ├── __init__.py │ ├── auth.py -│ └── rate_limit.py +│ ├── rate_limit/ +│ │ ├── __init__.py +│ │ ├── lockout.py +│ │ └── sliding_window.py +│ ├── session.py +│ └── validation.py ├── tests/ │ ├── __init__.py │ ├── conftest.py -│ ├── test_auth.py -│ ├── test_database.py -│ ├── test_health.py -│ ├── test_migrations.py -│ ├── test_rate_limit.py -│ └── test_security_pin.py +│ ├── api/ +│ │ ├── __init__.py +│ │ ├── auth/ +│ │ │ ├── __init__.py +│ │ │ ├── test_login.py +│ │ │ └── test_logout.py +│ │ ├── staff/ +│ │ │ ├── __init__.py +│ │ │ ├── test_audit.py +│ │ │ ├── test_delete.py +│ │ │ ├── test_recover.py +│ │ │ ├── test_reset_pin.py +│ │ │ └── test_users.py +│ │ ├── users/ +│ │ │ ├── __init__.py +│ │ │ ├── test_change_pin.py +│ │ │ ├── test_change_username.py +│ │ │ ├── test_profile.py +│ │ │ └── test_signup.py +│ │ └── test_health.py +│ ├── db/ +│ │ ├── __init__.py +│ │ ├── test_audit.py +│ │ ├── test_database.py +│ │ └── test_migrations.py +│ └── security/ +│ ├── __init__.py +│ ├── test_auth_pin.py +│ ├── test_rate_limit.py +│ ├── test_security_pin.py +│ └── test_session_auth.py ├── pyproject.toml └── README.md ``` diff --git a/backend/migrations/004_add_must_change_pin.sql b/backend/migrations/004_add_must_change_pin.sql new file mode 100644 index 0000000..d7cc10c --- /dev/null +++ b/backend/migrations/004_add_must_change_pin.sql @@ -0,0 +1,3 @@ +-- 004_add_must_change_pin.sql +-- Flags accounts that must rotate their PIN on next login (staff-initiated resets). +ALTER TABLE users ADD COLUMN must_change_pin INTEGER NOT NULL DEFAULT 0; diff --git a/backend/migrations/005_add_users_deleted_at.sql b/backend/migrations/005_add_users_deleted_at.sql new file mode 100644 index 0000000..e0df839 --- /dev/null +++ b/backend/migrations/005_add_users_deleted_at.sql @@ -0,0 +1,4 @@ +-- 005_add_users_deleted_at.sql +-- Soft-delete marker + original-username retention for account recovery. +ALTER TABLE users ADD COLUMN deleted_at TIMESTAMP; +ALTER TABLE users ADD COLUMN former_username TEXT; diff --git a/backend/migrations/006_add_audit_logs.sql b/backend/migrations/006_add_audit_logs.sql new file mode 100644 index 0000000..7f8e196 --- /dev/null +++ b/backend/migrations/006_add_audit_logs.sql @@ -0,0 +1,12 @@ +-- 006_add_audit_logs.sql +-- Append-only trail for privileged/account actions. +CREATE TABLE IF NOT EXISTS audit_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + actor_user_id INTEGER, + action TEXT NOT NULL, + target_user_id INTEGER, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_audit_logs_actor ON audit_logs (actor_user_id); +CREATE INDEX IF NOT EXISTS idx_audit_logs_target ON audit_logs (target_user_id); diff --git a/backend/src/api/__init__.py b/backend/src/api/__init__.py index f2b2610..a5006a5 100644 --- a/backend/src/api/__init__.py +++ b/backend/src/api/__init__.py @@ -1,5 +1,5 @@ """TutorBox API package.""" -from . import auth, health +from . import auth, health, staff, users -__all__ = ["auth", "health"] +__all__ = ["auth", "health", "staff", "users"] diff --git a/backend/src/api/auth/__init__.py b/backend/src/api/auth/__init__.py new file mode 100644 index 0000000..948832f --- /dev/null +++ b/backend/src/api/auth/__init__.py @@ -0,0 +1,26 @@ +"""Auth API package.""" + +from fastapi import APIRouter + +from .login import ( + LoginRequest, + LoginResponse, +) +from .login import ( + router as login_router, +) +from .logout import ( + router as logout_router, +) + +router = APIRouter() +router.include_router(login_router) +router.include_router(logout_router) + +__all__ = [ + "LoginRequest", + "LoginResponse", + "login_router", + "logout_router", + "router", +] diff --git a/backend/src/api/auth.py b/backend/src/api/auth/login.py similarity index 74% rename from backend/src/api/auth.py rename to backend/src/api/auth/login.py index 606d4e8..ee2d5ec 100644 --- a/backend/src/api/auth.py +++ b/backend/src/api/auth/login.py @@ -2,41 +2,32 @@ import uuid from fastapi import APIRouter, HTTPException, status -from pydantic import BaseModel, Field +from pydantic import BaseModel from db.database import get_db -from security.auth import verify_pin -from security.rate_limit import check_rate_limit, login_rate_limiter +from security import ( + PinField, + UsernameField, + check_rate_limit, + login_rate_limiter, + verify_pin, +) logger = logging.getLogger(__name__) router = APIRouter() -USERNAME_PATTERN = r"^[A-Za-z0-9_.-]{3,32}$" -PIN_PATTERN = r"^\d{4,8}$" - class LoginRequest(BaseModel): - username: str = Field( - ..., - min_length=3, - max_length=32, - pattern=USERNAME_PATTERN, - examples=["student1"], - ) - pin: str = Field( - ..., - min_length=4, - max_length=8, - pattern=PIN_PATTERN, - examples=["1234"], - ) + username: UsernameField + pin: PinField class LoginResponse(BaseModel): session_id: str username: str status: str = "authenticated" + must_change_pin: bool = False @router.post("/login", response_model=LoginResponse) @@ -51,7 +42,8 @@ def login(request: LoginRequest): with get_db() as conn: cursor = conn.cursor() cursor.execute( - "SELECT id, username, hashed_pin FROM users WHERE username = ?", + "SELECT id, username, hashed_pin, must_change_pin " + "FROM users WHERE username = ? AND deleted_at IS NULL", (request.username,), ) user = cursor.fetchone() @@ -89,4 +81,8 @@ def login(request: LoginRequest): conn.commit() logger.info("Login successful for user '%s'.", username) - return LoginResponse(session_id=session_id, username=username) + return LoginResponse( + session_id=session_id, + username=username, + must_change_pin=bool(user["must_change_pin"]), + ) diff --git a/backend/src/api/auth/logout.py b/backend/src/api/auth/logout.py new file mode 100644 index 0000000..067fc85 --- /dev/null +++ b/backend/src/api/auth/logout.py @@ -0,0 +1,28 @@ +import logging +from typing import Annotated + +from fastapi import APIRouter, Depends + +from db.database import get_db +from security import AuthContext, get_current_session + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post("/logout") +def logout(ctx: Annotated[AuthContext, Depends(get_current_session)]): + """ + Deactivates the caller's current session. + """ + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "UPDATE sessions SET is_active = 0 WHERE id = ? AND is_active = 1", + (ctx.session_id,), + ) + conn.commit() + + logger.info("User '%s' logged out.", ctx.username) + return {"detail": "Logged out."} diff --git a/backend/src/api/staff/__init__.py b/backend/src/api/staff/__init__.py new file mode 100644 index 0000000..5ca1589 --- /dev/null +++ b/backend/src/api/staff/__init__.py @@ -0,0 +1,54 @@ +"""Staff API package.""" + +from fastapi import APIRouter + +from .audit import ( + AuditLogsResponse, +) +from .audit import ( + router as audit_router, +) +from .lifecycle import ( + DeleteUserResponse, + RecoverUserRequest, + RecoverUserResponse, +) +from .lifecycle import ( + router as lifecycle_router, +) +from .reset_pin import ( + ResetPinResponse, +) +from .reset_pin import ( + router as reset_pin_router, +) +from .users import ( + CreateUserRequest, + CreateUserResponse, + UserListResponse, +) +from .users import ( + router as users_router, +) + +router = APIRouter() +router.include_router(users_router) +router.include_router(reset_pin_router) +router.include_router(lifecycle_router) +router.include_router(audit_router) + +__all__ = [ + "AuditLogsResponse", + "CreateUserRequest", + "CreateUserResponse", + "DeleteUserResponse", + "RecoverUserRequest", + "RecoverUserResponse", + "ResetPinResponse", + "UserListResponse", + "audit_router", + "lifecycle_router", + "reset_pin_router", + "router", + "users_router", +] diff --git a/backend/src/api/staff/audit.py b/backend/src/api/staff/audit.py new file mode 100644 index 0000000..b4922ac --- /dev/null +++ b/backend/src/api/staff/audit.py @@ -0,0 +1,54 @@ +"""Staff audit logs read endpoint.""" + +import logging +from typing import Annotated, Any + +from fastapi import APIRouter, Depends +from pydantic import BaseModel + +from db.database import get_db +from security import ( + AuthContext, + require_roles, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class AuditLogsResponse(BaseModel): + logs: list[dict[str, Any]] + + +@router.get( + "/audit-logs", + response_model=AuditLogsResponse, +) +def read_audit_logs( + ctx: Annotated[AuthContext, Depends(require_roles("admin"))], +): + """ + Read up to 500 audit logs in reverse chronological order. Admin only. + """ + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id, actor_user_id, action, target_user_id, created_at " + "FROM audit_logs ORDER BY id DESC LIMIT 500" + ) + rows = cursor.fetchall() + + logger.info("Audit logs viewed by admin '%s'.", ctx.username) + return AuditLogsResponse( + logs=[ + { + "id": r["id"], + "actor_user_id": r["actor_user_id"], + "action": r["action"], + "target_user_id": r["target_user_id"], + "created_at": str(r["created_at"]) if r["created_at"] else None, + } + for r in rows + ] + ) diff --git a/backend/src/api/staff/lifecycle.py b/backend/src/api/staff/lifecycle.py new file mode 100644 index 0000000..981bf10 --- /dev/null +++ b/backend/src/api/staff/lifecycle.py @@ -0,0 +1,160 @@ +import logging +import secrets +import sqlite3 +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel + +from db.audit import record_audit +from db.database import get_db +from security import ( + AuthContext, + UsernameField, + hash_pin, + require_roles, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class DeleteUserResponse(BaseModel): + detail: str = "Account deleted." + + +class RecoverUserRequest(BaseModel): + username: UsernameField + + +class RecoverUserResponse(BaseModel): + username: str + temporary_pin: str + detail: str = "Account recovered. User must set a new PIN on next login." + + +def _soft_delete_user( + conn: sqlite3.Connection, target_id: int, target_role: str +) -> None: + cursor = conn.cursor() + + # Last-admin guard: the appliance must never lose its final administrator. + if target_role == "admin": + cursor.execute( + "SELECT COUNT(*) AS n FROM users WHERE role = 'admin' AND deleted_at IS NULL" + ) + if cursor.fetchone()["n"] <= 1: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Cannot delete the last remaining admin account.", + ) + + anon_username = f"deleted_user_{target_id}_{secrets.token_hex(4)}" + unusable_hash = hash_pin(secrets.token_hex(16)) + cursor.execute( + "UPDATE users SET former_username = username, username = ?, hashed_pin = ?, " + "deleted_at = CURRENT_TIMESTAMP WHERE id = ? AND deleted_at IS NULL", + (anon_username, unusable_hash, target_id), + ) + cursor.execute( + "UPDATE sessions SET is_active = 0 WHERE user_id = ? AND is_active = 1", + (target_id,), + ) + + +@router.delete( + "/users/{user_id}", + response_model=DeleteUserResponse, + status_code=status.HTTP_200_OK, +) +def delete_user( + user_id: int, + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], +): + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id, username, role FROM users WHERE id = ? AND deleted_at IS NULL", + (user_id,), + ) + target = cursor.fetchone() + if target is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="User not found." + ) + if ctx.role == "teacher" and target["role"] == "admin": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only admins may delete admin accounts.", + ) + + _soft_delete_user(conn, user_id, target["role"]) + record_audit( + conn, + actor_user_id=ctx.user_id, + action="account_deleted", + target_user_id=user_id, + ) + conn.commit() + + logger.info("User id %d soft-deleted by '%s'.", user_id, ctx.username) + return DeleteUserResponse() + + +@router.post( + "/users/{user_id}/recover", + response_model=RecoverUserResponse, + status_code=status.HTTP_200_OK, +) +def recover_user( + user_id: int, + payload: RecoverUserRequest, + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], +): + temp_pin = f"{secrets.randbelow(10**6):06d}" + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id, role FROM users WHERE id = ? AND deleted_at IS NOT NULL", + (user_id,), + ) + target = cursor.fetchone() + if target is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Deleted user not found." + ) + # Uniform matrix: teachers recover students/teachers; admins anyone. + if ctx.role == "teacher" and target["role"] == "admin": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only admins may recover admin accounts.", + ) + + try: + cursor.execute( + "UPDATE users SET username = ?, hashed_pin = ?, " + "deleted_at = NULL, must_change_pin = 1 " + "WHERE id = ? AND deleted_at IS NOT NULL", + (payload.username, hash_pin(temp_pin), user_id), + ) + except sqlite3.IntegrityError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Username already taken. Choose another for this account.", + ) + record_audit( + conn, + actor_user_id=ctx.user_id, + action="account_recovered", + target_user_id=user_id, + ) + conn.commit() + + logger.info( + "Account recovered: user id %d by '%s'.", user_id, ctx.username + ) # NEVER log temp_pin + return RecoverUserResponse( + username=payload.username, + temporary_pin=temp_pin, + ) diff --git a/backend/src/api/staff/reset_pin.py b/backend/src/api/staff/reset_pin.py new file mode 100644 index 0000000..b152629 --- /dev/null +++ b/backend/src/api/staff/reset_pin.py @@ -0,0 +1,78 @@ +import logging +import secrets +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel + +from db.audit import record_audit +from db.database import get_db +from security import ( + AuthContext, + hash_pin, + require_roles, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class ResetPinResponse(BaseModel): + username: str + temporary_pin: str + + +@router.post( + "/users/{user_id}/reset-pin", + response_model=ResetPinResponse, + status_code=status.HTTP_200_OK, +) +def reset_pin( + user_id: int, + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], +): + """ + Teacher-initiated temporary PIN reset. + Generates a 6-digit temporary PIN, sets must_change_pin=1, and invalidates all active sessions for the target user. + """ + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id, username, role FROM users WHERE id = ? AND deleted_at IS NULL", + (user_id,), + ) + target = cursor.fetchone() + if target is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found.", + ) + + # Uniform matrix: teachers act on students/teachers; admins on anyone. + if ctx.role == "teacher" and target["role"] == "admin": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only admins may reset admin PINs.", + ) + + temp_pin = f"{secrets.randbelow(10**6):06d}" + cursor.execute( + "UPDATE users SET hashed_pin = ?, must_change_pin = 1 WHERE id = ?", + (hash_pin(temp_pin), user_id), + ) + # Invalidate all active sessions for the target user + cursor.execute( + "UPDATE sessions SET is_active = 0 WHERE user_id = ? AND is_active = 1", + (user_id,), + ) + record_audit( + conn, + actor_user_id=ctx.user_id, + action="pin_reset", + target_user_id=user_id, + ) + conn.commit() + + logger.info("PIN reset issued for user id %d.", user_id) # NEVER log temp_pin + return ResetPinResponse(username=target["username"], temporary_pin=temp_pin) diff --git a/backend/src/api/staff/users.py b/backend/src/api/staff/users.py new file mode 100644 index 0000000..24ac0ee --- /dev/null +++ b/backend/src/api/staff/users.py @@ -0,0 +1,147 @@ +import logging +import sqlite3 +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel + +from db.audit import record_audit +from db.database import get_db +from security import ( + AuthContext, + PinField, + RoleField, + UsernameField, + hash_pin, + require_roles, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class CreateUserRequest(BaseModel): + username: UsernameField + pin: PinField + role: RoleField = "student" + + +class CreateUserResponse(BaseModel): + username: str + role: str + + +class UserListResponse(BaseModel): + users: list[dict[str, Any]] + + +@router.get("/users", response_model=UserListResponse) +def list_users( + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], + include_deleted: bool = False, +): + """ + Roster view. Lists active users, or minimal metadata for deleted accounts when include_deleted=True. + """ + with get_db() as conn: + cursor = conn.cursor() + if include_deleted: + cursor.execute( + "SELECT id, role, former_username, deleted_at FROM users " + "WHERE deleted_at IS NOT NULL ORDER BY deleted_at DESC LIMIT 1000" + ) + return UserListResponse( + users=[ + { + "id": r["id"], + "role": r["role"], + "former_username": r["former_username"], + "deleted_at": r["deleted_at"], + } + for r in cursor.fetchall() + ] + ) + + cursor.execute( + "SELECT id, username, role, created_at, must_change_pin " + "FROM users WHERE deleted_at IS NULL " + "ORDER BY username LIMIT 1000" + ) + rows = cursor.fetchall() + + return UserListResponse( + users=[ + { + "id": r["id"], + "username": r["username"], + "role": r["role"], + "created_at": r["created_at"], + "must_change_pin": bool(r["must_change_pin"]), + } + for r in rows + ] + ) + + +@router.post( + "/users", + response_model=CreateUserResponse, + status_code=status.HTTP_201_CREATED, +) +def create_user( + payload: CreateUserRequest, + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], +): + """ + Staff user creation. Teachers can create student and teacher accounts. Admins can create any account. + """ + # Teachers create students AND teachers; only admins create admins. + if ctx.role == "teacher" and payload.role == "admin": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only admins may create admin accounts.", + ) + + logger.info( + "User creation attempt by '%s' (role: %s) for new user '%s' (role: %s).", + ctx.username, + ctx.role, + payload.username, + payload.role, + ) + + hashed = hash_pin(payload.pin) + + try: + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role) VALUES (?, ?, ?)", + (payload.username, hashed, payload.role), + ) + new_user_id = cursor.lastrowid + record_audit( + conn, + actor_user_id=ctx.user_id, + action="user_created", + target_user_id=new_user_id, + ) + conn.commit() + except sqlite3.IntegrityError: + logger.warning( + "User creation conflict: Username '%s' already exists.", + payload.username, + ) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Username already taken.", + ) + + logger.info( + "User '%s' (role: %s) created successfully by '%s'.", + payload.username, + payload.role, + ctx.username, + ) + return CreateUserResponse(username=payload.username, role=payload.role) diff --git a/backend/src/api/users/__init__.py b/backend/src/api/users/__init__.py new file mode 100644 index 0000000..fb629a9 --- /dev/null +++ b/backend/src/api/users/__init__.py @@ -0,0 +1,36 @@ +"""Users API package.""" + +from fastapi import APIRouter + +from .credentials import ( + ChangePinRequest, + ChangeUsernameRequest, + CredentialChangeResponse, + _change_credential, +) +from .credentials import ( + router as credentials_router, +) +from .profile import UserProfileResponse +from .profile import router as profile_router +from .signup import SignupRequest, SignupResponse +from .signup import router as signup_router + +router = APIRouter() +router.include_router(signup_router) +router.include_router(profile_router) +router.include_router(credentials_router) + +__all__ = [ + "ChangePinRequest", + "ChangeUsernameRequest", + "CredentialChangeResponse", + "SignupRequest", + "SignupResponse", + "UserProfileResponse", + "_change_credential", + "credentials_router", + "profile_router", + "router", + "signup_router", +] diff --git a/backend/src/api/users/credentials.py b/backend/src/api/users/credentials.py new file mode 100644 index 0000000..10ef502 --- /dev/null +++ b/backend/src/api/users/credentials.py @@ -0,0 +1,149 @@ +import logging +import sqlite3 +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel + +from db.audit import record_audit +from db.database import get_db +from security import ( + AuthContext, + PinField, + UsernameField, + check_rate_limit, + ensure_no_pending_rotation, + get_current_session, + hash_pin, + login_rate_limiter, + verify_pin, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class ChangeUsernameRequest(BaseModel): + current_pin: PinField + new_username: UsernameField + + +class ChangePinRequest(BaseModel): + current_pin: PinField + new_pin: PinField + + +class CredentialChangeResponse(BaseModel): + detail: str = "Credentials updated. Please sign in again." + + +def _change_credential( + ctx: AuthContext, + payload: ChangeUsernameRequest | ChangePinRequest, + *, + kind: str, +) -> CredentialChangeResponse: + logger.info("Credential change (%s) for user '%s'.", kind, ctx.username) + + check_rate_limit(ctx.username) + + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT id, username, hashed_pin FROM users WHERE id = ? AND deleted_at IS NULL", + (ctx.user_id,), + ) + user = cursor.fetchone() + if user is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid session.", + ) + + # 1) Anti-oracle check ordering: verify current PIN FIRST + if not verify_pin(payload.current_pin, user["hashed_pin"]): + login_rate_limiter.record_failure(ctx.username) + logger.warning( + "Credential change failed (bad current PIN): %s", ctx.username + ) + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid current PIN.", + ) + login_rate_limiter.record_success(ctx.username) + + # 2) Compare new against current only AFTER successful PIN verification + if kind == "pin": + assert isinstance(payload, ChangePinRequest) + if payload.new_pin == payload.current_pin: + raise HTTPException( + status_code=422, + detail="New PIN must differ from current PIN.", + ) + cursor.execute( + "UPDATE users SET hashed_pin = ?, must_change_pin = 0 WHERE id = ?", + (hash_pin(payload.new_pin), ctx.user_id), + ) + record_audit( + conn, + actor_user_id=ctx.user_id, + action="pin_changed", + target_user_id=ctx.user_id, + ) + else: + assert isinstance(payload, ChangeUsernameRequest) + if payload.new_username == ctx.username: + raise HTTPException( + status_code=422, + detail="New username must differ from current username.", + ) + try: + cursor.execute( + "UPDATE users SET username = ? WHERE id = ?", + (payload.new_username, ctx.user_id), + ) + except sqlite3.IntegrityError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Username already taken.", + ) + record_audit( + conn, + actor_user_id=ctx.user_id, + action="username_changed", + target_user_id=ctx.user_id, + ) + + # 3) Deactivate ONLY the caller's active session + cursor.execute( + "UPDATE sessions SET is_active = 0 WHERE id = ? AND is_active = 1", + (ctx.session_id,), + ) + conn.commit() + + logger.info("Credential change (%s) successful for user.", kind) + return CredentialChangeResponse() + + +@router.patch("/users/me/username", response_model=CredentialChangeResponse) +def change_username( + payload: ChangeUsernameRequest, + ctx: Annotated[AuthContext, Depends(ensure_no_pending_rotation)], +): + """ + Update username. Requires current PIN verification. Invalidates caller's session. + """ + return _change_credential(ctx, payload, kind="username") + + +@router.patch("/users/me/pin", response_model=CredentialChangeResponse) +def change_pin( + payload: ChangePinRequest, + ctx: Annotated[AuthContext, Depends(get_current_session)], +): + """ + Update PIN. Requires current PIN verification. Clears must_change_pin flag. + Permitted during pending rotation (allowlist). Invalidates caller's session. + """ + return _change_credential(ctx, payload, kind="pin") diff --git a/backend/src/api/users/profile.py b/backend/src/api/users/profile.py new file mode 100644 index 0000000..4ad4003 --- /dev/null +++ b/backend/src/api/users/profile.py @@ -0,0 +1,31 @@ +import logging +from typing import Annotated + +from fastapi import APIRouter, Depends +from pydantic import BaseModel + +from security import AuthContext, get_current_session + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class UserProfileResponse(BaseModel): + user_id: int + username: str + role: str + must_change_pin: bool + + +@router.get("/users/me", response_model=UserProfileResponse) +def get_me(ctx: Annotated[AuthContext, Depends(get_current_session)]): + """ + Returns caller's profile. Permitted during pending rotation (allowlist). + """ + return UserProfileResponse( + user_id=ctx.user_id, + username=ctx.username, + role=ctx.role, + must_change_pin=ctx.must_change_pin, + ) diff --git a/backend/src/api/users/signup.py b/backend/src/api/users/signup.py new file mode 100644 index 0000000..fad0f2a --- /dev/null +++ b/backend/src/api/users/signup.py @@ -0,0 +1,76 @@ +import logging +import sqlite3 + +from fastapi import APIRouter, HTTPException, status +from pydantic import BaseModel + +from db.audit import record_audit +from db.database import get_db +from security import ( + PinField, + UsernameField, + hash_pin, + signup_rate_limiter, +) + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +class SignupRequest(BaseModel): + username: UsernameField + pin: PinField + + +class SignupResponse(BaseModel): + username: str + role: str + + +@router.post( + "/signup", + response_model=SignupResponse, + status_code=status.HTTP_201_CREATED, +) +def signup(request: SignupRequest): + """ + Student self-signup. Role is always 'student' with direct activation. + """ + logger.info("Signup attempt for username: %s", request.username) + + if not signup_rate_limiter.allow(): + logger.warning("Signup rate limit exceeded.") + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail="Too many signup attempts. Please try again later.", + ) + + hashed = hash_pin(request.pin) + + try: + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role) VALUES (?, ?, 'student')", + (request.username, hashed), + ) + new_user_id = cursor.lastrowid + record_audit( + conn, + actor_user_id=None, + action="signup", + target_user_id=new_user_id, + ) + conn.commit() + except sqlite3.IntegrityError: + logger.warning( + "Signup conflict: Username '%s' already exists.", request.username + ) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Username already taken.", + ) + + logger.info("User '%s' registered successfully.", request.username) + return SignupResponse(username=request.username, role="student") diff --git a/backend/src/db/__init__.py b/backend/src/db/__init__.py index 8863ff6..dc450bc 100644 --- a/backend/src/db/__init__.py +++ b/backend/src/db/__init__.py @@ -1,5 +1,6 @@ """TutorBox DB package.""" -from . import database, migrations +from . import audit, database, migrations +from .audit import VALID_ACTIONS, record_audit -__all__ = ["database", "migrations"] +__all__ = ["VALID_ACTIONS", "audit", "database", "migrations", "record_audit"] diff --git a/backend/src/db/audit.py b/backend/src/db/audit.py new file mode 100644 index 0000000..e8951e6 --- /dev/null +++ b/backend/src/db/audit.py @@ -0,0 +1,34 @@ +"""Audit trail helper.""" + +import sqlite3 + +VALID_ACTIONS = frozenset( + { + "signup", + "user_created", + "pin_reset", + "username_changed", + "pin_changed", + "account_deleted", + "account_recovered", + } +) + + +def record_audit( + conn: sqlite3.Connection, + *, + actor_user_id: int | None, + action: str, + target_user_id: int | None = None, +) -> None: + """ + Records an append-only audit trail event in the audit_logs table. + """ + if action not in VALID_ACTIONS: + raise ValueError(f"Unknown audit action: {action}") + + conn.execute( + "INSERT INTO audit_logs (actor_user_id, action, target_user_id) VALUES (?, ?, ?)", + (actor_user_id, action, target_user_id), + ) diff --git a/backend/src/main.py b/backend/src/main.py index 7a41b96..c97ce28 100644 --- a/backend/src/main.py +++ b/backend/src/main.py @@ -5,6 +5,8 @@ from api.auth import router as auth_router from api.health import router as health_router +from api.staff import router as staff_router +from api.users import router as users_router from db.database import get_db_path from db.migrations import apply_migrations @@ -35,3 +37,5 @@ async def lifespan(app: FastAPI): app.include_router(health_router) app.include_router(auth_router) +app.include_router(users_router) +app.include_router(staff_router) diff --git a/backend/src/security/__init__.py b/backend/src/security/__init__.py index 53acd44..cbb234a 100644 --- a/backend/src/security/__init__.py +++ b/backend/src/security/__init__.py @@ -1,5 +1,57 @@ """TutorBox Security package.""" -from . import auth, rate_limit +from . import auth, rate_limit, session, validation +from .auth import hash_pin, verify_pin +from .rate_limit import ( + InMemoryRateLimiter, + SlidingWindowLimiter, + check_rate_limit, + login_rate_limiter, + signup_rate_limiter, +) +from .session import ( + AuthContext, + ensure_no_pending_rotation, + get_current_session, + require_roles, +) +from .validation import ( + ALLOWED_ROLES, + PIN_MAX_LENGTH, + PIN_MIN_LENGTH, + PIN_PATTERN, + USERNAME_MAX_LENGTH, + USERNAME_MIN_LENGTH, + USERNAME_PATTERN, + PinField, + RoleField, + UsernameField, +) -__all__ = ["auth", "rate_limit"] +__all__ = [ + "ALLOWED_ROLES", + "PIN_MAX_LENGTH", + "PIN_MIN_LENGTH", + "PIN_PATTERN", + "USERNAME_MAX_LENGTH", + "USERNAME_MIN_LENGTH", + "USERNAME_PATTERN", + "AuthContext", + "InMemoryRateLimiter", + "PinField", + "RoleField", + "SlidingWindowLimiter", + "UsernameField", + "auth", + "check_rate_limit", + "ensure_no_pending_rotation", + "get_current_session", + "hash_pin", + "login_rate_limiter", + "rate_limit", + "require_roles", + "session", + "signup_rate_limiter", + "validation", + "verify_pin", +] diff --git a/backend/src/security/rate_limit/__init__.py b/backend/src/security/rate_limit/__init__.py new file mode 100644 index 0000000..c3fdee7 --- /dev/null +++ b/backend/src/security/rate_limit/__init__.py @@ -0,0 +1,29 @@ +"""Rate limiting package.""" + +from .lockout import ( + LOCKOUT_DURATION_SECONDS, + MAX_ATTEMPTS, + MAX_TRACKED_KEYS, + InMemoryRateLimiter, + check_rate_limit, + login_rate_limiter, +) +from .sliding_window import ( + SIGNUP_MAX_EVENTS, + SIGNUP_WINDOW_SECONDS, + SlidingWindowLimiter, + signup_rate_limiter, +) + +__all__ = [ + "LOCKOUT_DURATION_SECONDS", + "MAX_ATTEMPTS", + "MAX_TRACKED_KEYS", + "SIGNUP_MAX_EVENTS", + "SIGNUP_WINDOW_SECONDS", + "InMemoryRateLimiter", + "SlidingWindowLimiter", + "check_rate_limit", + "login_rate_limiter", + "signup_rate_limiter", +] diff --git a/backend/src/security/rate_limit.py b/backend/src/security/rate_limit/lockout.py similarity index 100% rename from backend/src/security/rate_limit.py rename to backend/src/security/rate_limit/lockout.py diff --git a/backend/src/security/rate_limit/sliding_window.py b/backend/src/security/rate_limit/sliding_window.py new file mode 100644 index 0000000..6f0b2ce --- /dev/null +++ b/backend/src/security/rate_limit/sliding_window.py @@ -0,0 +1,40 @@ +import threading +import time +from collections import deque + +SIGNUP_MAX_EVENTS = 30 +SIGNUP_WINDOW_SECONDS = 60 + + +class SlidingWindowLimiter: + """ + Thread-safe global event-window limiter (no per-key state). + Used to bound account-creation floods on the shared Jetson. + """ + + def __init__( + self, + max_events: int = SIGNUP_MAX_EVENTS, + window_seconds: int = SIGNUP_WINDOW_SECONDS, + ): + self.max_events = max_events + self.window_seconds = window_seconds + self._events: deque[float] = deque() + self._lock = threading.Lock() + + def allow(self) -> bool: + now = time.time() + with self._lock: + while self._events and now - self._events[0] > self.window_seconds: + self._events.popleft() + if len(self._events) >= self.max_events: + return False + self._events.append(now) + return True + + def clear(self) -> None: + with self._lock: + self._events.clear() + + +signup_rate_limiter = SlidingWindowLimiter(SIGNUP_MAX_EVENTS, SIGNUP_WINDOW_SECONDS) diff --git a/backend/src/security/session.py b/backend/src/security/session.py new file mode 100644 index 0000000..64c4df9 --- /dev/null +++ b/backend/src/security/session.py @@ -0,0 +1,95 @@ +import logging +from typing import Annotated + +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from pydantic import BaseModel + +from db.database import get_db +from security.validation import UUID_RE + +logger = logging.getLogger(__name__) + +bearer_scheme = HTTPBearer(auto_error=False) + + +class AuthContext(BaseModel): + user_id: int + username: str + role: str + session_id: str + must_change_pin: bool + + +def get_current_session( + credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)], +) -> AuthContext: + if credentials is None or credentials.scheme.lower() != "bearer": + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing or malformed Authorization header.", + ) + token = credentials.credentials.strip() + + # Validate canonical UUID shape BEFORE hitting SQLite (junk-in guard). + if UUID_RE.match(token) is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid session token.", + ) + + with get_db() as conn: + cursor = conn.cursor() + cursor.execute( + "SELECT s.user_id, u.username, u.role, u.must_change_pin " + "FROM sessions s " + "JOIN users u ON u.id = s.user_id " + "WHERE s.id = ? AND s.is_active = 1 AND u.deleted_at IS NULL", + (token,), + ) + row = cursor.fetchone() + + if row is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid or expired session.", + ) + + return AuthContext( + user_id=row["user_id"], + username=row["username"], + role=row["role"], + session_id=token, + must_change_pin=bool(row["must_change_pin"]), + ) + + +def ensure_no_pending_rotation( + ctx: Annotated[AuthContext, Depends(get_current_session)], +) -> AuthContext: + """ + Blocks privileged/interactive endpoints while a PIN rotation is pending. + Allowlist: PATCH /users/me/pin, GET /users/me, POST /logout. + """ + if ctx.must_change_pin: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="PIN change required.", + ) + return ctx + + +def require_roles(*allowed: str): + """Dependency factory: 403 unless the caller's role is in *allowed*.""" + + def checker( + ctx: Annotated[AuthContext, Depends(ensure_no_pending_rotation)], + ) -> AuthContext: + if ctx.role not in allowed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Insufficient permissions.", + ) + return ctx + + return checker diff --git a/backend/src/security/validation.py b/backend/src/security/validation.py new file mode 100644 index 0000000..e4ec108 --- /dev/null +++ b/backend/src/security/validation.py @@ -0,0 +1,52 @@ +"""Validation constants, regex patterns, and reusable Pydantic field types.""" + +import re +from typing import Annotated + +from pydantic import Field + +USERNAME_MIN_LENGTH = 3 +USERNAME_MAX_LENGTH = 32 +USERNAME_PATTERN = r"^[A-Za-z0-9_.-]{3,32}$" + +PIN_MIN_LENGTH = 4 +PIN_MAX_LENGTH = 8 +PIN_PATTERN = r"^\d{4,8}$" + +UUID_PATTERN = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +UUID_RE = re.compile(UUID_PATTERN, re.IGNORECASE) + +ROLE_PATTERN = r"^(student|teacher|admin)$" +ALLOWED_ROLES = frozenset({"student", "teacher", "admin"}) + +# Reusable Pydantic annotated field types +UsernameField = Annotated[ + str, + Field( + ..., + min_length=USERNAME_MIN_LENGTH, + max_length=USERNAME_MAX_LENGTH, + pattern=USERNAME_PATTERN, + examples=["student1"], + ), +] + +PinField = Annotated[ + str, + Field( + ..., + min_length=PIN_MIN_LENGTH, + max_length=PIN_MAX_LENGTH, + pattern=PIN_PATTERN, + examples=["1234"], + ), +] + +RoleField = Annotated[ + str, + Field( + "student", + pattern=ROLE_PATTERN, + examples=["student"], + ), +] diff --git a/backend/tests/api/__init__.py b/backend/tests/api/__init__.py new file mode 100644 index 0000000..19094c6 --- /dev/null +++ b/backend/tests/api/__init__.py @@ -0,0 +1 @@ +"""API test package.""" diff --git a/backend/tests/api/auth/__init__.py b/backend/tests/api/auth/__init__.py new file mode 100644 index 0000000..52f6bee --- /dev/null +++ b/backend/tests/api/auth/__init__.py @@ -0,0 +1 @@ +"""Auth API tests.""" diff --git a/backend/tests/test_auth.py b/backend/tests/api/auth/test_login.py similarity index 79% rename from backend/tests/test_auth.py rename to backend/tests/api/auth/test_login.py index 2b32dd0..c82ff2f 100644 --- a/backend/tests/test_auth.py +++ b/backend/tests/api/auth/test_login.py @@ -3,57 +3,10 @@ from fastapi.testclient import TestClient from src.db.database import get_db_connection -from src.security.auth import hash_pin, verify_pin +from src.security.auth import hash_pin from src.security.rate_limit import LOCKOUT_DURATION_SECONDS -def test_hash_pin_generates_bcrypt_hash(): - """ - Test that hash_pin produces a valid bcrypt hash starting with $2b$. - """ - pin = "1234" - hashed = hash_pin(pin) - assert hashed != pin - assert hashed.startswith(("$2b$", "$2a$")) - - -def test_hash_pin_generates_unique_salts(): - """ - Test that hashing the same PIN twice yields different hash strings due to salt. - """ - pin = "1234" - hash1 = hash_pin(pin) - hash2 = hash_pin(pin) - assert hash1 != hash2 - assert verify_pin(pin, hash1) is True - assert verify_pin(pin, hash2) is True - - -def test_verify_pin_success(): - """ - Test verify_pin returns True for valid PIN and hash. - """ - pin = "5678" - hashed = hash_pin(pin) - assert verify_pin(pin, hashed) is True - - -def test_verify_pin_failure(): - """ - Test verify_pin returns False for incorrect PIN. - """ - pin = "5678" - hashed = hash_pin(pin) - assert verify_pin("0000", hashed) is False - - -def test_verify_pin_invalid_hash_format(): - """ - Test verify_pin handles invalid hash string format gracefully without crashing. - """ - assert verify_pin("1234", "not_a_valid_bcrypt_hash") is False - - def test_login_success(seeded_db, client: TestClient): """ Test successful student login returns 200 OK and session_id. @@ -224,3 +177,52 @@ def test_login_rejects_wrong_pin_length(seeded_db, client: TestClient): long_pin = client.post("/login", json={"username": "student1", "pin": "1" * 9}) assert short.status_code == 422 assert long_pin.status_code == 422 + + +def test_login_response_surfaces_must_change_pin(temp_db, client: TestClient): + """ + Login must return must_change_pin=True when flagged in the database, + and must_change_pin=False by default. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, must_change_pin) VALUES (?, ?, 1)", + ("rotate_user", hashed), + ) + cursor.execute( + "INSERT INTO users (username, hashed_pin, must_change_pin) VALUES (?, ?, 0)", + ("normal_user", hashed), + ) + conn.commit() + + # User with rotation pending + res1 = client.post("/login", json={"username": "rotate_user", "pin": "1234"}) + assert res1.status_code == 200 + assert res1.json()["must_change_pin"] is True + + # User with no rotation pending + res2 = client.post("/login", json={"username": "normal_user", "pin": "1234"}) + assert res2.status_code == 200 + assert res2.json()["must_change_pin"] is False + + +def test_login_soft_deleted_user_returns_401(temp_db, client: TestClient): + """ + Soft-deleted users must receive generic 401 on login without existence leak. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, deleted_at) VALUES (?, ?, '2026-08-23 00:00:00')", + ("deleted_student", hashed), + ) + conn.commit() + + response = client.post( + "/login", json={"username": "deleted_student", "pin": "1234"} + ) + assert response.status_code == 401 + assert response.json()["detail"] == "Invalid username or PIN." diff --git a/backend/tests/api/auth/test_logout.py b/backend/tests/api/auth/test_logout.py new file mode 100644 index 0000000..7bfe15a --- /dev/null +++ b/backend/tests/api/auth/test_logout.py @@ -0,0 +1,54 @@ +from fastapi.testclient import TestClient + +from src.db.database import get_db_connection + + +def test_logout_success_deactivates_session(seeded_db, client: TestClient): + """ + POST /logout must deactivate the active session in SQLite. + """ + db_path, _ = seeded_db + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + session_id = login_res.json()["session_id"] + + # Logout + logout_res = client.post( + "/logout", headers={"Authorization": f"Bearer {session_id}"} + ) + assert logout_res.status_code == 200 + assert logout_res.json()["detail"] == "Logged out." + + # Verify session is deactivated in DB + conn = get_db_connection(db_path) + cursor = conn.cursor() + cursor.execute("SELECT is_active FROM sessions WHERE id = ?", (session_id,)) + row = cursor.fetchone() + assert row is not None + assert row["is_active"] == 0 + conn.close() + + +def test_logout_idempotent_and_subsequent_request_fails(seeded_db, client: TestClient): + """ + Calling logout deactivates session; a second attempt with that dead token + fails at the Bearer dependency (401). + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + session_id = login_res.json()["session_id"] + + # First logout succeeds + res1 = client.post("/logout", headers={"Authorization": f"Bearer {session_id}"}) + assert res1.status_code == 200 + + # Second call with the same token fails with 401 because session is no longer active + res2 = client.post("/logout", headers={"Authorization": f"Bearer {session_id}"}) + assert res2.status_code == 401 + + +def test_logout_unauthenticated_returns_401(client: TestClient): + """ + Calling /logout without Bearer header returns 401. + """ + res = client.post("/logout") + assert res.status_code == 401 diff --git a/backend/tests/api/staff/__init__.py b/backend/tests/api/staff/__init__.py new file mode 100644 index 0000000..ac46134 --- /dev/null +++ b/backend/tests/api/staff/__init__.py @@ -0,0 +1 @@ +"""Staff API tests.""" diff --git a/backend/tests/api/staff/test_audit.py b/backend/tests/api/staff/test_audit.py new file mode 100644 index 0000000..f93021d --- /dev/null +++ b/backend/tests/api/staff/test_audit.py @@ -0,0 +1,183 @@ +from fastapi.testclient import TestClient + +from tests.conftest import auth_headers, get_user_id + + +def test_read_audit_logs_admin_success(staff_db, client: TestClient): + """ + Admin can read audit logs (200 OK). + """ + admin_headers = auth_headers(client, "admin1", "1234") + res = client.get("/audit-logs", headers=admin_headers) + assert res.status_code == 200 + data = res.json() + assert "logs" in data + assert isinstance(data["logs"], list) + + +def test_read_audit_logs_teacher_forbidden(staff_db, client: TestClient): + """ + Teachers are forbidden from viewing audit logs (403 Forbidden). + """ + teacher_headers = auth_headers(client, "teacher1", "1234") + res = client.get("/audit-logs", headers=teacher_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Insufficient permissions." + + +def test_read_audit_logs_student_forbidden(staff_db, client: TestClient): + """ + Students are forbidden from viewing audit logs (403 Forbidden). + """ + student_headers = auth_headers(client, "student1", "1234") + res = client.get("/audit-logs", headers=student_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Insufficient permissions." + + +def test_read_audit_logs_unauthenticated(client: TestClient): + """ + Unauthenticated caller receives 401 Unauthorized. + """ + res = client.get("/audit-logs") + assert res.status_code == 401 + + +def test_read_audit_logs_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + Admin caller with must_change_pin=1 is blocked (403) by the rotation gate. + """ + from src.security.auth import hash_pin + + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, must_change_pin) VALUES ('admin_rot', ?, 'admin', 1)", + (hashed,), + ) + conn.commit() + + login_res = client.post("/login", json={"username": "admin_rot", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + res = client.get("/audit-logs", headers=headers) + assert res.status_code == 403 + assert res.json()["detail"] == "PIN change required." + + +def test_audit_trail_captures_all_lifecycle_actions(staff_db, client: TestClient): + """ + Comprehensive verification: All 7 lifecycle mutations record audit entries. + 1. signup + 2. user_created + 3. pin_reset + 4. username_changed + 5. pin_changed + 6. account_deleted + 7. account_recovered + """ + _, conn = staff_db + admin_id = get_user_id(conn, "admin1") + teacher_id = get_user_id(conn, "teacher1") + admin_headers = auth_headers(client, "admin1", "1234") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # 1. Signup + signup_res = client.post( + "/signup", json={"username": "audit_student", "pin": "1234"} + ) + assert signup_res.status_code == 201 + audit_student_id = get_user_id(conn, "audit_student") + + # 2. Staff user creation + create_res = client.post( + "/users", + headers=admin_headers, + json={"username": "audit_staff_user", "pin": "1234", "role": "student"}, + ) + assert create_res.status_code == 201 + audit_staff_user_id = get_user_id(conn, "audit_staff_user") + + # 3. Staff reset PIN + reset_res = client.post( + f"/users/{audit_staff_user_id}/reset-pin", + headers=teacher_headers, + ) + assert reset_res.status_code == 200 + + # 4. Username changed + stud_login = client.post( + "/login", json={"username": "audit_student", "pin": "1234"} + ) + stud_token = stud_login.json()["session_id"] + change_user_res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {stud_token}"}, + json={"current_pin": "1234", "new_username": "audit_student_renamed"}, + ) + assert change_user_res.status_code == 200 + + # 5. PIN changed + renamed_login = client.post( + "/login", json={"username": "audit_student_renamed", "pin": "1234"} + ) + renamed_token = renamed_login.json()["session_id"] + change_pin_res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {renamed_token}"}, + json={"current_pin": "1234", "new_pin": "9876"}, + ) + assert change_pin_res.status_code == 200 + + # 6. Account deleted + del_res = client.delete( + f"/users/{audit_staff_user_id}", + headers=admin_headers, + ) + assert del_res.status_code == 200 + + # 7. Account recovered + rec_res = client.post( + f"/users/{audit_staff_user_id}/recover", + headers=admin_headers, + json={"username": "audit_staff_recovered"}, + ) + assert rec_res.status_code == 200 + + # Read audit logs as admin + audit_res = client.get("/audit-logs", headers=admin_headers) + assert audit_res.status_code == 200 + logs = audit_res.json()["logs"] + + # Map actions from logs + action_map = {log["action"]: log for log in logs} + + assert "signup" in action_map + assert action_map["signup"]["actor_user_id"] is None + assert action_map["signup"]["target_user_id"] == audit_student_id + + assert "user_created" in action_map + assert action_map["user_created"]["actor_user_id"] == admin_id + assert action_map["user_created"]["target_user_id"] == audit_staff_user_id + + assert "pin_reset" in action_map + assert action_map["pin_reset"]["actor_user_id"] == teacher_id + assert action_map["pin_reset"]["target_user_id"] == audit_staff_user_id + + assert "username_changed" in action_map + assert action_map["username_changed"]["actor_user_id"] == audit_student_id + assert action_map["username_changed"]["target_user_id"] == audit_student_id + + assert "pin_changed" in action_map + assert action_map["pin_changed"]["actor_user_id"] == audit_student_id + assert action_map["pin_changed"]["target_user_id"] == audit_student_id + + assert "account_deleted" in action_map + assert action_map["account_deleted"]["actor_user_id"] == admin_id + assert action_map["account_deleted"]["target_user_id"] == audit_staff_user_id + + assert "account_recovered" in action_map + assert action_map["account_recovered"]["actor_user_id"] == admin_id + assert action_map["account_recovered"]["target_user_id"] == audit_staff_user_id diff --git a/backend/tests/api/staff/test_delete.py b/backend/tests/api/staff/test_delete.py new file mode 100644 index 0000000..01d3684 --- /dev/null +++ b/backend/tests/api/staff/test_delete.py @@ -0,0 +1,272 @@ +import uuid + +from fastapi.testclient import TestClient + +from src.security.auth import hash_pin, verify_pin +from tests.conftest import auth_headers, get_user_id + + +def test_teacher_deletes_student_success(staff_db, client: TestClient): + """ + Teacher can soft-delete a student. + - Returns 200 OK with 'Account deleted.'. + - Database row has deleted_at set, former_username populated, username anonymized. + - Old sessions are invalidated. + - Login with old username/pin fails. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + + # Log student1 in to create an active session + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + student_token = login_res.json()["session_id"] + student_headers = {"Authorization": f"Bearer {student_token}"} + + # Verify session works + assert client.get("/users/me", headers=student_headers).status_code == 200 + + # Teacher deletes student1 + teacher_headers = auth_headers(client, "teacher1", "1234") + del_res = client.delete(f"/users/{student_id}", headers=teacher_headers) + assert del_res.status_code == 200 + assert del_res.json() == {"detail": "Account deleted."} + + # Active session is immediately invalidated (401) + assert client.get("/users/me", headers=student_headers).status_code == 401 + + # Login fails + assert ( + client.post("/login", json={"username": "student1", "pin": "1234"}).status_code + == 401 + ) + + # Inspect DB record + cursor = conn.cursor() + cursor.execute( + "SELECT username, former_username, hashed_pin, deleted_at FROM users WHERE id = ?", + (student_id,), + ) + row = cursor.fetchone() + assert row["deleted_at"] is not None + assert row["former_username"] == "student1" + assert row["username"].startswith(f"deleted_user_{student_id}_") + # Verify the anonymized hash fails any PIN check cleanly without exceptions + assert verify_pin("1234", row["hashed_pin"]) is False + + +def test_teacher_deletes_another_teacher_success(staff_db, client: TestClient): + """ + Under the uniform staff matrix, a teacher can soft-delete another teacher. + """ + _, conn = staff_db + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Create teacher2 + client.post( + "/users", + headers=teacher_headers, + json={"username": "teacher2", "pin": "1234", "role": "teacher"}, + ) + teacher2_id = get_user_id(conn, "teacher2") + + del_res = client.delete(f"/users/{teacher2_id}", headers=teacher_headers) + assert del_res.status_code == 200 + assert del_res.json() == {"detail": "Account deleted."} + + +def test_teacher_deleting_admin_returns_403(staff_db, client: TestClient): + """ + Teacher cannot delete an admin account (403 Forbidden). + """ + _, conn = staff_db + admin_id = get_user_id(conn, "admin1") + teacher_headers = auth_headers(client, "teacher1", "1234") + + res = client.delete(f"/users/{admin_id}", headers=teacher_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Only admins may delete admin accounts." + + +def test_admin_deletes_anyone_success(staff_db, client: TestClient): + """ + Admin can delete students, teachers, and other admins (when not the last admin). + """ + _, conn = staff_db + admin_headers = auth_headers(client, "admin1", "1234") + + # Admin deletes student + student_id = get_user_id(conn, "student1") + assert ( + client.delete(f"/users/{student_id}", headers=admin_headers).status_code == 200 + ) + + # Admin deletes teacher + teacher_id = get_user_id(conn, "teacher1") + assert ( + client.delete(f"/users/{teacher_id}", headers=admin_headers).status_code == 200 + ) + + # Admin creates second admin and deletes them + client.post( + "/users", + headers=admin_headers, + json={"username": "admin2", "pin": "1234", "role": "admin"}, + ) + admin2_id = get_user_id(conn, "admin2") + assert ( + client.delete(f"/users/{admin2_id}", headers=admin_headers).status_code == 200 + ) + + +def test_admin_cannot_delete_last_remaining_admin(staff_db, client: TestClient): + """ + Last-admin guard: Appliance must never lose its final administrator (409 Conflict). + """ + _, conn = staff_db + admin_id = get_user_id(conn, "admin1") + admin_headers = auth_headers(client, "admin1", "1234") + + # Only admin1 exists + res = client.delete(f"/users/{admin_id}", headers=admin_headers) + assert res.status_code == 409 + assert res.json()["detail"] == "Cannot delete the last remaining admin account." + + +def test_delete_user_not_found(staff_db, client: TestClient): + """ + Deleting a non-existent user returns 404 Not Found. + """ + admin_headers = auth_headers(client, "admin1", "1234") + res = client.delete("/users/99999", headers=admin_headers) + assert res.status_code == 404 + assert res.json()["detail"] == "User not found." + + +def test_delete_already_deleted_user_returns_404(staff_db, client: TestClient): + """ + Deleting an already soft-deleted user returns 404 Not Found. + """ + _, conn = staff_db + student1_id = get_user_id(conn, "student1") + admin_headers = auth_headers(client, "admin1", "1234") + + # First delete succeeds + assert ( + client.delete(f"/users/{student1_id}", headers=admin_headers).status_code == 200 + ) + + # Second delete returns 404 + res = client.delete(f"/users/{student1_id}", headers=admin_headers) + assert res.status_code == 404 + assert res.json()["detail"] == "User not found." + + +def test_delete_user_forbidden_for_students(staff_db, client: TestClient): + """ + Students cannot delete accounts (403 Forbidden). + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + student_headers = auth_headers(client, "student1", "1234") + + res = client.delete(f"/users/{student2_id}", headers=student_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Insufficient permissions." + + +def test_delete_user_unauthenticated(client: TestClient): + """ + Unauthenticated caller receives 401 Unauthorized. + """ + assert client.delete("/users/1").status_code == 401 + + +def test_delete_user_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + Staff caller with must_change_pin=1 is blocked (403) by the rotation gate. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, must_change_pin) VALUES ('teacher_rot', ?, 'teacher', 1)", + (hashed,), + ) + cursor.execute( + "INSERT INTO users (username, hashed_pin, role) VALUES ('target_stud', ?, 'student')", + (hashed,), + ) + conn.commit() + + cursor.execute("SELECT id FROM users WHERE username = 'target_stud'") + target_id = cursor.fetchone()["id"] + + login_res = client.post("/login", json={"username": "teacher_rot", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + res = client.delete(f"/users/{target_id}", headers=headers) + assert res.status_code == 403 + assert res.json()["detail"] == "PIN change required." + + +def test_soft_delete_preserves_telemetry_turn_logs(staff_db, client: TestClient): + """ + Telemetry preservation: turn_logs records remain intact and joinable to the soft-deleted user row. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + session_id = str(uuid.uuid4()) + + cursor = conn.cursor() + cursor.execute( + "INSERT INTO sessions (id, user_id, is_active) VALUES (?, ?, 1)", + (session_id, student_id), + ) + cursor.execute( + "INSERT INTO turn_logs (session_id, user_input, final_response) VALUES (?, ?, ?)", + (session_id, "2 + 2", "4"), + ) + conn.commit() + + # Teacher deletes student1 + teacher_headers = auth_headers(client, "teacher1", "1234") + del_res = client.delete(f"/users/{student_id}", headers=teacher_headers) + assert del_res.status_code == 200 + + # Verify telemetry is intact + cursor.execute( + "SELECT t.id, t.user_input, u.former_username, u.deleted_at " + "FROM turn_logs t " + "JOIN sessions s ON s.id = t.session_id " + "JOIN users u ON u.id = s.user_id " + "WHERE u.id = ?", + (student_id,), + ) + row = cursor.fetchone() + assert row is not None + assert row["user_input"] == "2 + 2" + assert row["former_username"] == "student1" + assert row["deleted_at"] is not None + + +def test_original_username_reusable_after_deletion(staff_db, client: TestClient): + """ + Once an account is soft-deleted and its username anonymized, the original username + is immediately available for new registration / creation. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Delete student1 + assert ( + client.delete(f"/users/{student_id}", headers=teacher_headers).status_code + == 200 + ) + + # Re-register with 'student1' + signup_res = client.post("/signup", json={"username": "student1", "pin": "5678"}) + assert signup_res.status_code == 201 + assert signup_res.json()["username"] == "student1" diff --git a/backend/tests/api/staff/test_recover.py b/backend/tests/api/staff/test_recover.py new file mode 100644 index 0000000..9372895 --- /dev/null +++ b/backend/tests/api/staff/test_recover.py @@ -0,0 +1,295 @@ +import logging + +from fastapi.testclient import TestClient + +from src.security.auth import hash_pin +from tests.conftest import auth_headers, get_user_id + + +def test_teacher_recovers_student_success(staff_db, client: TestClient): + """ + Teacher can recover a soft-deleted student account with a new username. + - Returns 200 OK with RecoverUserResponse. + - Database row has deleted_at cleared and must_change_pin=1. + - Student can log in with temporary PIN and is prompted for rotation. + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Soft-delete student2 + assert ( + client.delete(f"/users/{student2_id}", headers=teacher_headers).status_code + == 200 + ) + + # Recover student2 under new username 'student2_restored' + rec_res = client.post( + f"/users/{student2_id}/recover", + headers=teacher_headers, + json={"username": "student2_restored"}, + ) + assert rec_res.status_code == 200 + data = rec_res.json() + assert data["username"] == "student2_restored" + temp_pin = data["temporary_pin"] + assert len(temp_pin) == 6 + assert temp_pin.isdigit() + assert data["detail"] == "Account recovered. User must set a new PIN on next login." + + # Inspect DB record + cursor = conn.cursor() + cursor.execute( + "SELECT username, deleted_at, must_change_pin FROM users WHERE id = ?", + (student2_id,), + ) + row = cursor.fetchone() + assert row["username"] == "student2_restored" + assert row["deleted_at"] is None + assert row["must_change_pin"] == 1 + + # Login with temp PIN works and flags must_change_pin=True + login_res = client.post( + "/login", json={"username": "student2_restored", "pin": temp_pin} + ) + assert login_res.status_code == 200 + assert login_res.json()["must_change_pin"] is True + + +def test_teacher_recovers_another_teacher_success(staff_db, client: TestClient): + """ + Under the uniform staff matrix, a teacher can recover another teacher's account. + """ + _, conn = staff_db + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Create teacher2 + client.post( + "/users", + headers=teacher_headers, + json={"username": "teacher2", "pin": "1234", "role": "teacher"}, + ) + teacher2_id = get_user_id(conn, "teacher2") + + # Delete teacher2 + client.delete(f"/users/{teacher2_id}", headers=teacher_headers) + + # Recover teacher2 + rec_res = client.post( + f"/users/{teacher2_id}/recover", + headers=teacher_headers, + json={"username": "teacher2_new"}, + ) + assert rec_res.status_code == 200 + assert rec_res.json()["username"] == "teacher2_new" + + +def test_teacher_recovering_admin_returns_403(staff_db, client: TestClient): + """ + Teacher cannot recover an admin account (403 Forbidden). + """ + _, conn = staff_db + admin_headers = auth_headers(client, "admin1", "1234") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Create and delete admin2 + client.post( + "/users", + headers=admin_headers, + json={"username": "admin2", "pin": "1234", "role": "admin"}, + ) + admin2_id = get_user_id(conn, "admin2") + client.delete(f"/users/{admin2_id}", headers=admin_headers) + + # Teacher attempts recovery + rec_res = client.post( + f"/users/{admin2_id}/recover", + headers=teacher_headers, + json={"username": "admin2_restored"}, + ) + assert rec_res.status_code == 403 + assert rec_res.json()["detail"] == "Only admins may recover admin accounts." + + +def test_admin_recovers_any_account(staff_db, client: TestClient): + """ + Admin can recover student, teacher, and admin accounts. + """ + _, conn = staff_db + admin_headers = auth_headers(client, "admin1", "1234") + + # Create and delete admin2 + client.post( + "/users", + headers=admin_headers, + json={"username": "admin2", "pin": "1234", "role": "admin"}, + ) + admin2_id = get_user_id(conn, "admin2") + client.delete(f"/users/{admin2_id}", headers=admin_headers) + + # Admin recovers admin2 + rec_res = client.post( + f"/users/{admin2_id}/recover", + headers=admin_headers, + json={"username": "admin2_restored"}, + ) + assert rec_res.status_code == 200 + assert rec_res.json()["username"] == "admin2_restored" + + +def test_recover_username_conflict_returns_409(staff_db, client: TestClient): + """ + Recovering an account using an already-taken username returns 409 Conflict. + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Soft-delete student2 + client.delete(f"/users/{student2_id}", headers=teacher_headers) + + # Attempt to recover with taken username 'student1' + rec_res = client.post( + f"/users/{student2_id}/recover", + headers=teacher_headers, + json={"username": "student1"}, + ) + assert rec_res.status_code == 409 + assert ( + rec_res.json()["detail"] + == "Username already taken. Choose another for this account." + ) + + +def test_recover_target_not_found_or_not_deleted(staff_db, client: TestClient): + """ + Recovering non-existent or currently-active user returns 404 Not Found. + """ + _, conn = staff_db + student1_id = get_user_id(conn, "student1") + admin_headers = auth_headers(client, "admin1", "1234") + + # Non-existent ID + res_nonexistent = client.post( + "/users/99999/recover", + headers=admin_headers, + json={"username": "some_user"}, + ) + assert res_nonexistent.status_code == 404 + assert res_nonexistent.json()["detail"] == "Deleted user not found." + + # Active user (not deleted) + res_active = client.post( + f"/users/{student1_id}/recover", + headers=admin_headers, + json={"username": "some_user"}, + ) + assert res_active.status_code == 404 + assert res_active.json()["detail"] == "Deleted user not found." + + +def test_recover_validation_errors(staff_db, client: TestClient): + """ + POST /users/{id}/recover validates username format (422 Unprocessable Entity). + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + teacher_headers = auth_headers(client, "teacher1", "1234") + client.delete(f"/users/{student2_id}", headers=teacher_headers) + + # Invalid username with space + res = client.post( + f"/users/{student2_id}/recover", + headers=teacher_headers, + json={"username": "bad name"}, + ) + assert res.status_code == 422 + + +def test_recover_user_forbidden_for_students(staff_db, client: TestClient): + """ + Students cannot recover accounts (403 Forbidden). + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + student_headers = auth_headers(client, "student1", "1234") + + res = client.post( + f"/users/{student2_id}/recover", + headers=student_headers, + json={"username": "new_student2"}, + ) + assert res.status_code == 403 + assert res.json()["detail"] == "Insufficient permissions." + + +def test_recover_user_unauthenticated(client: TestClient): + """ + Unauthenticated caller receives 401 Unauthorized. + """ + assert ( + client.post("/users/1/recover", json={"username": "new_student"}).status_code + == 401 + ) + + +def test_recover_user_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + Staff caller with must_change_pin=1 is blocked (403) by the rotation gate. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, must_change_pin) VALUES ('teacher_rot', ?, 'teacher', 1)", + (hashed,), + ) + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, deleted_at) VALUES ('deleted_stud', ?, 'student', CURRENT_TIMESTAMP)", + (hashed,), + ) + conn.commit() + + cursor.execute("SELECT id FROM users WHERE username = 'deleted_stud'") + deleted_id = cursor.fetchone()["id"] + + login_res = client.post("/login", json={"username": "teacher_rot", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + res = client.post( + f"/users/{deleted_id}/recover", + headers=headers, + json={"username": "restored_stud"}, + ) + assert res.status_code == 403 + assert res.json()["detail"] == "PIN change required." + + +def test_recover_temporary_pin_never_logged(staff_db, client: TestClient, caplog): + """ + SECURITY PROOF: + The temporary PIN generated during account recovery must never appear in any log output. + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Soft-delete student2 + client.delete(f"/users/{student2_id}", headers=teacher_headers) + + # Recover student2 under caplog + with caplog.at_level(logging.DEBUG): + rec_res = client.post( + f"/users/{student2_id}/recover", + headers=teacher_headers, + json={"username": "student2_restored"}, + ) + + assert rec_res.status_code == 200 + temp_pin = rec_res.json()["temporary_pin"] + + for record in caplog.records: + assert temp_pin not in record.getMessage(), ( + f"Security violation: Temporary PIN leaked in log message: '{record.getMessage()}'" + ) diff --git a/backend/tests/api/staff/test_reset_pin.py b/backend/tests/api/staff/test_reset_pin.py new file mode 100644 index 0000000..5275904 --- /dev/null +++ b/backend/tests/api/staff/test_reset_pin.py @@ -0,0 +1,231 @@ +import logging + +from fastapi.testclient import TestClient + +from tests.conftest import auth_headers, get_user_id + + +def test_teacher_resets_student_pin(staff_db, client: TestClient): + """ + Teacher can reset a student's PIN. + Returns 200 OK with a 6-digit temporary PIN. + The database flags must_change_pin=1, old PIN fails, and temp PIN succeeds. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + teacher_headers = auth_headers(client, "teacher1", "1234") + + res = client.post(f"/users/{student_id}/reset-pin", headers=teacher_headers) + assert res.status_code == 200 + data = res.json() + assert data["username"] == "student1" + temp_pin = data["temporary_pin"] + assert len(temp_pin) == 6 + assert temp_pin.isdigit() + + # Verify must_change_pin in DB + cursor = conn.cursor() + cursor.execute("SELECT must_change_pin FROM users WHERE id = ?", (student_id,)) + assert cursor.fetchone()["must_change_pin"] == 1 + + # Old PIN must fail + old_login = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert old_login.status_code == 401 + + # Temp PIN must succeed and indicate must_change_pin=True + new_login = client.post("/login", json={"username": "student1", "pin": temp_pin}) + assert new_login.status_code == 200 + assert new_login.json()["must_change_pin"] is True + + +def test_teacher_resets_target_invalidates_all_target_sessions( + staff_db, client: TestClient +): + """ + Resetting a user's PIN immediately invalidates all active sessions for that user. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + + # Student logs in + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + student_token = login_res.json()["session_id"] + student_headers = {"Authorization": f"Bearer {student_token}"} + + # Verify student session works + me_res = client.get("/users/me", headers=student_headers) + assert me_res.status_code == 200 + + # Teacher resets student PIN + teacher_headers = auth_headers(client, "teacher1", "1234") + reset_res = client.post(f"/users/{student_id}/reset-pin", headers=teacher_headers) + assert reset_res.status_code == 200 + + # Student's prior session is now invalid (401 Unauthorized) + me_after = client.get("/users/me", headers=student_headers) + assert me_after.status_code == 401 + + +def test_teacher_resetting_admin_returns_403(staff_db, client: TestClient): + """ + A teacher cannot reset an admin's PIN (403 Forbidden). + """ + _, conn = staff_db + admin_id = get_user_id(conn, "admin1") + teacher_headers = auth_headers(client, "teacher1", "1234") + + res = client.post(f"/users/{admin_id}/reset-pin", headers=teacher_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Only admins may reset admin PINs." + + +def test_teacher_resets_another_teacher(staff_db, client: TestClient): + """ + Under the uniform staff matrix, a teacher CAN reset another teacher's PIN. + """ + _, conn = staff_db + teacher_headers = auth_headers(client, "teacher1", "1234") + + # Create teacher2 + create_res = client.post( + "/users", + headers=teacher_headers, + json={"username": "teacher2", "pin": "1234", "role": "teacher"}, + ) + assert create_res.status_code == 201 + teacher2_id = get_user_id(conn, "teacher2") + + reset_res = client.post(f"/users/{teacher2_id}/reset-pin", headers=teacher_headers) + assert reset_res.status_code == 200 + assert reset_res.json()["username"] == "teacher2" + assert len(reset_res.json()["temporary_pin"]) == 6 + + +def test_admin_resets_any_account(staff_db, client: TestClient): + """ + An admin can reset any account: student, teacher, or another admin. + """ + _, conn = staff_db + admin_headers = auth_headers(client, "admin1", "1234") + + # Admin resets student + student_id = get_user_id(conn, "student1") + res_s = client.post(f"/users/{student_id}/reset-pin", headers=admin_headers) + assert res_s.status_code == 200 + + # Admin resets teacher + teacher_id = get_user_id(conn, "teacher1") + res_t = client.post(f"/users/{teacher_id}/reset-pin", headers=admin_headers) + assert res_t.status_code == 200 + + # Admin creates and resets another admin + client.post( + "/users", + headers=admin_headers, + json={"username": "admin2", "pin": "1234", "role": "admin"}, + ) + admin2_id = get_user_id(conn, "admin2") + res_a = client.post(f"/users/{admin2_id}/reset-pin", headers=admin_headers) + assert res_a.status_code == 200 + + +def test_reset_pin_target_not_found(staff_db, client: TestClient): + """ + Resetting a non-existent user returns 404 Not Found. + """ + admin_headers = auth_headers(client, "admin1", "1234") + res = client.post("/users/99999/reset-pin", headers=admin_headers) + assert res.status_code == 404 + assert res.json()["detail"] == "User not found." + + +def test_reset_pin_target_soft_deleted(staff_db, client: TestClient): + """ + Resetting a soft-deleted user returns 404 Not Found. + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + cursor = conn.cursor() + cursor.execute( + "UPDATE users SET deleted_at = CURRENT_TIMESTAMP WHERE id = ?", + (student2_id,), + ) + conn.commit() + + admin_headers = auth_headers(client, "admin1", "1234") + res = client.post(f"/users/{student2_id}/reset-pin", headers=admin_headers) + assert res.status_code == 404 + assert res.json()["detail"] == "User not found." + + +def test_reset_pin_forbidden_for_students(staff_db, client: TestClient): + """ + Student role is forbidden (403) from resetting PINs. + """ + _, conn = staff_db + student2_id = get_user_id(conn, "student2") + student_headers = auth_headers(client, "student1", "1234") + + res = client.post(f"/users/{student2_id}/reset-pin", headers=student_headers) + assert res.status_code == 403 + assert res.json()["detail"] == "Insufficient permissions." + + +def test_reset_pin_unauthenticated(client: TestClient): + """ + Unauthenticated caller receives 401 Unauthorized. + """ + res = client.post("/users/1/reset-pin") + assert res.status_code == 401 + + +def test_reset_pin_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + Staff caller with must_change_pin=1 is blocked (403) by the rotation gate. + """ + from src.security.auth import hash_pin + + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, must_change_pin) VALUES ('teacher_rot', ?, 'teacher', 1)", + (hashed,), + ) + cursor.execute( + "INSERT INTO users (username, hashed_pin, role) VALUES ('target_stud', ?, 'student')", + (hashed,), + ) + conn.commit() + + cursor.execute("SELECT id FROM users WHERE username = 'target_stud'") + target_id = cursor.fetchone()["id"] + + login_res = client.post("/login", json={"username": "teacher_rot", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + res = client.post(f"/users/{target_id}/reset-pin", headers=headers) + assert res.status_code == 403 + assert res.json()["detail"] == "PIN change required." + + +def test_reset_pin_temporary_pin_never_logged(staff_db, client: TestClient, caplog): + """ + SECURITY PROOF: + The temporary PIN generated for a reset must never appear in any log output. + """ + _, conn = staff_db + student_id = get_user_id(conn, "student1") + teacher_headers = auth_headers(client, "teacher1", "1234") + + with caplog.at_level(logging.DEBUG): + res = client.post(f"/users/{student_id}/reset-pin", headers=teacher_headers) + + assert res.status_code == 200 + temp_pin = res.json()["temporary_pin"] + + for record in caplog.records: + assert temp_pin not in record.getMessage(), ( + f"Security violation: Temporary PIN leaked in log message: '{record.getMessage()}'" + ) diff --git a/backend/tests/api/staff/test_users.py b/backend/tests/api/staff/test_users.py new file mode 100644 index 0000000..b83da2e --- /dev/null +++ b/backend/tests/api/staff/test_users.py @@ -0,0 +1,243 @@ +from fastapi.testclient import TestClient + +from tests.conftest import auth_headers + + +def test_list_users_active_only(staff_db, client: TestClient): + """ + GET /users lists active users and hides deleted accounts by default. + """ + _, conn = staff_db + cursor = conn.cursor() + # Soft-delete student2 + cursor.execute( + "UPDATE users SET deleted_at = CURRENT_TIMESTAMP, former_username = 'student2' WHERE username = 'student2'" + ) + conn.commit() + + headers = auth_headers(client, "teacher1", "1234") + res = client.get("/users", headers=headers) + assert res.status_code == 200 + usernames = [u["username"] for u in res.json()["users"]] + assert "student1" in usernames + assert "teacher1" in usernames + assert "admin1" in usernames + assert "student2" not in usernames + + +def test_list_users_include_deleted(staff_db, client: TestClient): + """ + GET /users?include_deleted=true lists deleted accounts with minimal recovery metadata. + """ + _, conn = staff_db + cursor = conn.cursor() + # Soft-delete student2 + cursor.execute( + "UPDATE users SET deleted_at = CURRENT_TIMESTAMP, former_username = 'student2' WHERE username = 'student2'" + ) + conn.commit() + + headers = auth_headers(client, "teacher1", "1234") + res = client.get("/users?include_deleted=true", headers=headers) + assert res.status_code == 200 + users = res.json()["users"] + assert len(users) == 1 + deleted = users[0] + assert deleted["former_username"] == "student2" + assert deleted["role"] == "student" + assert deleted["deleted_at"] is not None + assert "hashed_pin" not in deleted + + +def test_create_student_by_teacher(staff_db, client: TestClient): + """ + Teacher can create a student account (201 Created) with must_change_pin=0. + """ + _, conn = staff_db + headers = auth_headers(client, "teacher1", "1234") + res = client.post( + "/users", + headers=headers, + json={"username": "new_student", "pin": "1234", "role": "student"}, + ) + assert res.status_code == 201 + assert res.json() == {"username": "new_student", "role": "student"} + + cursor = conn.cursor() + cursor.execute( + "SELECT role, must_change_pin, deleted_at FROM users WHERE username = 'new_student'" + ) + row = cursor.fetchone() + assert row is not None + assert row["role"] == "student" + assert row["must_change_pin"] == 0 + assert row["deleted_at"] is None + + +def test_create_teacher_by_teacher(staff_db, client: TestClient): + """ + Teacher can create another teacher account (201 Created). + """ + headers = auth_headers(client, "teacher1", "1234") + res = client.post( + "/users", + headers=headers, + json={"username": "teacher2", "pin": "1234", "role": "teacher"}, + ) + assert res.status_code == 201 + assert res.json() == {"username": "teacher2", "role": "teacher"} + + +def test_teacher_creating_admin_returns_403(staff_db, client: TestClient): + """ + Teacher attempting to create an admin account is blocked with 403 Forbidden. + """ + headers = auth_headers(client, "teacher1", "1234") + res = client.post( + "/users", + headers=headers, + json={"username": "admin2", "pin": "1234", "role": "admin"}, + ) + assert res.status_code == 403 + assert res.json()["detail"] == "Only admins may create admin accounts." + + +def test_create_any_role_by_admin(staff_db, client: TestClient): + """ + Admin can create accounts of any role: student, teacher, and admin. + """ + headers = auth_headers(client, "admin1", "1234") + + # Admin creates student + res_student = client.post( + "/users", + headers=headers, + json={"username": "adm_student", "pin": "1234", "role": "student"}, + ) + assert res_student.status_code == 201 + + # Admin creates teacher + res_teacher = client.post( + "/users", + headers=headers, + json={"username": "adm_teacher", "pin": "1234", "role": "teacher"}, + ) + assert res_teacher.status_code == 201 + + # Admin creates admin + res_admin = client.post( + "/users", + headers=headers, + json={"username": "adm_admin", "pin": "1234", "role": "admin"}, + ) + assert res_admin.status_code == 201 + + +def test_create_user_duplicate_returns_409(staff_db, client: TestClient): + """ + POST /users with an existing username returns 409 Conflict. + """ + headers = auth_headers(client, "admin1", "1234") + res = client.post( + "/users", + headers=headers, + json={"username": "student1", "pin": "1234", "role": "student"}, + ) + assert res.status_code == 409 + assert res.json()["detail"] == "Username already taken." + + +def test_create_user_validation_errors(staff_db, client: TestClient): + """ + POST /users validates username, pin, and role fields (422 Unprocessable Entity). + """ + headers = auth_headers(client, "admin1", "1234") + + # Invalid username format + res = client.post( + "/users", + headers=headers, + json={"username": "bad name", "pin": "1234", "role": "student"}, + ) + assert res.status_code == 422 + + # Invalid PIN format + res = client.post( + "/users", + headers=headers, + json={"username": "valid_user", "pin": "abcd", "role": "student"}, + ) + assert res.status_code == 422 + + # Invalid role + res = client.post( + "/users", + headers=headers, + json={"username": "valid_user", "pin": "1234", "role": "superadmin"}, + ) + assert res.status_code == 422 + + +def test_staff_endpoints_forbidden_for_students(staff_db, client: TestClient): + """ + Student role is forbidden (403) from accessing GET /users and POST /users. + """ + headers = auth_headers(client, "student1", "1234") + + res_get = client.get("/users", headers=headers) + assert res_get.status_code == 403 + assert res_get.json()["detail"] == "Insufficient permissions." + + res_post = client.post( + "/users", + headers=headers, + json={"username": "another_user", "pin": "1234", "role": "student"}, + ) + assert res_post.status_code == 403 + assert res_post.json()["detail"] == "Insufficient permissions." + + +def test_staff_endpoints_unauthenticated(client: TestClient): + """ + Unauthenticated callers receive 401 Unauthorized. + """ + assert client.get("/users").status_code == 401 + assert ( + client.post( + "/users", + json={"username": "another_user", "pin": "1234", "role": "student"}, + ).status_code + == 401 + ) + + +def test_staff_endpoints_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + Staff caller with must_change_pin=1 is blocked (403) by the rotation gate. + """ + from src.security.auth import hash_pin + + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, must_change_pin) VALUES ('teacher_rot', ?, 'teacher', 1)", + (hashed,), + ) + conn.commit() + + login_res = client.post("/login", json={"username": "teacher_rot", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + res_get = client.get("/users", headers=headers) + assert res_get.status_code == 403 + assert res_get.json()["detail"] == "PIN change required." + + res_post = client.post( + "/users", + headers=headers, + json={"username": "some_user", "pin": "1234", "role": "student"}, + ) + assert res_post.status_code == 403 + assert res_post.json()["detail"] == "PIN change required." diff --git a/backend/tests/test_health.py b/backend/tests/api/test_health.py similarity index 100% rename from backend/tests/test_health.py rename to backend/tests/api/test_health.py diff --git a/backend/tests/api/users/__init__.py b/backend/tests/api/users/__init__.py new file mode 100644 index 0000000..10ce892 --- /dev/null +++ b/backend/tests/api/users/__init__.py @@ -0,0 +1 @@ +"""Users API tests.""" diff --git a/backend/tests/api/users/test_change_pin.py b/backend/tests/api/users/test_change_pin.py new file mode 100644 index 0000000..42c557c --- /dev/null +++ b/backend/tests/api/users/test_change_pin.py @@ -0,0 +1,204 @@ +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient + +from src.api.users.credentials import ChangePinRequest, _change_credential +from src.security.auth import hash_pin +from src.security.session import AuthContext + + +def test_change_pin_success(seeded_db, client: TestClient): + """ + PATCH /users/me/pin changes PIN, clears must_change_pin, and deactivates caller's session. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_pin": "5678"}, + ) + assert res.status_code == 200 + assert res.json()["detail"] == "Credentials updated. Please sign in again." + + # Caller's old session is now inactive + me_res = client.get("/users/me", headers={"Authorization": f"Bearer {token}"}) + assert me_res.status_code == 401 + + # Old PIN rejected + login_old = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_old.status_code == 401 + + # New PIN accepted with must_change_pin = False + login_new = client.post("/login", json={"username": "student1", "pin": "5678"}) + assert login_new.status_code == 200 + assert login_new.json()["must_change_pin"] is False + + +def test_change_pin_clears_must_change_pin(temp_db, client: TestClient): + """ + PATCH /users/me/pin clears must_change_pin flag during forced rotation. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, must_change_pin) VALUES (?, ?, 1)", + ("rotate_user", hashed), + ) + conn.commit() + + login_res = client.post("/login", json={"username": "rotate_user", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_pin": "9876"}, + ) + assert res.status_code == 200 + + login_new = client.post("/login", json={"username": "rotate_user", "pin": "9876"}) + assert login_new.status_code == 200 + assert login_new.json()["must_change_pin"] is False + + +def test_anti_oracle_pin_change_order(seeded_db, client: TestClient): + """ + Anti-oracle check ordering: wrong current_pin with new_pin == current_pin + MUST return 401 Unauthorized, never 422. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "9999", "new_pin": "9999"}, + ) + assert res.status_code == 401 + assert res.json()["detail"] == "Invalid current PIN." + + +def test_change_pin_same_pin_returns_422(seeded_db, client: TestClient): + """ + Correct current_pin but new_pin == current_pin returns 422 Unprocessable Entity. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_pin": "1234"}, + ) + assert res.status_code == 422 + assert "New PIN must differ" in res.json()["detail"] + + +def test_change_pin_wrong_current_pin_returns_401(seeded_db, client: TestClient): + """ + Wrong current_pin on PIN change returns 401. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "0000", "new_pin": "5678"}, + ) + assert res.status_code == 401 + assert res.json()["detail"] == "Invalid current PIN." + + +def test_credential_change_rate_limiting(seeded_db, client: TestClient): + """ + Repeated bad current PIN on credential change increments login rate limiter and triggers 429. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + # 4 bad attempts return 401 + for _ in range(4): + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "0000", "new_pin": "5678"}, + ) + assert res.status_code == 401 + + # 5th bad attempt triggers lockout + res_5th = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "0000", "new_pin": "5678"}, + ) + assert res_5th.status_code == 401 + + # 6th attempt is blocked by rate limiter with 429 + res_6th = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "0000", "new_pin": "5678"}, + ) + assert res_6th.status_code == 429 + assert "Too many failed login attempts" in res_6th.json()["detail"] + + # /login is also locked out uniformly + login_locked = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_locked.status_code == 429 + + +def test_change_credential_user_deleted_returns_401(temp_db, client: TestClient): + """ + If a user is soft-deleted after session creation, credential change returns 401 Invalid session. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin) VALUES (?, ?)", + ("temp_user", hashed), + ) + conn.commit() + + login_res = client.post("/login", json={"username": "temp_user", "pin": "1234"}) + token = login_res.json()["session_id"] + + # Soft-delete the user directly in SQLite + cursor.execute( + "UPDATE users SET deleted_at = CURRENT_TIMESTAMP WHERE username = 'temp_user'" + ) + conn.commit() + + res = client.patch( + "/users/me/pin", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_pin": "5678"}, + ) + assert res.status_code == 401 + assert res.json()["detail"] == "Invalid or expired session." + + +def test_change_credential_missing_user(temp_db): + """ + Defensive check: if caller's user record is missing in _change_credential, raise 401. + """ + ctx = AuthContext( + user_id=9999, + username="ghost", + role="student", + session_id="00000000-0000-0000-0000-000000000000", + must_change_pin=False, + ) + with pytest.raises(HTTPException) as exc: + _change_credential( + ctx, + ChangePinRequest(current_pin="1234", new_pin="5678"), + kind="pin", + ) + assert exc.value.status_code == 401 + assert exc.value.detail == "Invalid session." diff --git a/backend/tests/api/users/test_change_username.py b/backend/tests/api/users/test_change_username.py new file mode 100644 index 0000000..0aad038 --- /dev/null +++ b/backend/tests/api/users/test_change_username.py @@ -0,0 +1,114 @@ +from fastapi.testclient import TestClient + +from src.security.auth import hash_pin + + +def test_change_username_success(seeded_db, client: TestClient): + """ + PATCH /users/me/username updates username, deactivates session, and frees old username. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_username": "renamed_student"}, + ) + assert res.status_code == 200 + assert res.json()["detail"] == "Credentials updated. Please sign in again." + + # Caller session is now inactive + me_res = client.get("/users/me", headers={"Authorization": f"Bearer {token}"}) + assert me_res.status_code == 401 + + # Login with old username fails + login_old = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_old.status_code == 401 + + # Login with new username succeeds + login_new = client.post( + "/login", json={"username": "renamed_student", "pin": "1234"} + ) + assert login_new.status_code == 200 + + # Old username is immediately freed for self-signup reuse + signup_reuse = client.post("/signup", json={"username": "student1", "pin": "4321"}) + assert signup_reuse.status_code == 201 + + +def test_change_username_same_name_returns_422(seeded_db, client: TestClient): + """ + PATCH /users/me/username with same username returns 422 Unprocessable Entity. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_username": "student1"}, + ) + assert res.status_code == 422 + assert "New username must differ" in res.json()["detail"] + + +def test_change_username_duplicate_returns_409(staff_db, client: TestClient): + """ + PATCH /users/me/username with an already taken username returns 409 Conflict. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_username": "student2"}, + ) + assert res.status_code == 409 + assert res.json()["detail"] == "Username already taken." + + +def test_change_username_blocked_during_pending_rotation(temp_db, client: TestClient): + """ + PATCH /users/me/username is blocked (403 Forbidden) when user must rotate PIN. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, must_change_pin) VALUES (?, ?, 1)", + ("must_rotate_user", hashed), + ) + conn.commit() + + login_res = client.post( + "/login", json={"username": "must_rotate_user", "pin": "1234"} + ) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "1234", "new_username": "fresh_user"}, + ) + assert res.status_code == 403 + assert res.json()["detail"] == "PIN change required." + + +def test_anti_oracle_username_change_order(seeded_db, client: TestClient): + """ + Anti-oracle check ordering on username change: wrong current PIN with same username + MUST return 401 Unauthorized, never 422. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + + res = client.patch( + "/users/me/username", + headers={"Authorization": f"Bearer {token}"}, + json={"current_pin": "9999", "new_username": "student1"}, + ) + assert res.status_code == 401 + assert res.json()["detail"] == "Invalid current PIN." diff --git a/backend/tests/api/users/test_profile.py b/backend/tests/api/users/test_profile.py new file mode 100644 index 0000000..b7bb8c5 --- /dev/null +++ b/backend/tests/api/users/test_profile.py @@ -0,0 +1,52 @@ +from fastapi.testclient import TestClient + +from src.security.auth import hash_pin + + +def test_get_me_success(seeded_db, client: TestClient): + """ + GET /users/me returns authenticated caller profile. + """ + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + token = login_res.json()["session_id"] + + response = client.get("/users/me", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 200 + data = response.json() + assert data["username"] == "student1" + assert data["role"] == "student" + assert data["must_change_pin"] is False + assert isinstance(data["user_id"], int) + + +def test_get_me_unauthenticated(client: TestClient): + """ + GET /users/me without Bearer header returns 401. + """ + response = client.get("/users/me") + assert response.status_code == 401 + + +def test_get_me_permitted_during_pending_rotation(temp_db, client: TestClient): + """ + GET /users/me is on the forced-rotation allowlist and returns 200 with must_change_pin=True. + """ + _, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin, must_change_pin) VALUES (?, ?, 1)", + ("must_rotate", hashed), + ) + conn.commit() + + login_res = client.post("/login", json={"username": "must_rotate", "pin": "1234"}) + assert login_res.status_code == 200 + token = login_res.json()["session_id"] + + response = client.get("/users/me", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 200 + data = response.json() + assert data["username"] == "must_rotate" + assert data["must_change_pin"] is True diff --git a/backend/tests/api/users/test_signup.py b/backend/tests/api/users/test_signup.py new file mode 100644 index 0000000..3b7c781 --- /dev/null +++ b/backend/tests/api/users/test_signup.py @@ -0,0 +1,104 @@ +from fastapi.testclient import TestClient + +from src.db.database import get_db_connection + + +def test_signup_success(temp_db, client: TestClient): + """ + POST /signup creates an active student account and returns 201 Created. + """ + db_path, _ = temp_db + response = client.post("/signup", json={"username": "new_student", "pin": "1234"}) + assert response.status_code == 201 + data = response.json() + assert data["username"] == "new_student" + assert data["role"] == "student" + + # Verify user row in SQLite + conn = get_db_connection(db_path) + cursor = conn.cursor() + cursor.execute( + "SELECT id, username, role, must_change_pin, deleted_at FROM users WHERE username = ?", + ("new_student",), + ) + user = cursor.fetchone() + assert user is not None + assert user["role"] == "student" + assert user["must_change_pin"] == 0 + assert user["deleted_at"] is None + conn.close() + + +def test_signup_duplicate_username_returns_409(seeded_db, client: TestClient): + """ + POST /signup with an existing username returns 409 Conflict. + """ + response = client.post("/signup", json={"username": "student1", "pin": "5678"}) + assert response.status_code == 409 + assert response.json()["detail"] == "Username already taken." + + +def test_signup_rate_limiting_triggers_429(temp_db, client: TestClient, monkeypatch): + """ + Global signup limiter throttles after limit is reached, returning 429. + """ + import sys + + for mod_name in ("security.rate_limit", "src.security.rate_limit"): + if mod_name in sys.modules: + monkeypatch.setattr( + sys.modules[mod_name].signup_rate_limiter, "allow", lambda: False + ) + + res = client.post("/signup", json={"username": "flood_blocked", "pin": "1234"}) + assert res.status_code == 429 + assert "Too many signup attempts" in res.json()["detail"] + + +def test_signup_validation_errors(temp_db, client: TestClient): + """ + POST /signup validates input bounds and formats, returning 422 Unprocessable Entity. + """ + # Oversized username + assert ( + client.post("/signup", json={"username": "a" * 33, "pin": "1234"}).status_code + == 422 + ) + + # Undersized username + assert ( + client.post("/signup", json={"username": "ab", "pin": "1234"}).status_code + == 422 + ) + + # Invalid characters in username + assert ( + client.post( + "/signup", json={"username": "user space", "pin": "1234"} + ).status_code + == 422 + ) + + # Non-numeric PIN + assert ( + client.post( + "/signup", json={"username": "valid_user", "pin": "abcd"} + ).status_code + == 422 + ) + + # Short PIN (<4) + assert ( + client.post( + "/signup", json={"username": "valid_user", "pin": "123"} + ).status_code + == 422 + ) + + # Long PIN (>8) + assert ( + client.post( + "/signup", json={"username": "valid_user", "pin": "123456789"} + ).status_code + == 422 + ) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 9377a4f..f4c63ad 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,5 +1,6 @@ import os import sqlite3 +import sys import tempfile import pytest @@ -11,15 +12,23 @@ @pytest.fixture(autouse=True) -def _reset_rate_limiter(): +def _reset_rate_limiters(): """ - Ensure rate limiter state is clean before and after every test. + Ensure rate limiter state is clean before and after every test across all import aliases. """ - from src.security.rate_limit import login_rate_limiter - login_rate_limiter.clear() + def _clear_all(): + for mod_name in ("security.rate_limit", "src.security.rate_limit"): + mod = sys.modules.get(mod_name) + if mod: + if hasattr(mod, "login_rate_limiter"): + mod.login_rate_limiter.clear() + if hasattr(mod, "signup_rate_limiter"): + mod.signup_rate_limiter.clear() + + _clear_all() yield - login_rate_limiter.clear() + _clear_all() @pytest.fixture @@ -70,3 +79,49 @@ def client(): """ with TestClient(app) as test_client: yield test_client + + +@pytest.fixture +def staff_db(temp_db): + """ + Pre-seeds standard test roster across all roles: + - student1 (pin: 1234, role: student) + - student2 (pin: 1234, role: student) + - teacher1 (pin: 1234, role: teacher) + - admin1 (pin: 1234, role: admin) + """ + db_path, conn = temp_db + hashed = hash_pin("1234") + cursor = conn.cursor() + users = [ + ("student1", hashed, "student"), + ("student2", hashed, "student"), + ("teacher1", hashed, "teacher"), + ("admin1", hashed, "admin"), + ] + cursor.executemany( + "INSERT INTO users (username, hashed_pin, role) VALUES (?, ?, ?)", + users, + ) + conn.commit() + return db_path, conn + + +def auth_headers( + client: TestClient, username: str, pin: str = "1234" +) -> dict[str, str]: + """Helper to log in and return Bearer authorization headers.""" + response = client.post("/login", json={"username": username, "pin": pin}) + assert response.status_code == 200, ( + f"Login failed for {username}: {response.json()}" + ) + return {"Authorization": f"Bearer {response.json()['session_id']}"} + + +def get_user_id(conn: sqlite3.Connection, username: str) -> int: + """Helper to query user id by username.""" + cursor = conn.cursor() + cursor.execute("SELECT id FROM users WHERE username = ?", (username,)) + row = cursor.fetchone() + assert row is not None, f"User {username} not found in test database." + return row["id"] diff --git a/backend/tests/db/__init__.py b/backend/tests/db/__init__.py new file mode 100644 index 0000000..1c35617 --- /dev/null +++ b/backend/tests/db/__init__.py @@ -0,0 +1 @@ +"""Database tests.""" diff --git a/backend/tests/db/test_audit.py b/backend/tests/db/test_audit.py new file mode 100644 index 0000000..fefc4fa --- /dev/null +++ b/backend/tests/db/test_audit.py @@ -0,0 +1,47 @@ +import pytest + +from src.db.audit import VALID_ACTIONS, record_audit + + +def test_record_audit_valid_actions(temp_db): + """ + record_audit successfully persists rows for all defined valid actions. + """ + _, conn = temp_db + for action in VALID_ACTIONS: + record_audit( + conn, + actor_user_id=1, + action=action, + target_user_id=2, + ) + conn.commit() + + cursor = conn.cursor() + cursor.execute("SELECT COUNT(*) FROM audit_logs") + assert cursor.fetchone()[0] == len(VALID_ACTIONS) + + +def test_record_audit_none_actor_and_target(temp_db): + """ + record_audit accepts None for actor_user_id (e.g. self-signup) and target_user_id. + """ + _, conn = temp_db + record_audit(conn, actor_user_id=None, action="signup", target_user_id=None) + conn.commit() + + cursor = conn.cursor() + cursor.execute("SELECT actor_user_id, action, target_user_id FROM audit_logs") + row = cursor.fetchone() + assert row["actor_user_id"] is None + assert row["action"] == "signup" + assert row["target_user_id"] is None + + +def test_record_audit_invalid_action_raises_value_error(temp_db): + """ + record_audit raises ValueError when given an unknown action. + """ + _, conn = temp_db + with pytest.raises(ValueError, match="Unknown audit action"): + record_audit(conn, actor_user_id=1, action="invalid_action_xyz") diff --git a/backend/tests/test_database.py b/backend/tests/db/test_database.py similarity index 100% rename from backend/tests/test_database.py rename to backend/tests/db/test_database.py diff --git a/backend/tests/test_migrations.py b/backend/tests/db/test_migrations.py similarity index 88% rename from backend/tests/test_migrations.py rename to backend/tests/db/test_migrations.py index 220ee15..1ef5d46 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/db/test_migrations.py @@ -22,11 +22,17 @@ def test_migrations_applied_successfully(): assert 1 in versions assert 2 in versions assert 3 in versions + assert 4 in versions + assert 5 in versions + assert 6 in versions - # Verify role column added by migration 002 exists + # Verify columns added by migrations 002, 004, 005 exist on users cursor.execute("PRAGMA table_info(users)") columns = {col[1] for col in cursor.fetchall()} assert "role" in columns + assert "must_change_pin" in columns + assert "deleted_at" in columns + assert "former_username" in columns # Verify FK lookup indexes added by migration 003 exist cursor.execute( @@ -34,6 +40,22 @@ def test_migrations_applied_successfully(): "AND name='idx_turn_logs_session_id'" ) assert cursor.fetchone() is not None + + # Verify audit_logs table and indexes added by migration 006 exist + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='audit_logs'" + ) + assert cursor.fetchone() is not None + + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_audit_logs_actor'" + ) + assert cursor.fetchone() is not None + + cursor.execute( + "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_audit_logs_target'" + ) + assert cursor.fetchone() is not None conn.close() finally: if os.path.exists(db_path): diff --git a/backend/tests/security/__init__.py b/backend/tests/security/__init__.py new file mode 100644 index 0000000..ec33a68 --- /dev/null +++ b/backend/tests/security/__init__.py @@ -0,0 +1 @@ +"""Security tests.""" diff --git a/backend/tests/security/test_auth_pin.py b/backend/tests/security/test_auth_pin.py new file mode 100644 index 0000000..f38b972 --- /dev/null +++ b/backend/tests/security/test_auth_pin.py @@ -0,0 +1,48 @@ +from src.security.auth import hash_pin, verify_pin + + +def test_hash_pin_generates_bcrypt_hash(): + """ + Test that hash_pin produces a valid bcrypt hash starting with $2b$. + """ + pin = "1234" + hashed = hash_pin(pin) + assert hashed != pin + assert hashed.startswith(("$2b$", "$2a$")) + + +def test_hash_pin_generates_unique_salts(): + """ + Test that hashing the same PIN twice yields different hash strings due to salt. + """ + pin = "1234" + hash1 = hash_pin(pin) + hash2 = hash_pin(pin) + assert hash1 != hash2 + assert verify_pin(pin, hash1) is True + assert verify_pin(pin, hash2) is True + + +def test_verify_pin_success(): + """ + Test verify_pin returns True for valid PIN and hash. + """ + pin = "5678" + hashed = hash_pin(pin) + assert verify_pin(pin, hashed) is True + + +def test_verify_pin_failure(): + """ + Test verify_pin returns False for incorrect PIN. + """ + pin = "5678" + hashed = hash_pin(pin) + assert verify_pin("0000", hashed) is False + + +def test_verify_pin_invalid_hash_format(): + """ + Test verify_pin handles invalid hash string format gracefully without crashing. + """ + assert verify_pin("1234", "not_a_valid_bcrypt_hash") is False diff --git a/backend/tests/test_rate_limit.py b/backend/tests/security/test_rate_limit.py similarity index 68% rename from backend/tests/test_rate_limit.py rename to backend/tests/security/test_rate_limit.py index 518fb34..3e52b7a 100644 --- a/backend/tests/test_rate_limit.py +++ b/backend/tests/security/test_rate_limit.py @@ -70,3 +70,38 @@ def test_rate_limiter_sweeps_expired_lockouts_on_write(monkeypatch): limiter.record_failure("other_user") assert "student1" not in limiter._lockout_until assert "student1" not in limiter._failed_attempts + + +def test_sliding_window_limiter_enforces_limit_and_expires(monkeypatch): + """ + SlidingWindowLimiter allows events up to max_events, blocks excess, + and expires old events after the window elapses. + """ + import time + + from src.security.rate_limit import SlidingWindowLimiter + + limiter = SlidingWindowLimiter(max_events=2, window_seconds=10) + assert limiter.allow() is True + assert limiter.allow() is True + assert limiter.allow() is False # Limit reached + + # Advance time past window + current_time = time.time() + monkeypatch.setattr(time, "time", lambda: current_time + 15) + + # Expired events popped, new event allowed + assert limiter.allow() is True + + +def test_sliding_window_limiter_clear(): + """ + SlidingWindowLimiter.clear resets internal event queue. + """ + from src.security.rate_limit import SlidingWindowLimiter + + limiter = SlidingWindowLimiter(max_events=1, window_seconds=10) + assert limiter.allow() is True + assert limiter.allow() is False + limiter.clear() + assert limiter.allow() is True diff --git a/backend/tests/security/test_security_pin.py b/backend/tests/security/test_security_pin.py new file mode 100644 index 0000000..6e0bfbc --- /dev/null +++ b/backend/tests/security/test_security_pin.py @@ -0,0 +1,213 @@ +import logging + +from fastapi.testclient import TestClient + +from src.security.auth import hash_pin +from tests.conftest import auth_headers + + +def test_plain_text_pin_not_in_database(temp_db): + """ + SECURITY PROOF 1: + Verify that when a user is created with PIN '1234', the plain text string + '1234' is NEVER stored in the database row or columns. + """ + _, conn = temp_db + test_pin = "1234" + hashed = hash_pin(test_pin) + + cursor = conn.cursor() + cursor.execute( + "INSERT INTO users (username, hashed_pin) VALUES (?, ?)", + ("student_sec_test", hashed), + ) + conn.commit() + + # Query the user row with raw SQL + cursor.execute("SELECT * FROM users WHERE username = 'student_sec_test'") + row = cursor.fetchone() + assert row is not None + + # Verify no column is named exactly "pin" (only "hashed_pin" should exist) + row_dict = dict(row) + assert "hashed_pin" in row_dict + assert "pin" not in row_dict, "A bare 'pin' column exists — rename to 'hashed_pin'" + + # Strict assertion: The plain text PIN string must not be present in any row value + for col_name, value in row_dict.items(): + assert test_pin != str(value), f"Plain text PIN leaked in column: {col_name}" + assert test_pin not in str(value), ( + f"Plain text PIN found as substring in: {col_name}" + ) + + +def test_plain_text_pin_never_logged_during_login( + seeded_db, client: TestClient, caplog +): + """ + SECURITY PROOF 2: + Verify that during both successful and failed authentication attempts, + the plain-text PIN '1234' or '9999' is NEVER output to any logging handler. + """ + sensitive_pins = ["1234", "9999", "8888"] + + with caplog.at_level(logging.DEBUG): + # 1. Attempt successful login + client.post("/login", json={"username": "student1", "pin": "1234"}) + + # 2. Attempt failed login (wrong PIN) + client.post("/login", json={"username": "student1", "pin": "9999"}) + + # 3. Attempt failed login (unknown user) + client.post("/login", json={"username": "unknown_student", "pin": "8888"}) + + # Inspect all captured log messages + for record in caplog.records: + log_message = record.getMessage() + for sensitive_pin in sensitive_pins: + assert sensitive_pin not in log_message, ( + f"Security violation: Sensitive PIN '{sensitive_pin}' was logged in message: '{log_message}'" + ) + + +def test_plain_text_pin_never_logged_during_hashing(caplog): + """ + SECURITY PROOF 3: + Verify that the hash_pin and verify_pin functions do not log the plain text PIN. + """ + sensitive_pin = "4321" + + with caplog.at_level(logging.DEBUG): + hashed = hash_pin(sensitive_pin) + from src.security.auth import verify_pin + + verify_pin(sensitive_pin, hashed) + verify_pin("wrong_pin", hashed) + + for record in caplog.records: + log_message = record.getMessage() + assert sensitive_pin not in log_message, ( + f"Security violation: Plain text PIN '{sensitive_pin}' leaked in logger: '{log_message}'" + ) + assert "wrong_pin" not in log_message + + +def test_plain_text_pin_never_logged_during_signup(temp_db, client: TestClient, caplog): + """ + SECURITY PROOF 4: + Verify that during signup attempts (successful, conflict, or malformed), + plain-text PINs are never output to any logging handler. + """ + sensitive_pins = ["7777", "8888"] + + with caplog.at_level(logging.DEBUG): + # 1. Successful signup + client.post( + "/signup", json={"username": "sec_student", "pin": sensitive_pins[0]} + ) + # 2. Duplicate signup attempt + client.post( + "/signup", json={"username": "sec_student", "pin": sensitive_pins[1]} + ) + + for record in caplog.records: + log_message = record.getMessage() + for sensitive_pin in sensitive_pins: + assert sensitive_pin not in log_message, ( + f"Security violation: Sensitive PIN '{sensitive_pin}' logged in signup: '{log_message}'" + ) + + +def test_plain_text_pin_never_logged_during_credential_change( + seeded_db, client: TestClient, caplog +): + """ + SECURITY PROOF 5: + Verify that during PIN changes and username changes, + neither current nor new plain-text PINs are logged. + """ + sensitive_pins = ["1234", "9876", "1111"] + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + token = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {token}"} + + with caplog.at_level(logging.DEBUG): + # 1. Successful PIN change + client.patch( + "/users/me/pin", + headers=headers, + json={"current_pin": sensitive_pins[0], "new_pin": sensitive_pins[1]}, + ) + # 2. Failed username change with bad PIN + client.patch( + "/users/me/username", + headers=headers, + json={ + "current_pin": sensitive_pins[2], + "new_username": "new_student_name", + }, + ) + + for record in caplog.records: + log_message = record.getMessage() + for sensitive_pin in sensitive_pins: + assert sensitive_pin not in log_message, ( + f"Security violation: Sensitive PIN '{sensitive_pin}' logged in credential change: '{log_message}'" + ) + + +def test_plain_text_pin_never_logged_during_staff_user_creation( + staff_db, client: TestClient, caplog +): + """ + SECURITY PROOF 6: + Verify that when staff creates a user account, the initial PIN is never logged. + """ + sensitive_pin = "3333" + headers = auth_headers(client, "admin1", "1234") + + with caplog.at_level(logging.DEBUG): + client.post( + "/users", + headers=headers, + json={ + "username": "staff_created_user", + "pin": sensitive_pin, + "role": "student", + }, + ) + + for record in caplog.records: + log_message = record.getMessage() + assert sensitive_pin not in log_message, ( + f"Security violation: Sensitive PIN '{sensitive_pin}' logged in staff user creation: '{log_message}'" + ) + + +def test_session_id_never_logged_across_endpoints( + seeded_db, client: TestClient, caplog +): + """ + SECURITY PROOF 7: + Bearer session identifier must never appear in any log output across + login, user profile check, and logout workflows. + """ + with caplog.at_level(logging.DEBUG): + # 1. Login + login_res = client.post("/login", json={"username": "student1", "pin": "1234"}) + assert login_res.status_code == 200 + session_id = login_res.json()["session_id"] + headers = {"Authorization": f"Bearer {session_id}"} + + # 2. Profile + profile_res = client.get("/users/me", headers=headers) + assert profile_res.status_code == 200 + + # 3. Logout + logout_res = client.post("/logout", headers=headers) + assert logout_res.status_code == 200 + + for record in caplog.records: + assert session_id not in record.getMessage(), ( + f"Security violation: session ID leaked in log: '{record.getMessage()}'" + ) diff --git a/backend/tests/security/test_session_auth.py b/backend/tests/security/test_session_auth.py new file mode 100644 index 0000000..a76b8fd --- /dev/null +++ b/backend/tests/security/test_session_auth.py @@ -0,0 +1,242 @@ +import uuid +from typing import Annotated + +from fastapi import Depends, FastAPI, status +from fastapi.testclient import TestClient + +from src.db.database import get_db_connection +from src.security.auth import hash_pin +from src.security.session import ( + AuthContext, + ensure_no_pending_rotation, + get_current_session, + require_roles, +) + +# Sample FastAPI app to test session dependencies in isolation +sample_app = FastAPI() + + +@sample_app.get("/test-session") +def route_session(ctx: Annotated[AuthContext, Depends(get_current_session)]): + return { + "user_id": ctx.user_id, + "username": ctx.username, + "role": ctx.role, + "session_id": ctx.session_id, + "must_change_pin": ctx.must_change_pin, + } + + +@sample_app.get("/test-teacher-only") +def route_teacher_only( + ctx: Annotated[AuthContext, Depends(require_roles("teacher", "admin"))], +): + return {"status": "ok", "user": ctx.username} + + +@sample_app.get("/test-admin-only") +def route_admin_only( + ctx: Annotated[AuthContext, Depends(require_roles("admin"))], +): + return {"status": "ok", "user": ctx.username} + + +@sample_app.get("/test-no-rotation") +def route_no_rotation( + ctx: Annotated[AuthContext, Depends(get_current_session)], +): + ensure_no_pending_rotation(ctx) + return {"status": "ok"} + + +def _seed_user_and_session( + db_path: str, + username: str, + role: str = "student", + is_active: int = 1, + deleted: bool = False, + must_change_pin: int = 0, +) -> tuple[int, str]: + """Helper to seed a user and return (user_id, session_id).""" + conn = get_db_connection(db_path) + try: + cursor = conn.cursor() + hashed = hash_pin("1234") + deleted_at = "2026-08-23 00:00:00" if deleted else None + cursor.execute( + "INSERT INTO users (username, hashed_pin, role, deleted_at, must_change_pin) " + "VALUES (?, ?, ?, ?, ?)", + (username, hashed, role, deleted_at, must_change_pin), + ) + user_id = cursor.lastrowid + assert user_id is not None + session_id = str(uuid.uuid4()) + cursor.execute( + "INSERT INTO sessions (id, user_id, is_active) VALUES (?, ?, ?)", + (session_id, user_id, is_active), + ) + conn.commit() + return user_id, session_id + finally: + conn.close() + + +def test_missing_authorization_header(temp_db): + client = TestClient(sample_app) + response = client.get("/test-session") + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert "Missing or malformed" in response.json()["detail"] + + +def test_malformed_authorization_header(temp_db): + client = TestClient(sample_app) + # Basic scheme instead of Bearer + response = client.get( + "/test-session", headers={"Authorization": "Basic dXNlcjpwYXNz"} + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert "Missing or malformed" in response.json()["detail"] + + +def test_invalid_uuid_token_format(temp_db): + client = TestClient(sample_app) + # Non-UUID token format (junk in guard) + response = client.get( + "/test-session", headers={"Authorization": "Bearer not-a-uuid-token"} + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert response.json()["detail"] == "Invalid session token." + + +def test_nonexistent_session_token(temp_db): + client = TestClient(sample_app) + fake_token = str(uuid.uuid4()) + response = client.get( + "/test-session", headers={"Authorization": f"Bearer {fake_token}"} + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert response.json()["detail"] == "Invalid or expired session." + + +def test_inactive_session_token(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, session_id = _seed_user_and_session(db_path, "inactive_user", is_active=0) + response = client.get( + "/test-session", headers={"Authorization": f"Bearer {session_id}"} + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert response.json()["detail"] == "Invalid or expired session." + + +def test_soft_deleted_user_session_rejected(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, session_id = _seed_user_and_session(db_path, "deleted_user", deleted=True) + response = client.get( + "/test-session", headers={"Authorization": f"Bearer {session_id}"} + ) + assert response.status_code == status.HTTP_401_UNAUTHORIZED + assert response.json()["detail"] == "Invalid or expired session." + + +def test_valid_active_session_resolves_context(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + user_id, session_id = _seed_user_and_session( + db_path, "student_valid", role="student", must_change_pin=0 + ) + response = client.get( + "/test-session", headers={"Authorization": f"Bearer {session_id}"} + ) + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["user_id"] == user_id + assert data["username"] == "student_valid" + assert data["role"] == "student" + assert data["session_id"] == session_id + assert data["must_change_pin"] is False + + +def test_require_roles_allows_matching_role(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, teacher_token = _seed_user_and_session(db_path, "teacher1", role="teacher") + _, admin_token = _seed_user_and_session(db_path, "admin1", role="admin") + + # Teacher accesses teacher-only route + res_t = client.get( + "/test-teacher-only", + headers={"Authorization": f"Bearer {teacher_token}"}, + ) + assert res_t.status_code == status.HTTP_200_OK + + # Admin accesses teacher-only route (since admin is in allowed roles) + res_a = client.get( + "/test-teacher-only", + headers={"Authorization": f"Bearer {admin_token}"}, + ) + assert res_a.status_code == status.HTTP_200_OK + + +def test_require_roles_rejects_insufficient_role(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, student_token = _seed_user_and_session(db_path, "student_user", role="student") + _, teacher_token = _seed_user_and_session(db_path, "teacher_user", role="teacher") + + # Student cannot access teacher/admin route + res = client.get( + "/test-teacher-only", + headers={"Authorization": f"Bearer {student_token}"}, + ) + assert res.status_code == status.HTTP_403_FORBIDDEN + assert res.json()["detail"] == "Insufficient permissions." + + # Teacher cannot access admin-only route + res_admin = client.get( + "/test-admin-only", + headers={"Authorization": f"Bearer {teacher_token}"}, + ) + assert res_admin.status_code == status.HTTP_403_FORBIDDEN + assert res_admin.json()["detail"] == "Insufficient permissions." + + +def test_ensure_no_pending_rotation_enforces_gate(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, pending_token = _seed_user_and_session( + db_path, "must_rotate_user", must_change_pin=1 + ) + _, clean_token = _seed_user_and_session(db_path, "clean_user", must_change_pin=0) + + # Clean user passes + res_clean = client.get( + "/test-no-rotation", + headers={"Authorization": f"Bearer {clean_token}"}, + ) + assert res_clean.status_code == status.HTTP_200_OK + + # User with must_change_pin=1 is blocked with 403 + res_pending = client.get( + "/test-no-rotation", + headers={"Authorization": f"Bearer {pending_token}"}, + ) + assert res_pending.status_code == status.HTTP_403_FORBIDDEN + assert res_pending.json()["detail"] == "PIN change required." + + +def test_require_roles_blocks_user_with_pending_rotation(temp_db): + db_path, _ = temp_db + client = TestClient(sample_app) + _, teacher_pending = _seed_user_and_session( + db_path, "teacher_pending", role="teacher", must_change_pin=1 + ) + + res = client.get( + "/test-teacher-only", + headers={"Authorization": f"Bearer {teacher_pending}"}, + ) + assert res.status_code == status.HTTP_403_FORBIDDEN + assert res.json()["detail"] == "PIN change required." diff --git a/backend/tests/test_security_pin.py b/backend/tests/test_security_pin.py deleted file mode 100644 index 556f481..0000000 --- a/backend/tests/test_security_pin.py +++ /dev/null @@ -1,91 +0,0 @@ -import logging - -from fastapi.testclient import TestClient - -from src.security.auth import hash_pin - - -def test_plain_text_pin_not_in_database(temp_db): - """ - SECURITY PROOF 1: - Verify that when a user is created with PIN '1234', the plain text string - '1234' is NEVER stored in the database row or columns. - """ - _, conn = temp_db - test_pin = "1234" - hashed = hash_pin(test_pin) - - cursor = conn.cursor() - cursor.execute( - "INSERT INTO users (username, hashed_pin) VALUES (?, ?)", - ("student_sec_test", hashed), - ) - conn.commit() - - # Query the user row with raw SQL - cursor.execute("SELECT * FROM users WHERE username = 'student_sec_test'") - row = cursor.fetchone() - assert row is not None - - # Verify no column is named exactly "pin" (only "hashed_pin" should exist) - row_dict = dict(row) - assert "hashed_pin" in row_dict - assert "pin" not in row_dict, "A bare 'pin' column exists — rename to 'hashed_pin'" - - # Strict assertion: The plain text PIN string must not be present in any row value - for col_name, value in row_dict.items(): - assert test_pin != str(value), f"Plain text PIN leaked in column: {col_name}" - assert test_pin not in str(value), ( - f"Plain text PIN found as substring in: {col_name}" - ) - - -def test_plain_text_pin_never_logged_during_login( - seeded_db, client: TestClient, caplog -): - """ - SECURITY PROOF 2: - Verify that during both successful and failed authentication attempts, - the plain-text PIN '1234' or '9999' is NEVER output to any logging handler. - """ - sensitive_pins = ["1234", "9999", "8888"] - - with caplog.at_level(logging.DEBUG): - # 1. Attempt successful login - client.post("/login", json={"username": "student1", "pin": "1234"}) - - # 2. Attempt failed login (wrong PIN) - client.post("/login", json={"username": "student1", "pin": "9999"}) - - # 3. Attempt failed login (unknown user) - client.post("/login", json={"username": "unknown_student", "pin": "8888"}) - - # Inspect all captured log messages - for record in caplog.records: - log_message = record.getMessage() - for sensitive_pin in sensitive_pins: - assert sensitive_pin not in log_message, ( - f"Security violation: Sensitive PIN '{sensitive_pin}' was logged in message: '{log_message}'" - ) - - -def test_plain_text_pin_never_logged_during_hashing(caplog): - """ - SECURITY PROOF 3: - Verify that the hash_pin and verify_pin functions do not log the plain text PIN. - """ - sensitive_pin = "4321" - - with caplog.at_level(logging.DEBUG): - hashed = hash_pin(sensitive_pin) - from src.security.auth import verify_pin - - verify_pin(sensitive_pin, hashed) - verify_pin("wrong_pin", hashed) - - for record in caplog.records: - log_message = record.getMessage() - assert sensitive_pin not in log_message, ( - f"Security violation: Plain text PIN '{sensitive_pin}' leaked in logger: '{log_message}'" - ) - assert "wrong_pin" not in log_message