"""Pydantic v2 request/response models. extra='forbid' on every request body. Includes PII rejection (SSN/PAN regex) on all free-text fields. """ from __future__ import annotations import re from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator # --------------------------------------------------------------- # PII rejection — compliance-critical # --------------------------------------------------------------- _SSN_RE = re.compile(r"\b\d{3}-?\d{2}-?\d{4}\b") _PAN_RE = re.compile(r"\b(?:\d[ -]?){13,19}\b") def _scan_pii(value: str) -> str: """Raise ValueError if value matches SSN or PAN regex.""" if value is None: return value if _SSN_RE.search(value): raise ValueError("Do not include Social Security or bank/card numbers.") if _PAN_RE.search(value): raise ValueError("Do not include Social Security or bank/card numbers.") return value def _pii_validator(field_name: str): return field_validator(field_name)(lambda v: _scan_pii(v)) # --------------------------------------------------------------- # Intake request # --------------------------------------------------------------- BUSINESS_TYPES = ( "INDIVIDUAL", "SOLE_PROPRIETORSHIP", "LLC", "CORPORATION", "PARTNERSHIP", "OTHER" ) MESSAGE_SUBJECTS = ( "Question about my claim", "New information about the debtor", "Payment received / want to stop recovery", "Update my contact info", "Complaint or concern", "Other", ) CLAIM_STATUSES = ( "NEW", "UNDER_REVIEW", "ACTIVE", "NEGOTIATION", "LEGAL", "SETTLED", "CLOSED", "WRITE_OFF", "REJECTED", ) TIERS = ("TIER_1", "TIER_2", "TIER_2_5", "TIER_3", "TIER_4") class IntakeClient(BaseModel): model_config = ConfigDict(extra="forbid") company_name: str = Field(..., min_length=1, max_length=200) contact_name: str = Field(..., min_length=1, max_length=200) email: EmailStr phone: str | None = Field(None, max_length=50) @field_validator("company_name", "contact_name", "phone") @classmethod def _v(cls, v): return _scan_pii(v) class IntakeDebtor(BaseModel): model_config = ConfigDict(extra="forbid") name: str = Field(..., min_length=1, max_length=200) business_type: str = Field("OTHER") contact_email: str | None = Field(None, max_length=254) contact_phone: str | None = Field(None, max_length=50) physical_address: str | None = Field(None, max_length=500) @field_validator("business_type") @classmethod def _bt(cls, v): v = v.upper() if v not in BUSINESS_TYPES: raise ValueError(f"business_type must be one of {BUSINESS_TYPES}") return v @field_validator("name", "contact_email", "contact_phone", "physical_address") @classmethod def _v(cls, v): return _scan_pii(v) class IntakeClaim(BaseModel): model_config = ConfigDict(extra="forbid") amount_cents: int = Field(..., gt=0, le=100_000_000) description: str | None = Field(None, max_length=5000) client_reference: str | None = Field(None, max_length=200) invoice_date: str | None = Field(None, max_length=20) @field_validator("description", "client_reference", "invoice_date") @classmethod def _v(cls, v): return _scan_pii(v) class IntakeRequest(BaseModel): model_config = ConfigDict(extra="forbid") client: IntakeClient debtor: IntakeDebtor claim: IntakeClaim tos_accepted: bool = True no_other_agency_accepted: bool no_prior_action_accepted: bool turnstile_token: str | None = None @field_validator("tos_accepted", "no_other_agency_accepted", "no_prior_action_accepted") @classmethod def _bool_req(cls, v): if v is not True: raise ValueError("must be true") return v # --------------------------------------------------------------- # Auth # --------------------------------------------------------------- class AuthRequest(BaseModel): model_config = ConfigDict(extra="forbid") email: EmailStr class AuthVerify(BaseModel): model_config = ConfigDict(extra="forbid") token: str = Field(..., min_length=10, max_length=200) # --------------------------------------------------------------- # Messages # --------------------------------------------------------------- class MessageCreate(BaseModel): model_config = ConfigDict(extra="forbid") subject: str content: str = Field(..., min_length=1, max_length=10000) @field_validator("subject") @classmethod def _subj(cls, v): if v not in MESSAGE_SUBJECTS: raise ValueError(f"subject must be one of {MESSAGE_SUBJECTS}") return v @field_validator("content") @classmethod def _cont(cls, v): return _scan_pii(v) # --------------------------------------------------------------- # Staff # --------------------------------------------------------------- class StaffClaimPatch(BaseModel): model_config = ConfigDict(extra="forbid") status: str | None = None tier: str | None = None reason: str | None = Field(None, max_length=500) @field_validator("status") @classmethod def _st(cls, v): if v is not None and v not in CLAIM_STATUSES: raise ValueError(f"status must be one of {CLAIM_STATUSES}") return v @field_validator("tier") @classmethod def _tr(cls, v): if v is not None and v not in TIERS: raise ValueError(f"tier must be one of {TIERS}") return v class StaffNoteCreate(BaseModel): model_config = ConfigDict(extra="forbid") content: str = Field(..., min_length=1, max_length=10000) visibility: str = "INTERNAL" author_name: str | None = Field(None, max_length=200) @field_validator("visibility") @classmethod def _vis(cls, v): if v not in ("SHARED", "INTERNAL"): raise ValueError("visibility must be SHARED or INTERNAL") return v @field_validator("content", "author_name") @classmethod def _v(cls, v): return _scan_pii(v) class StaffApproval(BaseModel): model_config = ConfigDict(extra="forbid") decision: str staff_name: str | None = Field(None, max_length=200) reason: str | None = Field(None, max_length=500) @field_validator("decision") @classmethod def _dec(cls, v): v = v.upper() if v not in ("APPROVE", "REJECT"): raise ValueError("decision must be APPROVE or REJECT") return v @field_validator("staff_name", "reason") @classmethod def _v(cls, v): return _scan_pii(v) class StaffLetterUpdate(BaseModel): model_config = ConfigDict(extra="forbid") subject: str = Field(..., min_length=1, max_length=200) body: str = Field(..., min_length=1, max_length=20000) staff_name: str | None = Field(None, max_length=200) @field_validator("subject", "body", "staff_name") @classmethod def _v(cls, v): return _scan_pii(v) class StaffOnboardingUpdate(BaseModel): model_config = ConfigDict(extra="forbid") received: bool received_by: str | None = Field(None, max_length=200) @field_validator("received_by") @classmethod def _v(cls, v): return _scan_pii(v)