From c7bbdcd7d375b535bb9dc098f4be03ad7ec3e623 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 01/12] fix: enforce moderation boundaries on public reports Public report APIs now return only verified reports, hide pending content with 404, and expose separate public schemas without internal reporter identifiers. --- backend/app/api/v1/reports.py | 63 ++++++-- backend/app/dependencies.py | 22 +++ backend/app/schemas.py | 94 ++++++++++- backend/app/services/report_eligibility.py | 48 ++++++ backend/app/services/reports.py | 180 ++++++++++++++++++--- backend/tests/test_reports.py | 140 +++++++++------- docs/moderation-model.md | 18 ++- 7 files changed, 460 insertions(+), 105 deletions(-) create mode 100644 backend/app/services/report_eligibility.py diff --git a/backend/app/api/v1/reports.py b/backend/app/api/v1/reports.py index f465291..260e4ad 100644 --- a/backend/app/api/v1/reports.py +++ b/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, diff --git a/backend/app/dependencies.py b/backend/app/dependencies.py index 9b3de22..081e72b 100644 --- a/backend/app/dependencies.py +++ b/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) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 9d1bff0..655125a 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -15,16 +15,33 @@ VerifiedStatus, VoteValue, ) +from app.security.password_policy import validate_password_byte_length from app.services.job_url_validation import validate_http_https_url +from app.services.user_identity import normalize_email, normalize_username class HealthResponse(BaseModel): - """Service health response.""" + """Service liveness response.""" status: str service: str +class ReadinessCheckStatus(BaseModel): + """Individual dependency readiness status.""" + + postgres: str + redis: str + + +class ReadinessResponse(BaseModel): + """Dependency readiness response.""" + + status: str + service: str + checks: ReadinessCheckStatus + + class ScoreBreakdown(BaseModel): """Named score components returned to clients.""" @@ -47,6 +64,25 @@ class RegisterRequest(BaseModel): username: str = Field(min_length=3, max_length=64, pattern=r"^[a-zA-Z0-9_\-]+$") password: str = Field(min_length=12, max_length=128) + @field_validator("email") + @classmethod + def normalize_registration_email(cls, value: str) -> str: + """Normalize email before validation and persistence.""" + return normalize_email(str(value)) + + @field_validator("username") + @classmethod + def normalize_registration_username(cls, value: str) -> str: + """Normalize username before validation and persistence.""" + return normalize_username(value) + + @field_validator("password") + @classmethod + def validate_registration_password_bytes(cls, value: str) -> str: + """Reject passwords that exceed bcrypt's UTF-8 byte boundary.""" + validate_password_byte_length(value) + return value + class LoginRequest(BaseModel): """User login payload.""" @@ -54,6 +90,13 @@ class LoginRequest(BaseModel): identifier: str = Field(min_length=3, max_length=320) password: str = Field(min_length=12, max_length=128) + @field_validator("password") + @classmethod + def validate_login_password_bytes(cls, value: str) -> str: + """Reject passwords that exceed bcrypt's UTF-8 byte boundary.""" + validate_password_byte_length(value) + return value + class TokenResponse(BaseModel): """JWT access and refresh token response.""" @@ -140,8 +183,24 @@ class JobPostingResponse(BaseModel): model_config = {"from_attributes": True} +class PublicReportResponse(BaseModel): + """Public report details without internal identity fields.""" + + id: UUID + job_posting_id: UUID + report_type: ReportType + description: str + status: ReportStatus + confidence_score: float + verification_votes: int + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + class ReportResponse(BaseModel): - """Submitted report details.""" + """Submitted report details for authenticated internal access.""" id: UUID job_posting_id: UUID @@ -165,8 +224,17 @@ class CreateReportRequest(BaseModel): description: str = Field(min_length=20, max_length=5000) +class PublicReportListResponse(BaseModel): + """Paginated public reports linked to a job posting.""" + + items: list[PublicReportResponse] + total: int + page: int + page_size: int + + class ReportListResponse(BaseModel): - """Paginated reports linked to a job posting.""" + """Paginated reports for moderation or internal review.""" items: list[ReportResponse] total: int @@ -247,6 +315,19 @@ class EmployerClaimListResponse(BaseModel): page_size: int +class PublicEmployerResponseResponse(BaseModel): + """Public employer response without internal user identity.""" + + id: UUID + report_id: UUID + company_id: UUID + response_text: str + evidence_urls: list[str] + created_at: datetime + + model_config = {"from_attributes": True} + + class EmployerResponseResponse(BaseModel): """Employer response to an integrity report.""" @@ -274,6 +355,13 @@ def validate_evidence_urls(cls, urls: list[str]) -> list[str]: return [validate_http_https_url(url) for url in urls] +class PublicEmployerResponseListResponse(BaseModel): + """Public employer responses linked to a report.""" + + items: list[PublicEmployerResponseResponse] + total: int + + class EmployerResponseListResponse(BaseModel): """Employer responses linked to a report.""" diff --git a/backend/app/services/report_eligibility.py b/backend/app/services/report_eligibility.py new file mode 100644 index 0000000..35d492f --- /dev/null +++ b/backend/app/services/report_eligibility.py @@ -0,0 +1,48 @@ +"""Canonical moderation and score-eligibility rules for reports.""" + +from sqlalchemy.sql import ColumnElement + +from app.models.enums import ReportStatus +from app.models.report import Report + +PUBLIC_VISIBLE_STATUSES: frozenset[ReportStatus] = frozenset({ReportStatus.VERIFIED}) + +SCORE_ELIGIBLE_STATUSES: frozenset[ReportStatus] = frozenset({ReportStatus.VERIFIED}) + +ACTIVE_DUPLICATE_STATUSES: frozenset[ReportStatus] = frozenset( + { + ReportStatus.PENDING, + ReportStatus.VERIFIED, + ReportStatus.DISPUTED, + } +) + + +def is_publicly_visible(status: ReportStatus) -> bool: + """Return whether a report may be exposed through public unauthenticated APIs.""" + return status in PUBLIC_VISIBLE_STATUSES + + +def is_score_eligible(status: ReportStatus) -> bool: + """Return whether a report may influence public integrity or risk scores.""" + return status in SCORE_ELIGIBLE_STATUSES + + +def is_active_duplicate_status(status: ReportStatus) -> bool: + """Return whether a report status blocks materially duplicate submissions.""" + return status in ACTIVE_DUPLICATE_STATUSES + + +def public_visibility_filter() -> ColumnElement[bool]: + """SQLAlchemy filter for publicly visible reports.""" + return Report.status.in_(tuple(PUBLIC_VISIBLE_STATUSES)) + + +def score_eligibility_filter() -> ColumnElement[bool]: + """SQLAlchemy filter for score-eligible reports.""" + return Report.status.in_(tuple(SCORE_ELIGIBLE_STATUSES)) + + +def active_duplicate_filter() -> ColumnElement[bool]: + """SQLAlchemy filter for active duplicate-blocking report statuses.""" + return Report.status.in_(tuple(ACTIVE_DUPLICATE_STATUSES)) diff --git a/backend/app/services/reports.py b/backend/app/services/reports.py index 8f42120..6894d74 100644 --- a/backend/app/services/reports.py +++ b/backend/app/services/reports.py @@ -3,18 +3,52 @@ from uuid import UUID from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession -from app.exceptions import NotFoundError +from app.db_errors import raise_conflict_from_integrity_error +from app.exceptions import ConflictError, NotFoundError from app.models.company import Company -from app.models.enums import ReportStatus +from app.models.enums import ReportStatus, ReportType from app.models.job_posting import JobPosting from app.models.report import Report from app.models.user import User -from app.schemas import CreateReportRequest, ReportListResponse, ReportResponse +from app.schemas import ( + CreateReportRequest, + PublicReportListResponse, + PublicReportResponse, + ReportResponse, +) from app.services.audit import log_report_created +from app.services.report_eligibility import ( + active_duplicate_filter, + is_publicly_visible, + public_visibility_filter, + score_eligibility_filter, +) from app.services.scoring_pipeline import recalculate_for_job_posting_and_company +DUPLICATE_REPORT_DETAIL = "A similar report from this account is already active for this posting" + + +async def _find_active_duplicate_report( + session: AsyncSession, + *, + reporter_id: UUID, + job_posting_id: UUID, + report_type: ReportType, +) -> Report | None: + """Return an active duplicate report when one already exists.""" + result = await session.execute( + select(Report).where( + Report.reporter_id == reporter_id, + Report.job_posting_id == job_posting_id, + Report.report_type == report_type, + active_duplicate_filter(), + ) + ) + return result.scalar_one_or_none() + async def create_report( session: AsyncSession, @@ -22,7 +56,7 @@ async def create_report( payload: CreateReportRequest, reporter: User, ) -> Report: - """Create a report and recalculate related scores. + """Create a pending report without affecting public scores. Args: session: Active database session. @@ -34,6 +68,7 @@ async def create_report( Raises: NotFoundError: When the job posting does not exist. + ConflictError: When a materially duplicate active report exists. """ posting_result = await session.execute( select(JobPosting).where(JobPosting.id == payload.job_posting_id) @@ -42,6 +77,15 @@ async def create_report( if posting is None: raise NotFoundError("Job posting not found") + duplicate = await _find_active_duplicate_report( + session, + reporter_id=reporter.id, + job_posting_id=payload.job_posting_id, + report_type=payload.report_type, + ) + if duplicate is not None: + raise ConflictError(DUPLICATE_REPORT_DETAIL) + report = Report( job_posting_id=payload.job_posting_id, reporter_id=reporter.id, @@ -50,11 +94,11 @@ async def create_report( status=ReportStatus.PENDING, ) session.add(report) - await session.flush() - - company_result = await session.execute(select(Company).where(Company.id == posting.company_id)) - company = company_result.scalar_one() - company.report_count += 1 + try: + await session.flush() + except IntegrityError as exc: + await session.rollback() + raise_conflict_from_integrity_error(exc) await log_report_created( session, @@ -63,40 +107,89 @@ async def create_report( job_posting_id=report.job_posting_id, report_type=payload.report_type.value, ) - await recalculate_for_job_posting_and_company(session, posting.id, posting.company_id) await session.commit() await session.refresh(report) return report -async def get_report(session: AsyncSession, report_id: UUID) -> Report: - """Load a report by primary key. +async def _load_report(session: AsyncSession, report_id: UUID) -> Report | None: + """Load a report by primary key.""" + result = await session.execute(select(Report).where(Report.id == report_id)) + return result.scalar_one_or_none() + + +def _user_may_view_internal_report(report: Report, viewer: User | None) -> bool: + """Return whether an authenticated viewer may access internal report fields.""" + if viewer is None: + return False + if viewer.is_admin: + return True + return report.reporter_id == viewer.id + + +async def get_report( + session: AsyncSession, + report_id: UUID, + *, + viewer: User | None = None, +) -> Report: + """Load a report respecting public visibility rules. Args: session: Active database session. report_id: Report UUID. + viewer: Optional authenticated viewer. Returns: Report: Matching report record. Raises: - NotFoundError: When the report does not exist. + NotFoundError: When the report is not visible to the caller. """ - result = await session.execute(select(Report).where(Report.id == report_id)) - report = result.scalar_one_or_none() + report = await _load_report(session, report_id) if report is None: raise NotFoundError("Report not found") - return report + if _user_may_view_internal_report(report, viewer): + return report -async def list_reports_for_job_posting( + if is_publicly_visible(report.status): + return report + + raise NotFoundError("Report not found") + + +def _to_public_response(report: Report) -> PublicReportResponse: + """Convert a report entity to a public response schema.""" + return PublicReportResponse.model_validate(report) + + +def _to_internal_response(report: Report) -> ReportResponse: + """Convert a report entity to an internal response schema.""" + return ReportResponse.model_validate(report) + + +async def get_report_response( + session: AsyncSession, + report_id: UUID, + *, + viewer: User | None = None, +) -> PublicReportResponse | ReportResponse: + """Return the appropriate report schema for the caller.""" + report = await get_report(session, report_id, viewer=viewer) + if _user_may_view_internal_report(report, viewer): + return _to_internal_response(report) + return _to_public_response(report) + + +async def list_public_reports_for_job_posting( session: AsyncSession, job_posting_id: UUID, *, page: int, page_size: int, -) -> ReportListResponse: - """List reports linked to a job posting. +) -> PublicReportListResponse: + """List publicly visible reports linked to a job posting. Args: session: Active database session. @@ -105,7 +198,7 @@ async def list_reports_for_job_posting( page_size: Page size capped by the API layer. Returns: - ReportListResponse: Paginated reports for the posting. + PublicReportListResponse: Paginated public reports for the posting. Raises: NotFoundError: When the job posting does not exist. @@ -119,23 +212,62 @@ async def list_reports_for_job_posting( safe_page = max(page, 1) safe_page_size = max(page_size, 1) offset = (safe_page - 1) * safe_page_size + visibility = public_visibility_filter() total_result = await session.execute( - select(func.count()).select_from(Report).where(Report.job_posting_id == job_posting_id) + select(func.count()) + .select_from(Report) + .where(Report.job_posting_id == job_posting_id, visibility) ) total = int(total_result.scalar_one()) result = await session.execute( select(Report) - .where(Report.job_posting_id == job_posting_id) + .where(Report.job_posting_id == job_posting_id, visibility) .order_by(Report.created_at.desc()) .offset(offset) .limit(safe_page_size) ) - items = [ReportResponse.model_validate(report) for report in result.scalars().all()] - return ReportListResponse( + items = [_to_public_response(report) for report in result.scalars().all()] + return PublicReportListResponse( items=items, total=total, page=safe_page, page_size=safe_page_size, ) + + +async def adjust_company_report_count( + session: AsyncSession, + company_id: UUID, + *, + delta: int, +) -> None: + """Adjust a company's public report count by a signed delta.""" + company_result = await session.execute(select(Company).where(Company.id == company_id)) + company = company_result.scalar_one() + company.report_count = max(0, company.report_count + delta) + + +async def sync_company_report_count(session: AsyncSession, company_id: UUID) -> None: + """Recompute a company's public report count from score-eligible reports.""" + count_result = await session.execute( + select(func.count()) + .select_from(Report) + .join(JobPosting, Report.job_posting_id == JobPosting.id) + .where(JobPosting.company_id == company_id, score_eligibility_filter()) + ) + eligible_count = int(count_result.scalar_one()) + company_result = await session.execute(select(Company).where(Company.id == company_id)) + company = company_result.scalar_one() + company.report_count = eligible_count + + +async def recalculate_scores_if_needed( + session: AsyncSession, + *, + job_posting_id: UUID, + company_id: UUID, +) -> None: + """Recalculate scores from eligible records only.""" + await recalculate_for_job_posting_and_company(session, job_posting_id, company_id) diff --git a/backend/tests/test_reports.py b/backend/tests/test_reports.py index 7468d19..53d6a89 100644 --- a/backend/tests/test_reports.py +++ b/backend/tests/test_reports.py @@ -1,8 +1,6 @@ """Report API tests.""" -from uuid import uuid4 - -from uuid import UUID +from uuid import UUID, uuid4 import pytest from httpx import AsyncClient @@ -18,6 +16,40 @@ ) +async def _create_report( + client: AsyncClient, + job_posting_id: str, + auth_headers: dict[str, str], + *, + report_type: ReportType = ReportType.STALE_POSTING, +) -> str: + """Create a report and return its identifier.""" + response = await client.post( + "/api/v1/reports", + json={ + "job_posting_id": job_posting_id, + "report_type": report_type.value, + "description": REPORT_DESCRIPTION, + }, + headers=auth_headers, + ) + assert response.status_code == 201 + return str(response.json()["id"]) + + +async def _verify_report( + client: AsyncClient, + report_id: str, + admin_auth_headers: dict[str, str], +) -> None: + """Verify a report through the moderation API.""" + response = await client.post( + f"/api/v1/moderation/reports/{report_id}/verify", + headers=admin_auth_headers, + ) + assert response.status_code == 200 + + @pytest.mark.asyncio async def test_create_report_requires_auth( client: AsyncClient, sample_job_posting: JobPosting @@ -41,17 +73,10 @@ async def test_create_report_success( auth_headers: dict[str, str], ) -> None: """Authenticated users should be able to submit reports.""" - response = await client.post( - "/api/v1/reports", - json={ - "job_posting_id": str(sample_job_posting.id), - "report_type": ReportType.STALE_POSTING.value, - "description": REPORT_DESCRIPTION, - }, - headers=auth_headers, - ) - assert response.status_code == 201 + report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + response = await client.get(f"/api/v1/reports/{report_id}", headers=auth_headers) body = response.json() + assert response.status_code == 200 assert body["job_posting_id"] == str(sample_job_posting.id) assert body["report_type"] == ReportType.STALE_POSTING.value assert body["status"] == "pending" @@ -76,28 +101,37 @@ async def test_create_report_unknown_job_posting_returns_404( @pytest.mark.asyncio -async def test_get_report_success( +async def test_get_report_success_for_reporter( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], ) -> None: - """Report detail should return a submitted report.""" - create_response = await client.post( - "/api/v1/reports", - json={ - "job_posting_id": str(sample_job_posting.id), - "report_type": ReportType.NO_RESPONSE.value, - "description": REPORT_DESCRIPTION, - }, - headers=auth_headers, + """Reporters should be able to read their own pending reports.""" + report_id = await _create_report( + client, + str(sample_job_posting.id), + auth_headers, + report_type=ReportType.NO_RESPONSE, ) - report_id = create_response.json()["id"] - response = await client.get(f"/api/v1/reports/{report_id}") + response = await client.get(f"/api/v1/reports/{report_id}", headers=auth_headers) assert response.status_code == 200 assert response.json()["id"] == report_id +@pytest.mark.asyncio +async def test_get_pending_report_is_hidden_from_public( + client: AsyncClient, + sample_job_posting: JobPosting, + auth_headers: dict[str, str], +) -> None: + """Pending reports should not be readable without authorization.""" + report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + + response = await client.get(f"/api/v1/reports/{report_id}") + assert response.status_code == 404 + + @pytest.mark.asyncio async def test_get_report_returns_404_for_missing_id(client: AsyncClient) -> None: """Unknown reports should return 404.""" @@ -110,27 +144,27 @@ async def test_list_reports_filtered_by_job_posting_id( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: - """Report list should filter by job posting identifier.""" - await client.post( - "/api/v1/reports", - json={ - "job_posting_id": str(sample_job_posting.id), - "report_type": ReportType.REPOST_LOOP.value, - "description": REPORT_DESCRIPTION, - }, - headers=auth_headers, + """Public report list should include only verified reports.""" + report_id = await _create_report( + client, + str(sample_job_posting.id), + auth_headers, + report_type=ReportType.REPOST_LOOP, ) + await _verify_report(client, report_id, admin_auth_headers) response = await client.get( f"/api/v1/reports?job_posting_id={sample_job_posting.id}", ) assert response.status_code == 200 body = response.json() - assert body["total"] >= 1 + assert body["total"] == 1 assert body["page"] == 1 assert body["page_size"] == 20 - assert all(item["job_posting_id"] == str(sample_job_posting.id) for item in body["items"]) + assert body["items"][0]["job_posting_id"] == str(sample_job_posting.id) + assert "reporter_id" not in body["items"][0] @pytest.mark.asyncio @@ -138,25 +172,28 @@ async def test_list_reports_supports_pagination( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: - """Report list should honor page and page_size query parameters.""" + """Public report list should honor page and page_size query parameters.""" + report_ids: list[str] = [] for report_type in (ReportType.REPOST_LOOP, ReportType.NO_RESPONSE, ReportType.STALE_POSTING): - await client.post( - "/api/v1/reports", - json={ - "job_posting_id": str(sample_job_posting.id), - "report_type": report_type.value, - "description": REPORT_DESCRIPTION, - }, - headers=auth_headers, + report_ids.append( + await _create_report( + client, + str(sample_job_posting.id), + auth_headers, + report_type=report_type, + ) ) + for report_id in report_ids: + await _verify_report(client, report_id, admin_auth_headers) response = await client.get( f"/api/v1/reports?job_posting_id={sample_job_posting.id}&page=1&page_size=2", ) assert response.status_code == 200 body = response.json() - assert body["total"] >= 3 + assert body["total"] == 3 assert body["page"] == 1 assert body["page_size"] == 2 assert len(body["items"]) == 2 @@ -170,16 +207,7 @@ async def test_create_report_writes_audit_row( auth_headers: dict[str, str], ) -> None: """Report creation should write an audit log entry.""" - response = await client.post( - "/api/v1/reports", - json={ - "job_posting_id": str(sample_job_posting.id), - "report_type": ReportType.STALE_POSTING.value, - "description": REPORT_DESCRIPTION, - }, - headers=auth_headers, - ) - report_id = response.json()["id"] + report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) audit_result = await db_session.execute( select(AuditLog).where( diff --git a/docs/moderation-model.md b/docs/moderation-model.md index 96d77f6..dfb2587 100644 --- a/docs/moderation-model.md +++ b/docs/moderation-model.md @@ -18,12 +18,14 @@ Future batches may require evidence attachments before promotion or verification | State | Meaning | | ----- | ------- | -| `pending` | Received; default on create | -| `verified` | Moderator or trusted process confirmed the report | -| `dismissed` | Report rejected or insufficient | -| `disputed` | Employer or reviewer challenged the report | +| `pending` | Received; default on create; not publicly visible | +| `verified` | Moderator confirmed; eligible for public display and scoring | +| `dismissed` | Report rejected or insufficient; excluded from public APIs and scoring | +| `disputed` | Employer challenged the report; not publicly visible until verified | -Moderation APIs that transition between these states are implemented in Batch 6. New reports always start as `pending`. +New reports always start as `pending`. Public unauthenticated report APIs return only `verified` reports and omit internal reporter identifiers. + +Score recalculation uses only `verified` reports. Pending report creation does not change public company or posting scores. ## Abuse prevention @@ -41,7 +43,13 @@ The platform must prevent: Current technical controls include: - authenticated report and vote submissions +- public visibility limited to verified reports +- duplicate active report rejection per reporter, posting, and report type (`409` on conflict) +- report submission rate limiting separate from auth rate limits - duplicate vote protection per user and report (`409` on conflict, including race-safe handling) +- duplicate employer response protection per employer user and report +- employer response rate scoring based on distinct eligible reports, not raw response rows +- refresh-token rotation and logout revocation - auth endpoint rate limiting - transparent scoring instead of accusatory labels From 2207f6cd04a75f229f9d75e856944956ee13130c Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 02/12] fix: protect score integrity from ineligible and duplicate reports Centralize score eligibility on verified reports, block duplicate active submissions, add report rate limits, and add database-backed duplicate protection. --- .../003_audit_remediation_constraints.py | 59 +++++ backend/app/config.py | 1 + backend/app/services/moderation.py | 16 +- backend/app/services/rate_limit.py | 30 +++ backend/app/services/scoring_pipeline.py | 166 ++++--------- backend/tests/test_audit_remediation.py | 230 ++++++++++++++++++ backend/tests/test_migrations.py | 2 +- 7 files changed, 379 insertions(+), 125 deletions(-) create mode 100644 backend/alembic/versions/003_audit_remediation_constraints.py create mode 100644 backend/tests/test_audit_remediation.py diff --git a/backend/alembic/versions/003_audit_remediation_constraints.py b/backend/alembic/versions/003_audit_remediation_constraints.py new file mode 100644 index 0000000..11d3983 --- /dev/null +++ b/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)") diff --git a/backend/app/config.py b/backend/app/config.py index 0245f0f..b19b6a6 100644 --- a/backend/app/config.py +++ b/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 diff --git a/backend/app/services/moderation.py b/backend/app/services/moderation.py index a1fdf75..616cc28 100644 --- a/backend/app/services/moderation.py +++ b/backend/app/services/moderation.py @@ -12,7 +12,7 @@ from app.models.user import User from app.schemas import ReportListResponse, ReportResponse from app.services.audit import log_report_dismissed, log_report_verified -from app.services.scoring_pipeline import recalculate_for_job_posting_and_company +from app.services.reports import recalculate_scores_if_needed, sync_company_report_count _ADMIN_VERIFY_FROM = frozenset({ReportStatus.PENDING, ReportStatus.DISPUTED}) _ADMIN_DISMISS_FROM = frozenset({ReportStatus.PENDING, ReportStatus.DISPUTED}) @@ -83,11 +83,12 @@ async def verify_report( job_posting_id=report.job_posting_id, previous_status=previous_status.value, ) - await recalculate_for_job_posting_and_company( + await recalculate_scores_if_needed( session, - report.job_posting_id, - report.job_posting.company_id, + job_posting_id=report.job_posting_id, + company_id=report.job_posting.company_id, ) + await sync_company_report_count(session, report.job_posting.company_id) await session.commit() await session.refresh(report) return report @@ -116,11 +117,12 @@ async def dismiss_report( previous_status=previous_status.value, reason=reason, ) - await recalculate_for_job_posting_and_company( + await recalculate_scores_if_needed( session, - report.job_posting_id, - report.job_posting.company_id, + job_posting_id=report.job_posting_id, + company_id=report.job_posting.company_id, ) + await sync_company_report_count(session, report.job_posting.company_id) await session.commit() await session.refresh(report) return report diff --git a/backend/app/services/rate_limit.py b/backend/app/services/rate_limit.py index 7d63dd2..17ee042 100644 --- a/backend/app/services/rate_limit.py +++ b/backend/app/services/rate_limit.py @@ -7,6 +7,7 @@ from app.redis_client import RedisClient RATE_LIMIT_WINDOW_SECONDS = 60 +REPORT_RATE_LIMIT_WINDOW_SECONDS = 3600 _RATE_LIMIT_SCRIPT = """ local current = redis.call('INCR', KEYS[1]) @@ -30,6 +31,35 @@ def rate_limit_key(route: str, client_ip: str) -> str: return f"auth_rl:{route}:{client_ip}" +async def check_report_submission_rate_limit( + redis: RedisClient, + user_id: str, + client_ip: str, + settings: Settings, +) -> None: + """Increment and enforce the report submission rate limit. + + Args: + redis: Active Redis client. + user_id: Authenticated reporter UUID string. + client_ip: Client IP address. + settings: Application settings. + + Raises: + RateLimitError: When the client exceeds the configured limit. + """ + key = f"report_rl:{user_id}:{client_ip}" + raw_count = await redis.eval( # type: ignore[no-untyped-call] + _RATE_LIMIT_SCRIPT, + 1, + key, + str(REPORT_RATE_LIMIT_WINDOW_SECONDS), + ) + count = int(cast(int, raw_count)) + if count > settings.report_submission_rate_limit_per_hour: + raise RateLimitError() + + async def check_auth_rate_limit( redis: RedisClient, route: str, diff --git a/backend/app/services/scoring_pipeline.py b/backend/app/services/scoring_pipeline.py index 3dbf7f6..2821a68 100644 --- a/backend/app/services/scoring_pipeline.py +++ b/backend/app/services/scoring_pipeline.py @@ -4,15 +4,17 @@ from decimal import Decimal from uuid import UUID -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.models.company import Company +from app.models.employer_response import EmployerResponse from app.models.enums import ReportType, SnapshotEntityType from app.models.job_posting import JobPosting from app.models.report import Report from app.models.score_snapshot import ScoreSnapshot +from app.services.report_eligibility import is_score_eligible, score_eligibility_filter from app.services.scoring import ( CompanyIntegrityInputs, JobGhostRiskInputs, @@ -24,67 +26,57 @@ def _posting_age_days(detected_at: datetime) -> int: - """Return whole days since a posting was first detected. - - Args: - detected_at: Posting detection timestamp. - - Returns: - int: Non-negative day count. - """ + """Return whole days since a posting was first detected.""" now = datetime.now(tz=UTC) if detected_at.tzinfo is None: detected_at = detected_at.replace(tzinfo=UTC) return max(0, (now - detected_at).days) -async def _reports_for_posting(session: AsyncSession, job_posting_id: UUID) -> list[Report]: - """Load reports linked to a job posting. +def _eligible_reports(reports: list[Report]) -> list[Report]: + """Filter reports to score-eligible moderation states.""" + return [report for report in reports if is_score_eligible(report.status)] - Args: - session: Active database session. - job_posting_id: Job posting UUID. - Returns: - list[Report]: Reports for the posting. - """ - result = await session.execute(select(Report).where(Report.job_posting_id == job_posting_id)) +async def _reports_for_posting(session: AsyncSession, job_posting_id: UUID) -> list[Report]: + """Load score-eligible reports linked to a job posting.""" + result = await session.execute( + select(Report).where( + Report.job_posting_id == job_posting_id, + score_eligibility_filter(), + ) + ) return list(result.scalars().all()) async def _reports_for_company(session: AsyncSession, company_id: UUID) -> list[Report]: - """Load reports linked to postings owned by a company. - - Args: - session: Active database session. - company_id: Company UUID. - - Returns: - list[Report]: Reports across the company's postings. - """ + """Load score-eligible reports linked to postings owned by a company.""" result = await session.execute( select(Report) .join(JobPosting, Report.job_posting_id == JobPosting.id) - .where(JobPosting.company_id == company_id) + .where(JobPosting.company_id == company_id, score_eligibility_filter()) ) return list(result.scalars().all()) +async def _eligible_responded_report_count(session: AsyncSession, company_id: UUID) -> int: + """Count distinct eligible reports that received an employer response.""" + result = await session.execute( + select(func.count(func.distinct(EmployerResponse.report_id))) + .select_from(EmployerResponse) + .join(Report, EmployerResponse.report_id == Report.id) + .join(JobPosting, Report.job_posting_id == JobPosting.id) + .where(JobPosting.company_id == company_id, score_eligibility_filter()) + ) + return int(result.scalar_one()) + + def _build_job_inputs( posting: JobPosting, company: Company, reports: list[Report], ) -> JobGhostRiskInputs: - """Build job ghost risk inputs from ORM entities. - - Args: - posting: Job posting entity. - company: Parent company entity. - reports: Reports linked to the posting. - - Returns: - JobGhostRiskInputs: Normalized scoring inputs. - """ + """Build job ghost risk inputs from ORM entities.""" report_statuses = [report.status for report in reports] return JobGhostRiskInputs( posting_age_days=_posting_age_days(posting.detected_at), @@ -102,23 +94,20 @@ def _build_job_inputs( ) -def _build_company_inputs(company: Company, reports: list[Report]) -> CompanyIntegrityInputs: - """Build company integrity inputs from ORM entities. - - Args: - company: Company entity. - reports: Reports linked to company postings. - - Returns: - CompanyIntegrityInputs: Normalized scoring inputs. - """ +def _build_company_inputs( + company: Company, + reports: list[Report], + responded_report_count: int, +) -> CompanyIntegrityInputs: + """Build company integrity inputs from ORM entities.""" report_statuses = [report.status for report in reports] + eligible_total = len(reports) return CompanyIntegrityInputs( total_postings=company.total_postings, total_hires=company.total_hires, verified_report_count=count_verified_reports(report_statuses), - total_reports=company.report_count, - employer_response_count=len(company.employer_responses), + total_reports=eligible_total, + employer_response_count=responded_report_count, verified_status=company.verified_status, average_days_to_fill=None, recruiter_follow_through_rate=0.0, @@ -130,15 +119,7 @@ async def calculate_job_posting_risk_score( session: AsyncSession, job_posting_id: UUID, ) -> ScoreResult: - """Calculate a job posting risk score without persisting changes. - - Args: - session: Active database session. - job_posting_id: Job posting UUID. - - Returns: - ScoreResult: Calculated score and breakdown. - """ + """Calculate a job posting risk score without persisting changes.""" posting = await _load_job_posting(session, job_posting_id) reports = await _reports_for_posting(session, job_posting_id) inputs = _build_job_inputs(posting, posting.company, reports) @@ -149,18 +130,11 @@ async def calculate_company_integrity_score_result( session: AsyncSession, company_id: UUID, ) -> ScoreResult: - """Calculate a company integrity score without persisting changes. - - Args: - session: Active database session. - company_id: Company UUID. - - Returns: - ScoreResult: Calculated score and breakdown. - """ + """Calculate a company integrity score without persisting changes.""" company = await _load_company(session, company_id) reports = await _reports_for_company(session, company_id) - inputs = _build_company_inputs(company, reports) + responded_count = await _eligible_responded_report_count(session, company_id) + inputs = _build_company_inputs(company, reports, responded_count) return calculate_company_integrity_score(inputs) @@ -168,15 +142,7 @@ async def recalculate_job_posting_score( session: AsyncSession, job_posting_id: UUID, ) -> ScoreResult: - """Recalculate, persist, and snapshot a job posting risk score. - - Args: - session: Active database session. - job_posting_id: Job posting UUID. - - Returns: - ScoreResult: Updated score and breakdown. - """ + """Recalculate, persist, and snapshot a job posting risk score.""" result = await calculate_job_posting_risk_score(session, job_posting_id) posting = await _load_job_posting(session, job_posting_id) posting.ghost_risk_score = Decimal(str(result.score)) @@ -195,15 +161,7 @@ async def recalculate_company_score( session: AsyncSession, company_id: UUID, ) -> ScoreResult: - """Recalculate, persist, and snapshot a company integrity score. - - Args: - session: Active database session. - company_id: Company UUID. - - Returns: - ScoreResult: Updated score and breakdown. - """ + """Recalculate, persist, and snapshot a company integrity score.""" result = await calculate_company_integrity_score_result(session, company_id) company = await _load_company(session, company_id) company.integrity_score = Decimal(str(result.score)) @@ -223,27 +181,13 @@ async def recalculate_for_job_posting_and_company( job_posting_id: UUID, company_id: UUID, ) -> None: - """Recalculate scores for a posting and its parent company. - - Args: - session: Active database session. - job_posting_id: Job posting UUID. - company_id: Company UUID. - """ + """Recalculate scores for a posting and its parent company.""" await recalculate_job_posting_score(session, job_posting_id) await recalculate_company_score(session, company_id) async def _load_job_posting(session: AsyncSession, job_posting_id: UUID) -> JobPosting: - """Load a job posting with its company relationship. - - Args: - session: Active database session. - job_posting_id: Job posting UUID. - - Returns: - JobPosting: Loaded posting with company attached. - """ + """Load a job posting with its company relationship.""" result = await session.execute( select(JobPosting) .options(selectinload(JobPosting.company)) @@ -253,18 +197,6 @@ async def _load_job_posting(session: AsyncSession, job_posting_id: UUID) -> JobP async def _load_company(session: AsyncSession, company_id: UUID) -> Company: - """Load a company with employer responses for scoring inputs. - - Args: - session: Active database session. - company_id: Company UUID. - - Returns: - Company: Loaded company with responses attached. - """ - result = await session.execute( - select(Company) - .options(selectinload(Company.employer_responses)) - .where(Company.id == company_id) - ) + """Load a company for scoring inputs.""" + result = await session.execute(select(Company).where(Company.id == company_id)) return result.scalar_one() diff --git a/backend/tests/test_audit_remediation.py b/backend/tests/test_audit_remediation.py new file mode 100644 index 0000000..ec78950 --- /dev/null +++ b/backend/tests/test_audit_remediation.py @@ -0,0 +1,230 @@ +"""Regression tests for independent deep-audit remediation.""" + +from uuid import uuid4 + +import pytest +from httpx import AsyncClient +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.company import Company +from app.models.enums import ReportType +from app.models.job_posting import JobPosting +from app.models.report import Report +from app.security.password_policy import BCRYPT_MAX_PASSWORD_BYTES, validate_password_byte_length +from app.services.job_url_validation import ( + JobSourceProvider, + extract_employer_identity, + validate_job_url, +) +from app.services.sheet_import import _map_posting_source, check_eligibility, resolve_column_map +from tests.test_reports import REPORT_DESCRIPTION, _create_report, _verify_report + +DEFAULT_PASSWORD = "StrongPass123!" + + +def test_password_byte_limit_rejects_multibyte_overflow() -> None: + """Passwords above bcrypt's UTF-8 byte boundary should be rejected.""" + password = "a" * 70 + "éé" + assert len(password.encode("utf-8")) > BCRYPT_MAX_PASSWORD_BYTES + with pytest.raises(ValueError, match="UTF-8 bytes"): + validate_password_byte_length(password) + + +@pytest.mark.asyncio +async def test_pending_report_does_not_change_company_score( + client: AsyncClient, + db_session: AsyncSession, + sample_job_posting: JobPosting, + auth_headers: dict[str, str], +) -> None: + """Creating a pending report must not change the public company score.""" + company = await db_session.get(Company, sample_job_posting.company_id) + assert company is not None + before = float(company.integrity_score) + + await _create_report( + client, + str(sample_job_posting.id), + auth_headers, + report_type=ReportType.NO_RESPONSE, + ) + await db_session.refresh(company) + assert float(company.integrity_score) == before + + +@pytest.mark.asyncio +async def test_duplicate_report_submission_returns_409( + client: AsyncClient, + sample_job_posting: JobPosting, + auth_headers: dict[str, str], +) -> None: + """Duplicate active reports from the same account should conflict.""" + await _create_report( + client, + str(sample_job_posting.id), + auth_headers, + report_type=ReportType.NO_RESPONSE, + ) + duplicate = await client.post( + "/api/v1/reports", + json={ + "job_posting_id": str(sample_job_posting.id), + "report_type": ReportType.NO_RESPONSE.value, + "description": REPORT_DESCRIPTION, + }, + headers=auth_headers, + ) + assert duplicate.status_code == 409 + + +@pytest.mark.asyncio +async def test_verified_report_updates_verification_votes( + client: AsyncClient, + db_session: AsyncSession, + sample_job_posting: JobPosting, + auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], +) -> None: + """Votes on verified reports should update report.verification_votes.""" + suffix = uuid4().hex[:8] + voter_register = await client.post( + "/api/v1/auth/register", + json={ + "email": f"voter-{suffix}@example.com", + "username": f"voter_{suffix}", + "password": DEFAULT_PASSWORD, + }, + ) + voter_headers = {"Authorization": f"Bearer {voter_register.json()['access_token']}"} + + report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + await _verify_report(client, report_id, admin_auth_headers) + + vote = await client.post( + f"/api/v1/reports/{report_id}/votes", + json={"vote": "up"}, + headers=voter_headers, + ) + assert vote.status_code == 201 + + report = await db_session.scalar(select(Report).where(Report.id == report_id)) + assert report is not None + assert report.verification_votes == 1 + + +@pytest.mark.asyncio +async def test_register_rejects_case_equivalent_email( + client: AsyncClient, +) -> None: + """Email uniqueness should be case-insensitive.""" + suffix = uuid4().hex[:8] + first = await client.post( + "/api/v1/auth/register", + json={ + "email": f"Case-{suffix}@Example.com", + "username": f"case_user_{suffix}", + "password": DEFAULT_PASSWORD, + }, + ) + assert first.status_code == 200 + + duplicate = await client.post( + "/api/v1/auth/register", + json={ + "email": f"case-{suffix}@example.com", + "username": f"case_user2_{suffix}", + "password": DEFAULT_PASSWORD, + }, + ) + assert duplicate.status_code == 409 + + +def test_supported_ats_providers_do_not_map_to_other() -> None: + """Supported ATS providers should not be classified as OTHER in sheet import.""" + urls = { + JobSourceProvider.ICIMS: "https://careers-acme.icims.com/jobs/1234/job", + JobSourceProvider.WORKABLE: "https://apply.workable.com/acme/j/ABC123DEF4", + JobSourceProvider.BAMBOOHR: "https://acme.bamboohr.com/careers/42", + JobSourceProvider.TALEO: "https://acme.taleo.net/careersection/jobdetail.ftl?job=123", + JobSourceProvider.JOBVITE: "https://jobs.jobvite.com/acme/job/oUZbYfwF", + } + for provider, url in urls.items(): + normalized = validate_job_url(url).normalized_url + assert normalized is not None + assert validate_job_url(url).provider == provider + assert _map_posting_source(normalized).value == "company_site" + + +def test_extract_employer_identity_avoids_shared_ats_domain() -> None: + """Hosted ATS URLs should not use the shared platform host as company domain.""" + normalized = validate_job_url("https://boards.greenhouse.io/acme/jobs/1234567").normalized_url + assert normalized is not None + identity = extract_employer_identity(normalized) + assert identity.tenant_identifier == "acme" + assert identity.company_domain_hint is None + + +def test_sheet_import_requires_reviewer_and_reviewed_at() -> None: + """Import-ready rows must include reviewer and reviewed_at.""" + headers = [ + "Timestamp", + "Job posting URL", + "Company name", + "Job title", + "Location or remote", + "Date seen", + "Why do you suspect this is a ghost job?", + "Has the company responded?", + "Consent", + "Evidence links", + "Optional contact email", + "review_status", + "reviewer", + "reviewed_at", + "decline_reason_code", + "duplicate_of", + "notes", + "pii_redacted", + "import_ready", + "escalation_level", + ] + column_map = resolve_column_map(headers) + row = { + "Timestamp": "2026-07-01 12:00:00", + "Job posting URL": "https://boards.greenhouse.io/acme/jobs/1234567", + "Company name": "Acme Corp", + "Job title": "Software Engineer", + "Location or remote": "Remote", + "Date seen": "2026-06-30", + "Why do you suspect this is a ghost job?": ( + "Listing has been open for months with no hiring updates." + ), + "Has the company responded?": "No", + "Consent": "I confirm this submission is made in good faith.", + "Evidence links": "", + "Optional contact email": "", + "review_status": "approved_for_import", + "reviewer": "", + "reviewed_at": "", + "decline_reason_code": "", + "duplicate_of": "", + "notes": "", + "pii_redacted": "yes", + "import_ready": "yes", + "escalation_level": "none", + } + result = check_eligibility(row, column_map) + assert result.eligible is False + assert result.reason_code == "missing_reviewer" + + +@pytest.mark.asyncio +async def test_readiness_endpoint_reports_dependency_status(client: AsyncClient) -> None: + """Readiness endpoint should report PostgreSQL and Redis status.""" + response = await client.get("/health/ready") + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ready" + assert body["checks"]["postgres"] == "ok" + assert body["checks"]["redis"] == "ok" diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index ac05017..8d1dd42 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -62,7 +62,7 @@ async def test_alembic_upgrade_head_records_revision() -> None: result = await connection.execute(text("SELECT version_num FROM alembic_version")) version = result.scalar_one() await engine.dispose() - assert version == "002_employer_claim_constraints" + assert version == "003_audit_remediation" @pytest.mark.asyncio From 20ade2433aab44c6d60e4fa34874116c3ff87bc2 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 03/12] fix: prevent employer response score inflation Reject duplicate employer responses per account, count distinct eligible reports in scoring, and hide responses for non-public reports. --- backend/app/services/employer_responses.py | 42 ++++++++++++++++------ backend/tests/test_employer_responses.py | 24 +++++++++---- 2 files changed, 48 insertions(+), 18 deletions(-) diff --git a/backend/app/services/employer_responses.py b/backend/app/services/employer_responses.py index 6f0d679..cf83126 100644 --- a/backend/app/services/employer_responses.py +++ b/backend/app/services/employer_responses.py @@ -3,23 +3,28 @@ from uuid import UUID from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload -from app.exceptions import ForbiddenError, NotFoundError +from app.db_errors import raise_conflict_from_integrity_error +from app.exceptions import ConflictError, ForbiddenError, NotFoundError from app.models.employer_response import EmployerResponse from app.models.enums import ReportStatus from app.models.report import Report from app.models.user import User from app.schemas import ( CreateEmployerResponseRequest, - EmployerResponseListResponse, - EmployerResponseResponse, + PublicEmployerResponseListResponse, + PublicEmployerResponseResponse, ) from app.services.audit import log_employer_response_created +from app.services.report_eligibility import is_publicly_visible +from app.services.reports import sync_company_report_count from app.services.scoring_pipeline import recalculate_for_job_posting_and_company _DISPUTE_FROM = frozenset({ReportStatus.PENDING, ReportStatus.VERIFIED}) +_DUPLICATE_RESPONSE_DETAIL = "An employer response from this account already exists for this report" async def _get_report_with_posting(session: AsyncSession, report_id: UUID) -> Report: @@ -51,6 +56,15 @@ async def create_employer_response( company_id = report.job_posting.company_id _ensure_verified_employer_for_company(employer, company_id) + existing = await session.execute( + select(EmployerResponse.id).where( + EmployerResponse.report_id == report_id, + EmployerResponse.user_id == employer.id, + ) + ) + if existing.scalar_one_or_none() is not None: + raise ConflictError(_DUPLICATE_RESPONSE_DETAIL) + previous_status = report.status if report.status in _DISPUTE_FROM: report.status = ReportStatus.DISPUTED @@ -63,7 +77,11 @@ async def create_employer_response( evidence_urls=payload.evidence_urls, ) session.add(response) - await session.flush() + try: + await session.flush() + except IntegrityError as exc: + await session.rollback() + raise_conflict_from_integrity_error(exc) await log_employer_response_created( session, @@ -75,19 +93,20 @@ async def create_employer_response( new_status=report.status.value, ) await recalculate_for_job_posting_and_company(session, report.job_posting_id, company_id) + await sync_company_report_count(session, company_id) await session.commit() await session.refresh(response) return response -async def list_employer_responses( +async def list_public_employer_responses( session: AsyncSession, *, report_id: UUID, -) -> EmployerResponseListResponse: - """Return employer responses linked to a report.""" - report_result = await session.execute(select(Report.id).where(Report.id == report_id)) - if report_result.scalar_one_or_none() is None: +) -> PublicEmployerResponseListResponse: + """Return employer responses only when the parent report is publicly visible.""" + report = await _get_report_with_posting(session, report_id) + if not is_publicly_visible(report.status): raise NotFoundError("Report not found") total_result = await session.execute( @@ -103,6 +122,7 @@ async def list_employer_responses( .order_by(EmployerResponse.created_at.asc()) ) items = [ - EmployerResponseResponse.model_validate(response) for response in result.scalars().all() + PublicEmployerResponseResponse.model_validate(response) + for response in result.scalars().all() ] - return EmployerResponseListResponse(items=items, total=total) + return PublicEmployerResponseListResponse(items=items, total=total) diff --git a/backend/tests/test_employer_responses.py b/backend/tests/test_employer_responses.py index 5232229..972b967 100644 --- a/backend/tests/test_employer_responses.py +++ b/backend/tests/test_employer_responses.py @@ -185,19 +185,19 @@ async def test_create_employer_response_disputes_pending_report( body = response.json() assert body["report_id"] == report_id - report = await client.get(f"/api/v1/reports/{report_id}") + report = await client.get(f"/api/v1/reports/{report_id}", headers=auth_headers) assert report.status_code == 200 assert report.json()["status"] == ReportStatus.DISPUTED.value @pytest.mark.asyncio -async def test_create_multiple_employer_responses_allowed( +async def test_create_duplicate_employer_response_returns_409( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], employer_auth_headers: dict[str, str], ) -> None: - """Multiple employer responses should be allowed on the same report.""" + """Duplicate employer responses from the same account should return 409.""" report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) first = await client.post( f"/api/v1/reports/{report_id}/responses", @@ -211,18 +211,18 @@ async def test_create_multiple_employer_responses_allowed( json={"response_text": "We posted an additional clarification for candidates."}, headers=employer_auth_headers, ) - assert second.status_code == 201 - assert first.json()["id"] != second.json()["id"] + assert second.status_code == 409 @pytest.mark.asyncio -async def test_list_employer_responses_is_public( +async def test_list_employer_responses_requires_public_report( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], employer_auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: - """Employer responses should be readable without authentication.""" + """Employer responses should be readable only for publicly visible reports.""" report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) create = await client.post( f"/api/v1/reports/{report_id}/responses", @@ -231,11 +231,21 @@ async def test_list_employer_responses_is_public( ) assert create.status_code == 201 + hidden = await client.get(f"/api/v1/reports/{report_id}/responses") + assert hidden.status_code == 404 + + verify = await client.post( + f"/api/v1/moderation/reports/{report_id}/verify", + headers=admin_auth_headers, + ) + assert verify.status_code == 200 + response = await client.get(f"/api/v1/reports/{report_id}/responses") assert response.status_code == 200 body = response.json() assert body["total"] == 1 assert body["items"][0]["response_text"] == RESPONSE_TEXT + assert "user_id" not in body["items"][0] @pytest.mark.asyncio From f6763086447f01e904c3444546307d0e45cd71b9 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 04/12] fix: connect community votes to report verification scoring Sync verification_votes from persisted upvotes and restrict voting to publicly visible reports. --- backend/app/services/votes.py | 26 ++++++++++++-- backend/tests/test_votes.py | 68 +++++++++++++++++++++++++++++++---- 2 files changed, 84 insertions(+), 10 deletions(-) diff --git a/backend/app/services/votes.py b/backend/app/services/votes.py index 4715e06..6093cc9 100644 --- a/backend/app/services/votes.py +++ b/backend/app/services/votes.py @@ -2,21 +2,36 @@ from uuid import UUID -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.db_errors import raise_conflict_from_integrity_error -from app.exceptions import ConflictError, NotFoundError +from app.exceptions import ConflictError, NotFoundError, ValidationError +from app.models.enums import VoteValue from app.models.job_posting import JobPosting from app.models.report import Report from app.models.user import User from app.models.vote import Vote from app.schemas import CreateVoteRequest from app.services.audit import log_vote_created +from app.services.report_eligibility import is_publicly_visible from app.services.scoring_pipeline import recalculate_for_job_posting_and_company +async def _sync_verification_votes(session: AsyncSession, report_id: UUID) -> None: + """Reconcile report.verification_votes from persisted vote records.""" + upvote_result = await session.execute( + select(func.count()) + .select_from(Vote) + .where(Vote.report_id == report_id, Vote.vote == VoteValue.UP) + ) + upvotes = int(upvote_result.scalar_one()) + report_result = await session.execute(select(Report).where(Report.id == report_id)) + report = report_result.scalar_one() + report.verification_votes = upvotes + + async def create_vote( session: AsyncSession, *, @@ -24,7 +39,7 @@ async def create_vote( payload: CreateVoteRequest, voter: User, ) -> Vote: - """Create a vote and recalculate related scores. + """Create a vote, sync verification totals, and recalculate related scores. Args: session: Active database session. @@ -37,12 +52,15 @@ async def create_vote( Raises: NotFoundError: When the report does not exist. + ValidationError: When the report is not publicly votable. ConflictError: When the user already voted on the report. """ report_result = await session.execute(select(Report).where(Report.id == report_id)) report = report_result.scalar_one_or_none() if report is None: raise NotFoundError("Report not found") + if not is_publicly_visible(report.status): + raise ValidationError("Votes are only accepted on publicly visible reports") existing_vote = await session.execute( select(Vote.id).where(Vote.report_id == report_id, Vote.user_id == voter.id) @@ -58,6 +76,8 @@ async def create_vote( await session.rollback() raise_conflict_from_integrity_error(exc) + await _sync_verification_votes(session, report_id) + posting_result = await session.execute( select(JobPosting).where(JobPosting.id == report.job_posting_id) ) diff --git a/backend/tests/test_votes.py b/backend/tests/test_votes.py index d114885..b2d6e70 100644 --- a/backend/tests/test_votes.py +++ b/backend/tests/test_votes.py @@ -40,7 +40,31 @@ async def _create_report( headers=auth_headers, ) assert response.status_code == 201 - report_id = str(response.json()["id"]) + return str(response.json()["id"]) + + +async def _verify_report( + client: AsyncClient, + report_id: str, + admin_auth_headers: dict[str, str], +) -> None: + """Verify a report through the moderation API.""" + response = await client.post( + f"/api/v1/moderation/reports/{report_id}/verify", + headers=admin_auth_headers, + ) + assert response.status_code == 200 + + +async def _create_verified_report( + client: AsyncClient, + job_posting_id: str, + auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], +) -> str: + """Create and verify a report for public vote tests.""" + report_id = await _create_report(client, job_posting_id, auth_headers) + await _verify_report(client, report_id, admin_auth_headers) return report_id @@ -64,9 +88,15 @@ async def test_create_vote_success( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: - """Authenticated users should be able to vote on reports.""" - report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + """Authenticated users should be able to vote on verified reports.""" + report_id = await _create_verified_report( + client, + str(sample_job_posting.id), + auth_headers, + admin_auth_headers, + ) response = await client.post( f"/api/v1/reports/{report_id}/votes", json={"vote": VoteValue.UP.value}, @@ -83,9 +113,15 @@ async def test_create_duplicate_vote_returns_409( client: AsyncClient, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: """Duplicate votes by the same user should return 409.""" - report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + report_id = await _create_verified_report( + client, + str(sample_job_posting.id), + auth_headers, + admin_auth_headers, + ) first = await client.post( f"/api/v1/reports/{report_id}/votes", json={"vote": VoteValue.UP.value}, @@ -107,10 +143,16 @@ async def test_create_vote_handles_integrity_error_on_flush( db_session: AsyncSession, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], client: AsyncClient, ) -> None: """Duplicate vote races should return conflict instead of 500.""" - report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + report_id = await _create_verified_report( + client, + str(sample_job_posting.id), + auth_headers, + admin_auth_headers, + ) me_response = await client.get("/api/v1/auth/me", headers=auth_headers) voter = User(id=me_response.json()["id"]) @@ -136,9 +178,15 @@ async def test_create_vote_writes_audit_row( db_session: AsyncSession, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: """Vote creation should write an audit log entry.""" - report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + report_id = await _create_verified_report( + client, + str(sample_job_posting.id), + auth_headers, + admin_auth_headers, + ) response = await client.post( f"/api/v1/reports/{report_id}/votes", json={"vote": VoteValue.UP.value}, @@ -163,6 +211,7 @@ async def test_create_vote_triggers_score_snapshot( db_session: AsyncSession, sample_job_posting: JobPosting, auth_headers: dict[str, str], + admin_auth_headers: dict[str, str], ) -> None: """Vote creation should persist score snapshots for posting and company.""" before_result = await db_session.execute( @@ -172,7 +221,12 @@ async def test_create_vote_triggers_score_snapshot( ) before_count = int(before_result.scalar_one()) - report_id = await _create_report(client, str(sample_job_posting.id), auth_headers) + report_id = await _create_verified_report( + client, + str(sample_job_posting.id), + auth_headers, + admin_auth_headers, + ) vote_response = await client.post( f"/api/v1/reports/{report_id}/votes", json={"vote": VoteValue.UP.value}, From 729215d085b4b8acbcd5e56ef46acf1ba48d5e5a Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 05/12] fix: rotate refresh tokens and revoke sessions on logout Rotate refresh tokens atomically, normalize identity for case-insensitive uniqueness, and enforce bcrypt UTF-8 byte limits. --- backend/app/db_errors.py | 12 ++++ backend/app/security/password_policy.py | 21 ++++++ backend/app/services/auth.py | 92 +++++++------------------ backend/app/services/refresh_tokens.py | 87 +++++++++++------------ backend/app/services/user_identity.py | 56 +++++++++++++++ backend/tests/test_refresh_tokens.py | 8 ++- 6 files changed, 166 insertions(+), 110 deletions(-) create mode 100644 backend/app/security/password_policy.py create mode 100644 backend/app/services/user_identity.py diff --git a/backend/app/db_errors.py b/backend/app/db_errors.py index 1a60990..db847a8 100644 --- a/backend/app/db_errors.py +++ b/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 diff --git a/backend/app/security/password_policy.py b/backend/app/security/password_policy.py new file mode 100644 index 0000000..8016e92 --- /dev/null +++ b/backend/app/security/password_policy.py @@ -0,0 +1,21 @@ +"""Password validation helpers aligned with bcrypt input limits.""" + +BCRYPT_MAX_PASSWORD_BYTES = 72 + + +def password_utf8_byte_length(password: str) -> int: + """Return the UTF-8 byte length of a password string.""" + return len(password.encode("utf-8")) + + +def validate_password_byte_length(password: str) -> None: + """Reject passwords whose UTF-8 encoding exceeds bcrypt's supported boundary. + + Args: + password: Candidate password. + + Raises: + ValueError: When the password exceeds the bcrypt byte limit. + """ + if password_utf8_byte_length(password) > BCRYPT_MAX_PASSWORD_BYTES: + raise ValueError(f"Password must be {BCRYPT_MAX_PASSWORD_BYTES} UTF-8 bytes or fewer") diff --git a/backend/app/services/auth.py b/backend/app/services/auth.py index 7f9d811..1559384 100644 --- a/backend/app/services/auth.py +++ b/backend/app/services/auth.py @@ -2,7 +2,7 @@ from uuid import UUID -from sqlalchemy import or_, select +from sqlalchemy import func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -17,7 +17,12 @@ from app.services.refresh_tokens import ( issue_refresh_token, revoke_refresh_token, - validate_refresh_token, + rotate_refresh_token, +) +from app.services.user_identity import ( + normalize_email, + normalize_login_identifier, + normalize_username, ) @@ -27,31 +32,23 @@ async def register_user( settings: Settings, payload: RegisterRequest, ) -> TokenResponse: - """Register a new user and return access and refresh tokens. - - Args: - session: Active database session. - redis: Active Redis client. - settings: Application settings. - payload: Registration request payload. - - Returns: - TokenResponse: Bearer access and refresh token response. + """Register a new user and return access and refresh tokens.""" + email = normalize_email(str(payload.email)) + username = normalize_username(payload.username) - Raises: - ConflictError: When the email or username already exists. - """ - email_result = await session.execute(select(User).where(User.email == payload.email)) + email_result = await session.execute(select(User.id).where(func.lower(User.email) == email)) if email_result.scalar_one_or_none() is not None: raise ConflictError("Email already registered") - username_result = await session.execute(select(User).where(User.username == payload.username)) + username_result = await session.execute( + select(User.id).where(func.lower(User.username) == username) + ) if username_result.scalar_one_or_none() is not None: raise ConflictError("Username already taken") user = User( - email=payload.email, - username=payload.username, + email=email, + username=username, hashed_password=hash_password(payload.password), ) session.add(user) @@ -75,23 +72,11 @@ async def authenticate_user( settings: Settings, payload: LoginRequest, ) -> TokenResponse: - """Authenticate a user and return access and refresh tokens. - - Args: - session: Active database session. - redis: Active Redis client. - settings: Application settings. - payload: Login request payload. - - Returns: - TokenResponse: Bearer access and refresh token response. - - Raises: - UnauthorizedError: When credentials are invalid. - """ + """Authenticate a user and return access and refresh tokens.""" + identifier = normalize_login_identifier(payload.identifier) result = await session.execute( select(User).where( - or_(User.email == payload.identifier, User.username == payload.identifier) + or_(func.lower(User.email) == identifier, func.lower(User.username) == identifier) ) ) user = result.scalar_one_or_none() @@ -111,53 +96,28 @@ async def refresh_user_tokens( settings: Settings, refresh_token: str, ) -> TokenResponse: - """Exchange a valid refresh token for a new access token. - - Args: - session: Active database session. - redis: Active Redis client. - settings: Application settings. - refresh_token: Opaque refresh token from the client. - - Returns: - TokenResponse: New access token and the same refresh token. - - Raises: - UnauthorizedError: When the refresh token is missing, expired, or revoked. - """ - user_id = await validate_refresh_token(redis, refresh_token) - if user_id is None: + """Exchange a valid refresh token for rotated access and refresh tokens.""" + rotation = await rotate_refresh_token(redis, refresh_token, settings) + if rotation is None: raise UnauthorizedError() + user_id, new_refresh_token = rotation user = await get_user_by_id(session, user_id) if user is None: raise UnauthorizedError() return TokenResponse( access_token=create_access_token(user.id, settings), - refresh_token=refresh_token, + refresh_token=new_refresh_token, ) async def logout_user(redis: RedisClient, refresh_token: str) -> None: - """Revoke a refresh token without invalidating existing access tokens. - - Args: - redis: Active Redis client. - refresh_token: Opaque refresh token from the client. - """ + """Revoke a refresh token without invalidating existing access tokens.""" await revoke_refresh_token(redis, refresh_token) async def get_user_by_id(session: AsyncSession, user_id: UUID) -> User | None: - """Load a user by primary key. - - Args: - session: Active database session. - user_id: User UUID. - - Returns: - User | None: Matching user when found. - """ + """Load a user by primary key.""" result = await session.execute(select(User).where(User.id == user_id)) return result.scalar_one_or_none() diff --git a/backend/app/services/refresh_tokens.py b/backend/app/services/refresh_tokens.py index 944f734..3d851b1 100644 --- a/backend/app/services/refresh_tokens.py +++ b/backend/app/services/refresh_tokens.py @@ -7,42 +7,27 @@ from app.config import Settings from app.redis_client import RedisClient +_CONSUME_REFRESH_SCRIPT = """ +local value = redis.call('GET', KEYS[1]) +if value then + redis.call('DEL', KEYS[1]) +end +return value +""" -def hash_refresh_token(token: str) -> str: - """Return the SHA-256 hex digest for a refresh token. - - Args: - token: Opaque refresh token. - Returns: - str: Hex-encoded SHA-256 digest. - """ +def hash_refresh_token(token: str) -> str: + """Return the SHA-256 hex digest for a refresh token.""" return hashlib.sha256(token.encode("utf-8")).hexdigest() def refresh_token_key(token_hash: str) -> str: - """Build the Redis key for a refresh token hash. - - Args: - token_hash: SHA-256 hex digest of the refresh token. - - Returns: - str: Redis key in the form refresh:{token_hash}. - """ + """Build the Redis key for a refresh token hash.""" return f"refresh:{token_hash}" async def issue_refresh_token(redis: RedisClient, user_id: UUID, settings: Settings) -> str: - """Issue an opaque refresh token and store its hash in Redis. - - Args: - redis: Active Redis client. - user_id: User UUID associated with the token. - settings: Application settings. - - Returns: - str: Opaque refresh token for the client. - """ + """Issue an opaque refresh token and store its hash in Redis.""" token = secrets.token_urlsafe(32) token_hash = hash_refresh_token(token) ttl_seconds = settings.refresh_token_expire_days * 86400 @@ -50,32 +35,48 @@ async def issue_refresh_token(redis: RedisClient, user_id: UUID, settings: Setti return token -async def validate_refresh_token(redis: RedisClient, token: str) -> UUID | None: - """Validate a refresh token against Redis storage. +async def consume_refresh_token(redis: RedisClient, token: str) -> UUID | None: + """Atomically validate and revoke a refresh token.""" + token_hash = hash_refresh_token(token) + user_id = await redis.eval( # type: ignore[no-untyped-call] + _CONSUME_REFRESH_SCRIPT, + 1, + refresh_token_key(token_hash), + ) + if user_id is None: + return None + try: + return UUID(str(user_id)) + except ValueError: + return None - Args: - redis: Active Redis client. - token: Opaque refresh token from the client. - Returns: - UUID | None: Associated user UUID when the token is valid. - """ +async def validate_refresh_token(redis: RedisClient, token: str) -> UUID | None: + """Validate a refresh token without consuming it.""" token_hash = hash_refresh_token(token) - user_id = await redis.get(refresh_token_key(token_hash)) - if user_id is None: + stored_user_id = await redis.get(refresh_token_key(token_hash)) + if stored_user_id is None: return None try: - return UUID(user_id) + return UUID(stored_user_id) except ValueError: return None async def revoke_refresh_token(redis: RedisClient, token: str) -> None: - """Revoke a refresh token by deleting its Redis entry. - - Args: - redis: Active Redis client. - token: Opaque refresh token from the client. - """ + """Revoke a refresh token by deleting its Redis entry.""" token_hash = hash_refresh_token(token) await redis.delete(refresh_token_key(token_hash)) + + +async def rotate_refresh_token( + redis: RedisClient, + token: str, + settings: Settings, +) -> tuple[UUID, str] | None: + """Consume an existing refresh token and issue a replacement.""" + user_id = await consume_refresh_token(redis, token) + if user_id is None: + return None + new_token = await issue_refresh_token(redis, user_id, settings) + return user_id, new_token diff --git a/backend/app/services/user_identity.py b/backend/app/services/user_identity.py new file mode 100644 index 0000000..d46f2b1 --- /dev/null +++ b/backend/app/services/user_identity.py @@ -0,0 +1,56 @@ +"""Canonical normalization for user identity fields.""" + +import re + +_USERNAME_PATTERN = re.compile(r"^[a-z0-9_\-]+$") + + +def normalize_email(email: str) -> str: + """Normalize an email address for storage and lookup. + + Args: + email: Raw email input. + + Returns: + str: Trimmed, lowercased email. + """ + return email.strip().lower() + + +def normalize_username(username: str) -> str: + """Normalize a username for storage and lookup. + + Usernames are stored and compared case-insensitively by lowercasing. + + Args: + username: Raw username input. + + Returns: + str: Trimmed, lowercased username. + + Raises: + ValueError: When the normalized username is empty or invalid. + """ + normalized = username.strip().lower() + if not normalized: + raise ValueError("Username must be non-empty") + if not _USERNAME_PATTERN.fullmatch(normalized): + raise ValueError("Username must contain only letters, numbers, underscores, and hyphens") + return normalized + + +def normalize_login_identifier(identifier: str) -> str: + """Normalize a login identifier for lookup. + + Email-like identifiers are lowercased. Usernames follow username rules. + + Args: + identifier: Email or username supplied at login. + + Returns: + str: Normalized identifier. + """ + trimmed = identifier.strip() + if "@" in trimmed: + return normalize_email(trimmed) + return normalize_username(trimmed) diff --git a/backend/tests/test_refresh_tokens.py b/backend/tests/test_refresh_tokens.py index 1070aca..1e29261 100644 --- a/backend/tests/test_refresh_tokens.py +++ b/backend/tests/test_refresh_tokens.py @@ -105,10 +105,16 @@ async def test_refresh_returns_new_access_token(client: AsyncClient) -> None: ) assert refresh_response.status_code == 200 refreshed = refresh_response.json() - assert refreshed["refresh_token"] == refresh_token + assert refreshed["refresh_token"] != refresh_token assert refreshed["access_token"] assert refreshed["access_token"] != original_access_token + reuse_response = await client.post( + "/api/v1/auth/refresh", + json={"refresh_token": refresh_token}, + ) + assert reuse_response.status_code == 401 + @pytest.mark.asyncio async def test_refresh_rejects_missing_token(client: AsyncClient) -> None: From 5730085aaa35b868e70d02cde2b418c29a4e3ce3 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 06/12] fix: await asynchronous Next.js route parameters Store refresh tokens in session state, call backend logout on sign out, and await App Router searchParams in page components. --- frontend/app/companies/page.test.tsx | 2 +- frontend/app/companies/page.tsx | 5 +-- frontend/app/page.test.tsx | 29 +++++++++++------ frontend/app/page.tsx | 10 +++--- frontend/components/AppShell.tsx | 4 ++- frontend/components/auth/LoginForm.tsx | 4 +-- frontend/components/auth/RegisterForm.tsx | 4 +-- frontend/lib/api/endpoints.test.ts | 20 ++++++++++++ frontend/lib/api/endpoints.ts | 14 +++++++++ frontend/lib/session/session-context.test.tsx | 26 +++++++++++----- frontend/lib/session/session-context.tsx | 31 ++++++++++++++----- frontend/test-utils/WithAccessToken.tsx | 6 ++-- 12 files changed, 114 insertions(+), 41 deletions(-) diff --git a/frontend/app/companies/page.test.tsx b/frontend/app/companies/page.test.tsx index 4c5c924..25fae5d 100644 --- a/frontend/app/companies/page.test.tsx +++ b/frontend/app/companies/page.test.tsx @@ -22,7 +22,7 @@ describe("CompaniesPage", () => { page_size: 20, }); - const page = await CompaniesPage({ searchParams: {} }); + const page = await CompaniesPage({ searchParams: Promise.resolve({}) }); render(page); expect(screen.getByText("No companies are indexed yet.")).toBeInTheDocument(); diff --git a/frontend/app/companies/page.tsx b/frontend/app/companies/page.tsx index 948235f..51c9d88 100644 --- a/frontend/app/companies/page.tsx +++ b/frontend/app/companies/page.tsx @@ -4,11 +4,12 @@ import { ApiError } from "@/lib/api/client"; import { DeferredNotice } from "@/components/DeferredNotice"; interface CompaniesPageProps { - searchParams?: { page?: string }; + searchParams?: Promise<{ page?: string }>; } export default async function CompaniesPage({ searchParams }: CompaniesPageProps) { - const page = Number(searchParams?.page ?? "1") || 1; + const resolvedSearchParams = searchParams ? await searchParams : undefined; + const page = Number(resolvedSearchParams?.page ?? "1") || 1; let error: string | null = null; let data = null; diff --git a/frontend/app/page.test.tsx b/frontend/app/page.test.tsx index 1309ab9..b3f8224 100644 --- a/frontend/app/page.test.tsx +++ b/frontend/app/page.test.tsx @@ -1,30 +1,39 @@ import { render, screen } from "@testing-library/react"; -import HomePage from "./page"; - jest.mock("@/components/HomeHero", () => ({ HomeHero: ({ postingUrl }: { postingUrl?: string }) => (
{postingUrl ?? ""}
), })); +import HomePage from "./page"; + describe("HomePage", () => { - it("passes posting_url query param to HomeHero", () => { - render(); + it("passes posting_url query param to HomeHero", async () => { + const page = await HomePage({ + searchParams: Promise.resolve({ posting_url: "https://example.com/jobs/123" }), + }); + render(page); expect(screen.getByTestId("home-hero")).toHaveTextContent("https://example.com/jobs/123"); }); - it("ignores empty posting_url values", () => { - render(); + it("ignores empty posting_url values", async () => { + const page = await HomePage({ + searchParams: Promise.resolve({ posting_url: " " }), + }); + render(page); expect(screen.getByTestId("home-hero")).toHaveTextContent(""); }); - it("uses the first posting_url when an array is provided", () => { - render( - , - ); + it("uses the first posting_url when an array is provided", async () => { + const page = await HomePage({ + searchParams: Promise.resolve({ + posting_url: ["https://example.com/a", "https://example.com/b"], + }), + }); + render(page); expect(screen.getByTestId("home-hero")).toHaveTextContent("https://example.com/a"); }); diff --git a/frontend/app/page.tsx b/frontend/app/page.tsx index 89b72ec..419637c 100644 --- a/frontend/app/page.tsx +++ b/frontend/app/page.tsx @@ -1,9 +1,9 @@ import { HomeHero } from "@/components/HomeHero"; interface HomePageProps { - searchParams?: { + searchParams?: Promise<{ posting_url?: string | string[]; - }; + }>; } function resolvePostingUrl(value: string | string[] | undefined): string | undefined { @@ -16,12 +16,14 @@ function resolvePostingUrl(value: string | string[] | undefined): string | undef return undefined; } -export default function HomePage({ searchParams }: HomePageProps) { +export default async function HomePage({ searchParams }: HomePageProps) { + const resolvedSearchParams = searchParams ? await searchParams : undefined; + return ( ); } diff --git a/frontend/components/AppShell.tsx b/frontend/components/AppShell.tsx index a3c3369..72e2539 100644 --- a/frontend/components/AppShell.tsx +++ b/frontend/components/AppShell.tsx @@ -24,7 +24,9 @@ export function AppShell({ children }: { children: React.ReactNode }) { ); } describe("SessionProvider", () => { - it("stores access tokens in React state only", () => { + it("stores access and refresh tokens in React state only", () => { render( , ); - expect(screen.getByTestId("token")).toHaveTextContent("none"); + expect(screen.getByTestId("access-token")).toHaveTextContent("none"); + expect(screen.getByTestId("refresh-token")).toHaveTextContent("none"); expect(window.localStorage.getItem("accessToken")).toBeNull(); expect(window.sessionStorage.getItem("accessToken")).toBeNull(); }); @@ -34,9 +43,10 @@ describe("SessionProvider", () => { ); act(() => { - screen.getByRole("button", { name: "Set token" }).click(); + screen.getByRole("button", { name: "Set tokens" }).click(); }); - expect(screen.getByTestId("token")).toHaveTextContent("in-memory-token"); + expect(screen.getByTestId("access-token")).toHaveTextContent("in-memory-access"); + expect(screen.getByTestId("refresh-token")).toHaveTextContent("in-memory-refresh"); expect(window.localStorage.getItem("accessToken")).toBeNull(); expect(window.sessionStorage.getItem("accessToken")).toBeNull(); }); diff --git a/frontend/lib/session/session-context.tsx b/frontend/lib/session/session-context.tsx index 824dc58..d2991c6 100644 --- a/frontend/lib/session/session-context.tsx +++ b/frontend/lib/session/session-context.tsx @@ -4,8 +4,9 @@ import { createContext, useCallback, useContext, useMemo, useState } from "react interface SessionContextValue { accessToken: string | null; - setAccessToken: (token: string | null) => void; - clearSession: () => void; + refreshToken: string | null; + setSessionTokens: (accessToken: string, refreshToken: string) => void; + clearSession: () => Promise; isAuthenticated: boolean; } @@ -13,23 +14,37 @@ const SessionContext = createContext(undefined) export function SessionProvider({ children }: { children: React.ReactNode }) { const [accessToken, setAccessTokenState] = useState(null); + const [refreshToken, setRefreshTokenState] = useState(null); - const setAccessToken = useCallback((token: string | null) => { - setAccessTokenState(token); + const setSessionTokens = useCallback((nextAccessToken: string, nextRefreshToken: string) => { + setAccessTokenState(nextAccessToken); + setRefreshTokenState(nextRefreshToken); }, []); - const clearSession = useCallback(() => { + const clearSession = useCallback(async () => { + const tokenToRevoke = refreshToken; setAccessTokenState(null); - }, []); + setRefreshTokenState(null); + if (!tokenToRevoke) { + return; + } + try { + const { logoutUser } = await import("@/lib/api/endpoints"); + await logoutUser(tokenToRevoke); + } catch { + // Local session state is already cleared even when revocation fails. + } + }, [refreshToken]); const value = useMemo( () => ({ accessToken, - setAccessToken, + refreshToken, + setSessionTokens, clearSession, isAuthenticated: accessToken !== null, }), - [accessToken, setAccessToken, clearSession], + [accessToken, refreshToken, setSessionTokens, clearSession], ); return {children}; diff --git a/frontend/test-utils/WithAccessToken.tsx b/frontend/test-utils/WithAccessToken.tsx index f4125fb..84c9552 100644 --- a/frontend/test-utils/WithAccessToken.tsx +++ b/frontend/test-utils/WithAccessToken.tsx @@ -8,11 +8,11 @@ export function WithAccessToken({ token: string; children: React.ReactNode; }) { - const { setAccessToken } = useSession(); + const { setSessionTokens } = useSession(); useEffect(() => { - setAccessToken(token); - }, [setAccessToken, token]); + setSessionTokens(token, "test-refresh-token"); + }, [setSessionTokens, token]); return <>{children}; } From 7cee1486c802abc971fde129776bf663e0f5c761 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 07/12] fix: enforce moderation integrity in sheet import dry runs Require reviewer and reviewed_at for import-ready rows, expand PII scanning, and avoid shared ATS hosts as company domains. --- backend/app/services/sheet_import.py | 108 ++++++++++++++++++++++++--- backend/tests/test_sheet_import.py | 3 +- 2 files changed, 101 insertions(+), 10 deletions(-) diff --git a/backend/app/services/sheet_import.py b/backend/app/services/sheet_import.py index 551976c..8fb4c82 100644 --- a/backend/app/services/sheet_import.py +++ b/backend/app/services/sheet_import.py @@ -6,18 +6,22 @@ import hashlib import re from dataclasses import dataclass, field +from datetime import UTC, datetime from pathlib import Path -from urllib.parse import urlsplit from app.models.enums import PostingSource, ReportType from app.services.job_url_validation import ( JobSourceProvider, detect_job_source_provider, + extract_employer_identity, normalize_job_url, validate_http_https_url, ) _EMAIL_IN_TEXT = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}") +_PHONE_IN_TEXT = re.compile(r"(? bool: return bool(_EMAIL_IN_TEXT.search(text)) +def _scan_text_for_pii(text: str) -> list[str]: + """Return conservative PII category labels detected in free text.""" + findings: list[str] = [] + if _contains_email(text): + findings.append("email") + if _PHONE_IN_TEXT.search(text): + findings.append("phone") + if _PRIVATE_KEY_HEADER.search(text): + findings.append("private_key_header") + if _TOKEN_LIKE.search(text): + findings.append("credential_like_token") + return findings + + +def _parse_reviewed_at(value: str) -> datetime | None: + """Parse a reviewed_at timestamp when present.""" + trimmed = value.strip() + if not trimmed: + return None + normalized = trimmed.replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(normalized) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed + + +def _company_domain_hint(normalized_url: str) -> str | None: + """Return a company domain hint without using shared ATS hosts.""" + identity = extract_employer_identity(normalized_url) + if identity.company_domain_hint: + return identity.company_domain_hint + if identity.tenant_identifier: + return identity.tenant_identifier + return None + + def _split_evidence_links(raw: str) -> tuple[list[str], list[str]]: if not raw.strip(): return [], [] @@ -258,6 +306,36 @@ def check_eligibility( message="import_ready must be yes", ) + reviewer = _cell(row, column_map, "reviewer") + if not reviewer: + return EligibilityResult( + eligible=False, + reason_code="missing_reviewer", + message="reviewer is required for import-ready rows", + ) + + reviewed_at_raw = _cell(row, column_map, "reviewed_at") + reviewed_at = _parse_reviewed_at(reviewed_at_raw) + if reviewed_at is None: + return EligibilityResult( + eligible=False, + reason_code="invalid_reviewed_at", + message="reviewed_at must be a valid ISO date or datetime", + ) + if reviewed_at > datetime.now(tz=UTC): + return EligibilityResult( + eligible=False, + reason_code="future_reviewed_at", + message="reviewed_at must not be in the future", + ) + + if review_status in {"pending", "reviewing"}: + return EligibilityResult( + eligible=False, + reason_code="review_status_pending", + message="import-ready rows cannot remain pending or reviewing", + ) + pii_redacted = _cell(row, column_map, "pii_redacted").lower() if pii_redacted != "yes": return EligibilityResult( @@ -332,12 +410,25 @@ def check_eligibility( message="Narrative must be at least 20 characters", ) - if _contains_email(narrative): - return EligibilityResult( - eligible=False, - reason_code="pii_in_narrative", - message="Narrative must not contain email addresses", - ) + pii_fields = { + "narrative": narrative, + "notes": _cell(row, column_map, "notes"), + "company_name": company_name, + "job_title": job_title, + "location": _cell(row, column_map, "location"), + "evidence_links": _cell(row, column_map, "evidence_links"), + "optional_contact_email": _cell(row, column_map, "optional_contact_email"), + } + for field_name, field_value in pii_fields.items(): + if field_name == "optional_contact_email" and not field_value.strip(): + continue + findings = _scan_text_for_pii(field_value) + if findings: + return EligibilityResult( + eligible=False, + reason_code=f"pii_in_{field_name}", + message=f"{field_name} contains potential PII ({', '.join(findings)})", + ) normalized = normalize_job_url(job_url) if normalized is None: @@ -413,7 +504,6 @@ def plan_row_import( evidence_urls=evidence_urls, ) - host = urlsplit(normalized).hostname or "" fingerprint = _row_fingerprint( timestamp=_cell(row, column_map, "timestamp"), normalized_url=normalized, @@ -428,7 +518,7 @@ def plan_row_import( message=None, company_action="match_or_create", company_name=company_name, - company_domain=host.lower() if host else None, + company_domain=_company_domain_hint(normalized), posting_action="match_or_create", original_job_url=job_url, normalized_job_url=normalized, diff --git a/backend/tests/test_sheet_import.py b/backend/tests/test_sheet_import.py index cdee536..3c502e0 100644 --- a/backend/tests/test_sheet_import.py +++ b/backend/tests/test_sheet_import.py @@ -48,7 +48,7 @@ def _eligible_row(**overrides: str) -> dict[str, str]: "Has the company responded?": "No", "Consent": "I confirm this submission is made in good faith.", "Evidence links": "https://example.com/archive", - "Optional contact email": "reporter@example.com", + "Optional contact email": "", "review_status": "approved_for_import", "reviewer": "maintainer", "reviewed_at": "2026-07-01", @@ -127,6 +127,7 @@ def test_plan_row_import_builds_description_preview() -> None: plan = plan_row_import(_eligible_row(), column_map=column_map, row_number=2) assert plan.action == "would_import" assert plan.normalized_job_url == "https://boards.greenhouse.io/acme/jobs/1234567" + assert plan.company_domain == "acme" assert plan.report_type == "ghost_job" assert plan.description_preview is not None assert "[ghost-sweep Sheet import]" in plan.description_preview From 9cab57c35d3451aa2d8249b3671c6814b47ed0cb Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 08/12] fix: normalize ATS classification and employer identity Map supported ATS providers to company_site and extract tenant identity without using shared platform hosts as company domains. --- backend/app/services/job_url_validation.py | 70 ++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/backend/app/services/job_url_validation.py b/backend/app/services/job_url_validation.py index 34e895f..7a828c6 100644 --- a/backend/app/services/job_url_validation.py +++ b/backend/app/services/job_url_validation.py @@ -250,6 +250,76 @@ def validate_http_https_url(raw_url: str) -> str: return trimmed +@dataclass(frozen=True) +class EmployerIdentity: + """Employer identity hints extracted from a hosted ATS or careers URL.""" + + provider: JobSourceProvider + hosted_platform_domain: str | None + tenant_identifier: str | None + company_domain_hint: str | None + + +def _first_path_segment(path: str) -> str | None: + segments = [segment for segment in path.split("/") if segment] + if not segments: + return None + return segments[0] + + +def extract_employer_identity(normalized_url: str) -> EmployerIdentity: + """Extract employer identity hints without inventing a shared ATS domain.""" + parts = urlsplit(normalized_url) + host = (parts.hostname or "").lower() + path = parts.path or "" + provider = detect_job_source_provider(normalized_url) + tenant: str | None = None + company_domain_hint: str | None = None + + if provider == JobSourceProvider.GREENHOUSE and host.endswith("greenhouse.io"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.LEVER and host.endswith("lever.co"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.ASHBY and host.endswith("ashbyhq.com"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.SMARTRECRUITERS and host.endswith("smartrecruiters.com"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.WORKABLE and host.endswith("workable.com"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.JOBVITE and host.endswith("jobvite.com"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.BAMBOOHR and host.endswith("bamboohr.com"): + tenant = host.split(".", maxsplit=1)[0] if host.count(".") >= 2 else None + elif provider == JobSourceProvider.WORKDAY: + if host.endswith("myworkdayjobs.com"): + tenant = host.split(".", maxsplit=1)[0] + elif host.endswith("myworkdaysite.com"): + tenant = _first_path_segment(path) + elif provider == JobSourceProvider.COMPANY_CAREERS: + company_domain_hint = host + + hosted_platform_domains = { + JobSourceProvider.GREENHOUSE, + JobSourceProvider.LEVER, + JobSourceProvider.ASHBY, + JobSourceProvider.SMARTRECRUITERS, + JobSourceProvider.WORKABLE, + JobSourceProvider.JOBVITE, + JobSourceProvider.ICIMS, + JobSourceProvider.TALEO, + JobSourceProvider.WORKDAY, + JobSourceProvider.BAMBOOHR, + } + hosted_domain = host if provider in hosted_platform_domains else None + + return EmployerIdentity( + provider=provider, + hosted_platform_domain=hosted_domain, + tenant_identifier=tenant, + company_domain_hint=company_domain_hint, + ) + + def validate_job_url(raw_url: str) -> JobUrlValidationResult: """Validate and classify a raw job URL offline. From 572705615c3ddb4ab5d94f7f54542882e776a084 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:26 -0500 Subject: [PATCH 09/12] fix: separate application liveness from dependency readiness Keep /health lightweight and add /health/ready with PostgreSQL and Redis checks for Docker health probes. --- backend/app/main.py | 49 +++++++++++++++++++++++++++++++++++++++++---- docker-compose.yml | 6 ++++++ 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/backend/app/main.py b/backend/app/main.py index 36bd996..9bc0ae4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,8 +4,10 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, status from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from sqlalchemy import text from app.api.v1.auth import router as auth_router from app.api.v1.companies import router as companies_router @@ -14,8 +16,14 @@ 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.database import SessionLocal +from app.redis_client import ( + check_redis_connection, + get_shared_redis_client, + init_redis_client, + shutdown_redis_client, +) +from app.schemas import HealthResponse, ReadinessCheckStatus, ReadinessResponse logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -52,10 +60,43 @@ 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() -> JSONResponse: + """Return dependency readiness for PostgreSQL and Redis.""" + postgres_status = "unavailable" + redis_status = "unavailable" + + try: + async with SessionLocal() as session: + await session.execute(text("SELECT 1")) + postgres_status = "ok" + except Exception: + logger.exception("PostgreSQL readiness check failed") + + try: + redis_client = get_shared_redis_client() + if await check_redis_connection(redis_client): + 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.""" diff --git a/docker-compose.yml b/docker-compose.yml index e7b4af6..211f730 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -50,6 +50,12 @@ services: CORS_ORIGINS: '["http://localhost:3000"]' ports: - "8000:8000" + healthcheck: + test: ["CMD-SHELL", "python -c \"import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health/ready')\""] + interval: 10s + timeout: 5s + retries: 10 + start_period: 20s depends_on: postgres: condition: service_healthy From 4d4456039ae14c90ba23cb1f7fe0452d43cabf4c Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:27 -0500 Subject: [PATCH 10/12] fix: exclude macOS metadata from audit bundles Set COPYFILE_DISABLE during archive creation and add packaging regression coverage for AppleDouble filenames. --- scripts/create_audit_bundle.py | 22 +++++++---- scripts/test_audit_bundle_packaging.py | 55 ++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 7 deletions(-) create mode 100644 scripts/test_audit_bundle_packaging.py diff --git a/scripts/create_audit_bundle.py b/scripts/create_audit_bundle.py index 9825d48..0b79535 100644 --- a/scripts/create_audit_bundle.py +++ b/scripts/create_audit_bundle.py @@ -450,13 +450,21 @@ def create_tarball(bundle_root: Path, tarball: Path) -> None: """Create gzip tarball from bundle root.""" if tarball.exists(): tarball.unlink() - with tarfile.open(tarball, "w:gz") as archive: - for path in sorted(bundle_root.rglob("*")): - if path.is_file(): - rel = path.relative_to(bundle_root.parent) - if path.name.startswith("._") or path.name == ".DS_Store": - continue - archive.add(path, arcname=str(rel)) + previous_copyfile_disable = os.environ.get("COPYFILE_DISABLE") + os.environ["COPYFILE_DISABLE"] = "1" + try: + with tarfile.open(tarball, "w:gz") as archive: + for path in sorted(bundle_root.rglob("*")): + if path.is_file(): + rel = path.relative_to(bundle_root.parent) + if path.name.startswith("._") or path.name == ".DS_Store": + continue + archive.add(path, arcname=str(rel)) + finally: + if previous_copyfile_disable is None: + os.environ.pop("COPYFILE_DISABLE", None) + else: + os.environ["COPYFILE_DISABLE"] = previous_copyfile_disable def verify_tarball(tarball: Path) -> list[str]: diff --git a/scripts/test_audit_bundle_packaging.py b/scripts/test_audit_bundle_packaging.py new file mode 100644 index 0000000..5586aa8 --- /dev/null +++ b/scripts/test_audit_bundle_packaging.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +"""Regression tests for audit bundle macOS metadata exclusion.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import tarfile +import tempfile +from pathlib import Path + + +def test_tarball_excludes_appledouble_files() -> None: + """Tarball creation must exclude AppleDouble metadata files.""" + repo_root = Path(__file__).resolve().parents[1] + script = repo_root / "scripts" / "create_audit_bundle.py" + with tempfile.TemporaryDirectory() as temp_dir: + bundle_root = Path(temp_dir) / "sample-bundle" + bundle_root.mkdir() + (bundle_root / "README.txt").write_text("sample", encoding="utf-8") + (bundle_root / "._README.txt").write_bytes(b"appledouble") + tarball = Path(temp_dir) / "sample-bundle.tar.gz" + + env = os.environ.copy() + env["COPYFILE_DISABLE"] = "1" + result = subprocess.run( + [ + sys.executable, + "-c", + ( + "from pathlib import Path; " + "import importlib.util; " + f"spec=importlib.util.spec_from_file_location('cab', '{script}'); " + "module=importlib.util.module_from_spec(spec); " + "spec.loader.exec_module(module); " + f"module.create_tarball(Path('{bundle_root}'), Path('{tarball}'))" + ), + ], + capture_output=True, + text=True, + check=False, + env=env, + ) + assert result.returncode == 0, result.stderr + + with tarfile.open(tarball, "r:gz") as archive: + names = archive.getnames() + assert any(name.endswith("README.txt") for name in names) + assert not any("/._" in name or name.startswith("._") for name in names) + + +if __name__ == "__main__": + test_tarball_excludes_appledouble_files() + print("PASS") From a687e9c9e51d5f9e92f6a8c98510b112daa957b9 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:19:27 -0500 Subject: [PATCH 11/12] docs: document deep-audit remediation behavior for v0.1.2 Record moderation, scoring, session, health, and sheet-import behavior aligned with the planned remediation release. --- docs/deep-audit-remediation-v0.1.2.md | 47 +++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 docs/deep-audit-remediation-v0.1.2.md diff --git a/docs/deep-audit-remediation-v0.1.2.md b/docs/deep-audit-remediation-v0.1.2.md new file mode 100644 index 0000000..3d1e19d --- /dev/null +++ b/docs/deep-audit-remediation-v0.1.2.md @@ -0,0 +1,47 @@ +# Deep Audit Remediation (planned v0.1.2) + +This document summarizes remediation work prepared after the independent deep audit of 2026-07-10. The changes are implemented on branch `fix/deep-audit-remediation-v0.1.2` and are not released until review, merge, and a separate release batch publish `v0.1.2`. + +## High-severity fixes + +- Public report list, detail, and employer-response reads return only `verified` reports. +- Public schemas omit reporter and employer-user UUIDs. +- Score calculation uses only verified reports. +- Pending report creation no longer changes public scores. +- Duplicate active report submissions are rejected with `409 Conflict`. +- Report submission rate limiting is enforced separately from auth rate limits. +- Employer response scoring counts distinct eligible reports with responses. +- Duplicate employer responses from the same account are rejected. + +## Secondary fixes + +- Community votes update `verification_votes` from persisted vote records. +- Refresh tokens rotate on use; reused tokens are rejected. +- Frontend logout calls backend revocation and clears in-memory session tokens. +- Next.js App Router pages await asynchronous `searchParams`. +- Sheet import dry runs require `reviewer` and `reviewed_at`, expand PII scanning, and avoid shared ATS hosts as company domains. +- Supported ATS providers map to `company_site` instead of `other`. +- Email and username uniqueness is case-insensitive. +- Passwords above bcrypt's UTF-8 byte boundary are rejected. +- `/health` remains liveness-only; `/health/ready` checks PostgreSQL and Redis. +- Audit bundle packaging excludes macOS `._*` metadata and sets `COPYFILE_DISABLE=1`. + +## Migration + +`003_audit_remediation` adds: + +- case-insensitive unique indexes on `users.email` and `users.username` +- unique index on `employer_responses (report_id, user_id)` +- partial unique index blocking duplicate active reports per reporter/posting/type + +The migration fails if unresolved case-equivalent user collisions already exist. + +## Dependency PR interaction + +Dependency file changes remain in open PRs #21, #22, #32, and #35. This remediation branch does not duplicate those updates. + +## Release status + +- `v0.1.0` unchanged +- `v0.1.1` unchanged +- `v0.1.2` not created in this batch From a6ad0b2792fcf0b233d0c12ebf9f81c3a4469784 Mon Sep 17 00:00:00 2001 From: Isaac Thor Date: Fri, 10 Jul 2026 19:22:31 -0500 Subject: [PATCH 12/12] fix: align readiness checks with configured test dependencies Use injected settings and Redis client for /health/ready so CI and local integration tests probe the same database and Redis URLs as the test suite. --- backend/app/main.py | 24 +++++++++++++++--------- backend/tests/conftest.py | 13 +++++++++++-- 2 files changed, 26 insertions(+), 11 deletions(-) diff --git a/backend/app/main.py b/backend/app/main.py index 9bc0ae4..6a092bd 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,10 +4,11 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from fastapi import FastAPI, status +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 @@ -15,11 +16,11 @@ 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.database import SessionLocal +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, - get_shared_redis_client, init_redis_client, shutdown_redis_client, ) @@ -65,21 +66,26 @@ async def health_check() -> HealthResponse: @app.get("/health/ready", response_model=ReadinessResponse) -async def readiness_check() -> JSONResponse: +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 SessionLocal() as session: - await session.execute(text("SELECT 1")) + 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: - redis_client = get_shared_redis_client() - if await check_redis_connection(redis_client): + if await check_redis_connection(redis): redis_status = "ok" except Exception: logger.exception("Redis readiness check failed") diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 290d7bb..ebb1b4f 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -175,6 +175,8 @@ async def redis_client(test_redis_url: str) -> AsyncGenerator[RedisClient, None] async def client( db_session: AsyncSession, redis_client: RedisClient, + test_database_url: str, + test_redis_url: str, ) -> AsyncGenerator[AsyncClient, None]: """Provide an HTTP client with database and Redis dependencies overridden. @@ -186,6 +188,13 @@ async def client( AsyncClient: Async HTTP client bound to the FastAPI application. """ + test_settings = get_settings().model_copy( + update={ + "database_url": test_database_url, + "redis_url": test_redis_url, + } + ) + async def override_get_db() -> AsyncGenerator[AsyncSession, None]: yield db_session @@ -193,13 +202,13 @@ async def override_get_redis() -> AsyncIterator[RedisClient]: yield redis_client async def override_get_settings() -> Settings: - return get_settings() + return test_settings app.dependency_overrides[get_db] = override_get_db app.dependency_overrides[get_redis] = override_get_redis app.dependency_overrides[get_settings_dependency] = override_get_settings - await init_redis_client(get_settings()) + await init_redis_client(test_settings) transport = ASGITransport(app=app) try: async with AsyncClient(transport=transport, base_url="http://testserver") as async_client: