fix: remediate deep-audit security and integrity findings

backend/alembic/versions/003_audit_remediation_constraints.py

@@ -0,0 +1,59 @@
+"""Add integrity constraints for audit remediation."""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "003_audit_remediation"
+down_revision: Union[str, None] = "002_employer_claim_constraints"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _assert_no_case_collisions(table: str, column: str) -> None:
+    """Fail the migration when case-equivalent duplicates exist."""
+    connection = op.get_bind()
+    result = connection.execute(sa.text(f"""
+            SELECT lower({column}) AS normalized_value, count(*) AS collision_count
+            FROM {table}
+            GROUP BY lower({column})
+            HAVING count(*) > 1
+            LIMIT 1
+            """))
+    row = result.first()
+    if row is not None:
+        raise RuntimeError(
+            f"Cannot add case-insensitive unique index on {table}.{column}; "
+            "resolve case-equivalent duplicates first."
+        )
+
+
+def upgrade() -> None:
+    """Add employer-response, duplicate-report, and identity constraints."""
+    _assert_no_case_collisions("users", "email")
+    _assert_no_case_collisions("users", "username")
+
+    op.execute("ALTER TABLE users DROP CONSTRAINT IF EXISTS users_email_key")
+    op.execute("ALTER TABLE users DROP CONSTRAINT IF EXISTS users_username_key")
+    op.execute("CREATE UNIQUE INDEX uq_users_email_lower ON users (lower(email))")
+    op.execute("CREATE UNIQUE INDEX uq_users_username_lower ON users (lower(username))")
+    op.execute("""
+        CREATE UNIQUE INDEX uq_employer_responses_report_user
+        ON employer_responses (report_id, user_id)
+        """)
+    op.execute("""
+        CREATE UNIQUE INDEX uq_reports_active_duplicate
+        ON reports (reporter_id, job_posting_id, report_type)
+        WHERE status IN ('pending', 'verified', 'disputed')
+        """)
+
+
+def downgrade() -> None:
+    """Remove audit remediation constraints and restore original user uniqueness."""
+    op.execute("DROP INDEX IF EXISTS uq_reports_active_duplicate")
+    op.execute("DROP INDEX IF EXISTS uq_employer_responses_report_user")
+    op.execute("DROP INDEX IF EXISTS uq_users_username_lower")
+    op.execute("DROP INDEX IF EXISTS uq_users_email_lower")
+    op.execute("ALTER TABLE users ADD CONSTRAINT users_email_key UNIQUE (email)")
+    op.execute("ALTER TABLE users ADD CONSTRAINT users_username_key UNIQUE (username)")

backend/app/api/v1/reports.py

@@ -2,35 +2,61 @@
 
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, Query, Request
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.dependencies import get_current_user, get_db
+from app.config import Settings
+from app.dependencies import (
+    get_current_user,
+    get_db,
+    get_optional_current_user,
+    get_redis,
+    get_settings_dependency,
+)
 from app.models.user import User
+from app.redis_client import RedisClient
 from app.schemas import (
     CreateEmployerResponseRequest,
     CreateReportRequest,
     CreateVoteRequest,
-    EmployerResponseListResponse,
     EmployerResponseResponse,
-    ReportListResponse,
+    PublicEmployerResponseListResponse,
+    PublicReportListResponse,
+    PublicReportResponse,
     ReportResponse,
     VoteResponse,
 )
 from app.services import employer_responses as employer_responses_service
 from app.services import reports as reports_service
 from app.services import votes as votes_service
+from app.services.rate_limit import check_report_submission_rate_limit
 
 router = APIRouter(prefix="/reports", tags=["reports"])
 
 
+def _client_ip(request: Request) -> str:
+    """Return the client IP address for rate limiting."""
+    if request.client is None:
+        return "unknown"
+    return request.client.host
+
+
 @router.post("", response_model=ReportResponse, status_code=201)
 async def create_report(
     payload: CreateReportRequest,
+    request: Request,
     session: AsyncSession = Depends(get_db),
+    redis: RedisClient = Depends(get_redis),
+    settings: Settings = Depends(get_settings_dependency),
     current_user: User = Depends(get_current_user),
 ) -> ReportResponse:
     """Submit an integrity report for a job posting."""
