DRE customer portal: FastAPI + SQLite backend (18 endpoints)
- Magic-link auth (sha256-only token storage, 15-min single-use, 7-day sessions) - Staff-key auth via X-DRE-Staff-Key (constant-time compare) - SQLite WAL, foreign_keys, parameterized queries, atomic DRE/CLT sequence allocation - Intake validator rejects SSN/PAN patterns (FDCPA/TDCPA compliance) - Document upload allowlist + magic-byte check, 20MB cap - Unified error envelope, money as integer cents - systemd unit (port 8093, User=root, hardening directives) - Fixes import bug (auth.py relative imports) and audit_log placeholder mismatch
This commit is contained in:
root
@@ -0,0 +1,190 @@
|
||||
"""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
|
||||
turnstile_token: str | None = None
|
||||
|
||||
@field_validator("tos_accepted")
|
||||
@classmethod
|
||||
def _tos(cls, v):
|
||||
if v is not True:
|
||||
raise ValueError("tos_accepted 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 = Field(..., min_length=1, 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)
|
||||
Reference in New Issue
Block a user