+    await check_report_submission_rate_limit(
+        redis,
+        str(current_user.id),
+        _client_ip(request),
+        settings,
+    )
     report = await reports_service.create_report(
         session,
         payload=payload,
@@ -39,23 +65,26 @@ async def create_report(
     return ReportResponse.model_validate(report)
 
 
-@router.get("/{report_id}", response_model=ReportResponse)
+@router.get("/{report_id}", response_model=PublicReportResponse | ReportResponse)
 async def get_report(
     report_id: UUID,
     session: AsyncSession = Depends(get_db),
-) -> ReportResponse:
-    """Return a submitted report."""
-    report = await reports_service.get_report(session, report_id)
-    return ReportResponse.model_validate(report)
+    viewer: User | None = Depends(get_optional_current_user),
+) -> PublicReportResponse | ReportResponse:
+    """Return a report when visible to the caller."""
+    return await reports_service.get_report_response(session, report_id, viewer=viewer)
 
 
-@router.get("/{report_id}/responses", response_model=EmployerResponseListResponse)
+@router.get("/{report_id}/responses", response_model=PublicEmployerResponseListResponse)
 async def list_employer_responses(
     report_id: UUID,
     session: AsyncSession = Depends(get_db),
-) -> EmployerResponseListResponse:
-    """Return employer responses linked to a report."""
-    return await employer_responses_service.list_employer_responses(session, report_id=report_id)
+) -> PublicEmployerResponseListResponse:
+    """Return public employer responses linked to a visible report."""
+    return await employer_responses_service.list_public_employer_responses(
+        session,
+        report_id=report_id,
+    )
 
 
 @router.post("/{report_id}/responses", response_model=EmployerResponseResponse, status_code=201)
@@ -75,15 +104,15 @@ async def create_employer_response(
     return EmployerResponseResponse.model_validate(response)
 
 
-@router.get("", response_model=ReportListResponse)
+@router.get("", response_model=PublicReportListResponse)
 async def list_reports(
     job_posting_id: UUID = Query(...),
     page: int = Query(default=1, ge=1),
     page_size: int = Query(default=20, ge=1, le=100),
     session: AsyncSession = Depends(get_db),
-) -> ReportListResponse:
-    """List reports linked to a job posting."""
-    return await reports_service.list_reports_for_job_posting(
+) -> PublicReportListResponse:
+    """List publicly visible reports linked to a job posting."""
+    return await reports_service.list_public_reports_for_job_posting(
         session,
         job_posting_id,
         page=page,

backend/app/config.py

@@ -37,6 +37,7 @@ class Settings(BaseSettings):
     cors_origins: list[str] = Field(default_factory=lambda: ["http://localhost:3000"])
 
     auth_rate_limit_per_minute: int = 20
+    report_submission_rate_limit_per_hour: int = 10
 
     @model_validator(mode="before")
     @classmethod

backend/app/db_errors.py

@@ -6,9 +6,13 @@
 
 USER_EMAIL_UNIQUE = "users_email_key"
 USER_USERNAME_UNIQUE = "users_username_key"
+USER_EMAIL_LOWER_UNIQUE = "uq_users_email_lower"
+USER_USERNAME_LOWER_UNIQUE = "uq_users_username_lower"
 VOTE_REPORT_USER_UNIQUE = "uq_votes_report_user"
 EMPLOYER_CLAIM_COMPANY_APPROVED = "uq_employer_claims_company_approved"
 EMPLOYER_CLAIM_USER_COMPANY_PENDING = "uq_employer_claims_user_company_pending"
+EMPLOYER_RESPONSE_REPORT_USER_UNIQUE = "uq_employer_responses_report_user"
+REPORT_ACTIVE_DUPLICATE_UNIQUE = "uq_reports_active_duplicate"
 
 
 def _unique_constraint_name(exc: IntegrityError) -> str | None:
@@ -32,12 +36,20 @@ def conflict_detail_from_integrity_error(exc: IntegrityError) -> str | None:
         return "Email already registered"
     if constraint_name == USER_USERNAME_UNIQUE:
         return "Username already taken"
+    if constraint_name in {USER_EMAIL_LOWER_UNIQUE, USER_EMAIL_UNIQUE}:
+        return "Email already registered"
+    if constraint_name in {USER_USERNAME_LOWER_UNIQUE, USER_USERNAME_UNIQUE}:
+        return "Username already taken"
     if constraint_name == VOTE_REPORT_USER_UNIQUE:
         return "Vote already recorded for this report"
     if constraint_name == EMPLOYER_CLAIM_COMPANY_APPROVED:
         return "Company already has an approved employer claim"
     if constraint_name == EMPLOYER_CLAIM_USER_COMPANY_PENDING:
         return "Pending claim already exists for this company"
+    if constraint_name == EMPLOYER_RESPONSE_REPORT_USER_UNIQUE:
+        return "An employer response from this account already exists for this report"
+    if constraint_name == REPORT_ACTIVE_DUPLICATE_UNIQUE:
+        return "A similar report from this account is already active for this posting"
     return None
 
 

backend/app/dependencies.py

@@ -95,3 +95,25 @@ async def require_admin(current_user: User = Depends(get_current_user)) -> User:
     if not current_user.is_admin:
         raise ForbiddenError()
     return current_user
+
+
+async def get_optional_current_user(
+    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer_scheme),
+    session: AsyncSession = Depends(get_db),
+    settings: Settings = Depends(get_settings_dependency),
+) -> User | None:
+    """Load the authenticated user when a valid bearer token is supplied.
+
+    Returns:
+        User | None: Authenticated user record, or None when unauthenticated.
+    """
+    if credentials is None:
+        return None
+
+    try:
+        subject = decode_access_token(credentials.credentials, settings)
+        user_id = UUID(subject)
+    except (UnauthorizedError, ValueError):
+        return None
+
+    return await get_user_by_id(session, user_id)

backend/app/main.py

@@ -4,18 +4,27 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI, status
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
 
 from app.api.v1.auth import router as auth_router
 from app.api.v1.companies import router as companies_router
 from app.api.v1.employer_claims import router as employer_claims_router
 from app.api.v1.job_postings import router as job_postings_router
 from app.api.v1.moderation import router as moderation_router
 from app.api.v1.reports import router as reports_router
-from app.config import get_settings
-from app.redis_client import init_redis_client, shutdown_redis_client
-from app.schemas import HealthResponse
+from app.config import Settings, get_settings
+from app.dependencies import get_redis, get_settings_dependency
+from app.redis_client import (
+    RedisClient,
+    check_redis_connection,
+    init_redis_client,
+    shutdown_redis_client,
+)
+from app.schemas import HealthResponse, ReadinessCheckStatus, ReadinessResponse
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -52,10 +61,48 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
 
 @app.get("/health", response_model=HealthResponse)
 async def health_check() -> HealthResponse:
-    """Return basic service health metadata."""
+    """Return lightweight process liveness without dependency probes."""
     return HealthResponse(status="ok", service=settings.app_name)
 
 
+@app.get("/health/ready", response_model=ReadinessResponse)
+async def readiness_check(
+    settings: Settings = Depends(get_settings_dependency),
+    redis: RedisClient = Depends(get_redis),
+) -> JSONResponse:
+    """Return dependency readiness for PostgreSQL and Redis."""
+    postgres_status = "unavailable"
+    redis_status = "unavailable"
+
+    engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+    try:
+        async with engine.connect() as connection:
+            await connection.execute(text("SELECT 1"))
+        postgres_status = "ok"
+    except Exception:
+        logger.exception("PostgreSQL readiness check failed")
+    finally:
+        await engine.dispose()
+
+    try:
+        if await check_redis_connection(redis):
+            redis_status = "ok"
+    except Exception:
+        logger.exception("Redis readiness check failed")
+
+    checks = ReadinessCheckStatus(postgres=postgres_status, redis=redis_status)
+    ready = postgres_status == "ok" and redis_status == "ok"
+    payload = ReadinessResponse(
+        status="ready" if ready else "not_ready",
+        service=settings.app_name,
+        checks=checks,
+    )
+    return JSONResponse(
+        status_code=status.HTTP_200_OK if ready else status.HTTP_503_SERVICE_UNAVAILABLE,
+        content=payload.model_dump(),
+    )
+
+
 @app.get("/")
 async def root() -> dict[str, str]:
     """Return API metadata."""