"""Public forms — fill in and submit a form without signing in, when the form allows it (FormSettings.public_access; what qualifies: app/crud/ public_forms.py). Nothing here needs a session; each request sets the form's workspace for row-level security itself. GET /public/forms//
the live form GET /public/forms///options/ a datasource-linked field's options POST /public/forms///verify/send a code for a verified email / phone POST /public/forms///verify/confirm that code -> a token POST /public/forms///drafts save to continue later (a private link) GET /public/forms///drafts/ that draft's answers POST /public/forms///submissions submit (a saved draft too, by its token) GET /public/forms///document.pdf the submission's PDF, by the token submitting gave POST /public/forms///files upload a file for one of its file fields GET /public/forms///files/ that file's name and size (not its content) POST /public/forms///status/send an applicant's code, by their reference number POST /public/forms///status/confirm that code -> a status token POST /public/forms///status/view their application's status and history POST /public/forms///status/resubmit their corrections to a returned application The form must be open (its schedule — app/crud/form_schedule.py) to save or submit; GET says whether it is. The status pages work whenever the form has FormReview.status_lookup on (app/crud/application_review.py). Submissions are checked exactly like a signed-in one (app/routers/ form_submissions.py::_create — types, options, required fields, rules, formulas, auto numbers, verification codes), plus reCAPTCHA when the form asks for it, and a per-address rate limit (app/core/throttle.py).""" import hashlib import secrets import uuid from datetime import datetime, timedelta, timezone import asyncio import mimetypes from fastapi import APIRouter, File, Form as FormParam, HTTPException, Request, UploadFile from fastapi.responses import JSONResponse, Response from jose import JWTError, jwt from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core import throttle from app.core.config import settings from app.core.database import AsyncSessionLocal from app.core.recaptcha import RecaptchaError, verify_recaptcha_token from app.core.security import AuthContext from app.crud import application_review, field_access, form_documents, form_schedule, form_verification, public_forms from app.models.upload import UploadedFile from app.models.workflow import FormSubmission from app.routers.form_submissions import _create, _get_form_version, _update, correct_returned from app.routers.uploads import UPLOAD_ROOT, _safe_filename, _write_sync from app.schemas.workflow import ( FormVerificationSent, FormVerificationToken, PublicFormOption, PublicFormRead, PublicFormSubmit, PublicDraftRead, PublicDraftSave, PublicDraftSaved, PublicFormSubmitted, PublicStatusConfirm, PublicStatusEvent, PublicStatusRead, PublicStatusResubmit, PublicStatusSend, PublicStatusToken, PublicStatusView, PublicVerificationConfirm, PublicVerificationSend, ) router = APIRouter(prefix="/public/forms", tags=["public forms"]) _NOT_FOUND = "This form isn't open to the public." SUBMIT_LIMIT = (10, 600.0) # per address and form, in 10 minutes SEND_LIMIT = (6, 600.0) DRAFT_LIMIT = (30, 600.0) RESUME_TTL = timedelta(days=30) DOCUMENT_TTL = timedelta(hours=1) _DOC_PURPOSE = "form_document" UPLOAD_LIMIT = (30, 600.0) STATUS_LIMIT = (10, 600.0) STATUS_TTL = timedelta(hours=1) _STATUS_PURPOSE = "application_status" _NO_APPLICATION = "No application with that number was found." def _document_token(tenant_id: uuid.UUID, form_id: uuid.UUID, submission_id: uuid.UUID) -> str: claims = {"purpose": _DOC_PURPOSE, "tid": str(tenant_id), "fid": str(form_id), "sid": str(submission_id), "exp": datetime.now(timezone.utc) + DOCUMENT_TTL} return jwt.encode(claims, settings.jwt_secret, algorithm=settings.jwt_algorithm) def _token_hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() async def _draft(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID, token: str | None) -> FormSubmission: """The live draft a "continue later" link's token opens — 404 otherwise.""" row = None if token: row = ( await db.execute( select(FormSubmission).where( FormSubmission.tenant_id == tenant_id, FormSubmission.form_id == form_id, FormSubmission.resume_token_hash == _token_hash(token), FormSubmission.status == "draft", FormSubmission.is_deleted.is_(False), FormSubmission.resume_expires_at > datetime.now(timezone.utc), ) ) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="This link has expired or was already used — start the form again.") return row async def _open(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID): found = await public_forms.live_public_form(db, tenant_id, form_id) if found is None: raise HTTPException(status_code=404, detail=_NOT_FOUND) return found @router.get("/{tenant_id}/{form_id}", response_model=PublicFormRead) async def get_public_form(tenant_id: uuid.UUID, form_id: uuid.UUID) -> PublicFormRead: async with AsyncSessionLocal() as db, db.begin(): tenant, form = await _open(db, tenant_id, form_id) # Fields hidden from everyone aren't sent; read-only ones are marked. access = field_access.effective_access(form.field_definitions, set(), None, True) fields = [f for f in form.field_definitions if access.get(f.get("name")) != "hidden"] form_settings = form.settings or {} captcha = bool(form_settings.get("public_captcha")) and settings.is_recaptcha_configured state, state_message = form_schedule.availability(form_settings) opens_at, closes_at = form_schedule.window(form_settings) return PublicFormRead( availability=state, availability_message=state_message, opens_at=opens_at, closes_at=closes_at, tenant_name=tenant.name, form_id=form.form_id, version=form.version, name=form.name, field_definitions=fields, field_access={k: v for k, v in access.items() if v == "read_only"}, events=form.events, settings={k: v for k, v in form_settings.items() if k not in ("test_scenarios",)}, recaptcha_site_key=settings.recaptcha_site_key if captcha else None, ) @router.get("/{tenant_id}/{form_id}/options/{field_name}", response_model=list[PublicFormOption]) async def get_public_form_options(tenant_id: uuid.UUID, form_id: uuid.UUID, field_name: str) -> list[PublicFormOption]: async with AsyncSessionLocal() as db, db.begin(): _tenant, form = await _open(db, tenant_id, form_id) records = await public_forms.option_records(db, tenant_id, form, field_name) if records is None: raise HTTPException(status_code=404, detail="This field has no list of options.") return [PublicFormOption.model_validate(r) for r in records] @router.post("/{tenant_id}/{form_id}/verify/send", response_model=FormVerificationSent) async def send_public_code(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicVerificationSend, request: Request) -> FormVerificationSent: throttle.check(f"form-code:{form_id}:{throttle.client_ip(request)}", *SEND_LIMIT, "Too many codes asked for — wait a few minutes and try again.") async with AsyncSessionLocal() as db, db.begin(): tenant, form = await _open(db, tenant_id, form_id) try: field = form_verification.verified_field(form.field_definitions, payload.field) verification_id, sent_to = await form_verification.send_code(db, tenant_id, tenant.name, form.form_id, form.name, field, payload.value) except form_verification.VerificationError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc return FormVerificationSent(verification_id=verification_id, sent_to=sent_to) @router.post("/{tenant_id}/{form_id}/verify/confirm", response_model=FormVerificationToken) async def confirm_public_code(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicVerificationConfirm) -> FormVerificationToken | JSONResponse: async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) try: token = await form_verification.confirm_code(db, tenant_id, form_id, payload.verification_id, payload.code) except form_verification.VerificationError as exc: # Answered rather than raised, so the wrong try is counted. return JSONResponse(status_code=exc.status_code, content={"detail": str(exc)}) return FormVerificationToken(token=token) @router.post("/{tenant_id}/{form_id}/drafts", response_model=PublicDraftSaved) async def save_public_draft(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicDraftSave, request: Request) -> PublicDraftSaved: """Save part-way (the form must allow drafts). The answers are checked for their shape only, as any draft's are; the private link's token is returned once — only its hash is kept — and lasts RESUME_TTL from the latest save.""" throttle.check(f"form-draft:{form_id}:{throttle.client_ip(request)}", *DRAFT_LIMIT) async with AsyncSessionLocal() as db, db.begin(): _tenant, form = await _open(db, tenant_id, form_id) if not (form.settings or {}).get("allow_drafts"): raise HTTPException(status_code=422, detail="This form can't be saved to finish later.") token = payload.resume_token or secrets.token_urlsafe(24) expires = datetime.now(timezone.utc) + RESUME_TTL anonymous = AuthContext(tenant_id=tenant_id, user_id=None, vendor_id=None) if payload.resume_token: draft = await _draft(db, tenant_id, form_id, payload.resume_token) draft_form = await _get_form_version(db, tenant_id, draft.form_id, draft.form_version) draft = await _update(db, anonymous, draft, draft_form, payload.field_values, submit=False) else: draft = await _create(db, anonymous, form, payload.field_values, draft=True) draft.resume_token_hash = _token_hash(token) draft.resume_expires_at = expires return PublicDraftSaved(resume_token=token, expires_at=expires) @router.get("/{tenant_id}/{form_id}/drafts/{token}", response_model=PublicDraftRead) async def get_public_draft(tenant_id: uuid.UUID, form_id: uuid.UUID, token: str) -> PublicDraftRead: async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) draft = await _draft(db, tenant_id, form_id, token) return PublicDraftRead(field_values=draft.field_values, expires_at=draft.resume_expires_at) @router.post("/{tenant_id}/{form_id}/submissions", response_model=PublicFormSubmitted, status_code=201) async def submit_public_form(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicFormSubmit, request: Request) -> PublicFormSubmitted: ip = throttle.client_ip(request) throttle.check(f"form-submit:{form_id}:{ip}", *SUBMIT_LIMIT, "Too many submissions from here — wait a few minutes and try again.") async with AsyncSessionLocal() as db, db.begin(): _tenant, form = await _open(db, tenant_id, form_id) if (form.settings or {}).get("public_captcha"): try: await verify_recaptcha_token(payload.captcha_token, ip) except RecaptchaError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc anonymous = AuthContext(tenant_id=tenant_id, user_id=None, vendor_id=None) if payload.resume_token: # A saved draft, finished: submitted on the version it was saved # on, and its link stops working. draft = await _draft(db, tenant_id, form_id, payload.resume_token) form = await _get_form_version(db, tenant_id, draft.form_id, draft.form_version) submission = await _update(db, anonymous, draft, form, payload.field_values, submit=True, verifications=payload.verifications) submission.resume_token_hash = None submission.resume_expires_at = None else: submission = await _create(db, anonymous, form, payload.field_values, draft=False, verifications=payload.verifications) # What the person may want to keep: the reference numbers they were given. references = { f.get("label") or f["name"]: str(submission.field_values.get(f["name"])) for f in form.field_definitions if f.get("type") == "autonumber" and submission.field_values.get(f["name"]) } return PublicFormSubmitted( id=submission.id, message=(form.settings or {}).get("public_message") or None, references=references, document_token=_document_token(tenant_id, form_id, submission.id) if form_documents.document_enabled(form) else None, ) @router.get("/{tenant_id}/{form_id}/document.pdf", response_class=Response) async def get_public_document(tenant_id: uuid.UUID, form_id: uuid.UUID, token: str) -> Response: """The PDF of what the visitor just submitted — only with the token the submission gave back (for an hour), never by the submission's id alone.""" try: claims = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) except JWTError: raise HTTPException(status_code=404, detail="This link has expired — the document can't be downloaded any more.") from None if claims.get("purpose") != _DOC_PURPOSE or claims.get("tid") != str(tenant_id) or claims.get("fid") != str(form_id): raise HTTPException(status_code=404, detail="This link has expired — the document can't be downloaded any more.") async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) submission = (await db.execute(select(FormSubmission).where(FormSubmission.form_id == form_id, FormSubmission.id == uuid.UUID(claims["sid"])))).scalar_one_or_none() if submission is None or submission.is_deleted or submission.form_id != form_id: raise HTTPException(status_code=404, detail="This submission can't be found.") form = await _get_form_version(db, tenant_id, submission.form_id, submission.form_version) if not form_documents.document_enabled(form): raise HTTPException(status_code=404, detail="This form has no document to download.") # Fields hidden from everyone stay out of it. access = field_access.effective_access(form.field_definitions, set(), None, True) values = field_access.without_hidden(submission.field_values, access) pdf = await form_documents.submission_pdf(db, tenant_id, form, submission, values) name = form_documents.pdf_file_name(form.name, form_documents.reference_of(form.field_definitions, values)) return Response(pdf, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{name}"', "Cache-Control": "no-store"}) # ---- Files a visitor attaches ------------------------------------------------------------------------- def _accepts(accept: str | None, filename: str, content_type: str) -> bool: """The field's accept list (".pdf,image/*") — read as the browser does.""" if not (accept or "").strip(): return True name, ctype = filename.lower(), (content_type or "").lower() for raw in accept.split(","): item = raw.strip().lower() if not item: continue if item.startswith(".") and name.endswith(item): return True if item.endswith("/*") and ctype.startswith(item[:-1]): return True if item == ctype: return True return False @router.post("/{tenant_id}/{form_id}/files", status_code=201) async def upload_public_file( tenant_id: uuid.UUID, form_id: uuid.UUID, request: Request, field: str = FormParam(...), file: UploadFile = File(...) ) -> dict: """A file for one of the form's file fields — its type and size checked against the field (and at most PUBLIC_UPLOAD_MAX_MB). It can only be attached to a submission of a public form, and only staff can open it.""" throttle.check(f"form-upload:{form_id}:{throttle.client_ip(request)}", *UPLOAD_LIMIT, "Too many uploads from here — wait a few minutes and try again.") async with AsyncSessionLocal() as db, db.begin(): _tenant, form = await _open(db, tenant_id, form_id) target = public_forms.file_field(form.field_definitions, field) if target is None: raise HTTPException(status_code=422, detail="This form has no such file field.") filename = file.filename or "file" content_type = file.content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream" if not _accepts(target.get("accept"), filename, content_type): raise HTTPException(status_code=422, detail=f"{filename} isn't an accepted file type ({target.get('accept')}).") limit_mb = min(float(target.get("max_file_size_mb") or public_forms.PUBLIC_UPLOAD_MAX_MB), public_forms.PUBLIC_UPLOAD_MAX_MB) data = await file.read(int(limit_mb * 1024 * 1024) + 1) if len(data) > limit_mb * 1024 * 1024: raise HTTPException(status_code=413, detail=f"{filename} is larger than {limit_mb:g} MB.") if not data: raise HTTPException(status_code=422, detail=f"{filename} is empty.") file_id = uuid.uuid4() path = UPLOAD_ROOT / str(tenant_id) / f"{file_id}_{_safe_filename(filename)}" await asyncio.to_thread(_write_sync, path, data) db.add( UploadedFile( id=file_id, tenant_id=tenant_id, filename=filename[:255], content_type=content_type[:127], size_bytes=len(data), storage_path=str(path), public_form_id=form_id ) ) return {"id": str(file_id), "filename": filename, "content_type": content_type, "size_bytes": len(data)} @router.get("/{tenant_id}/{form_id}/files/{file_id}") async def get_public_file(tenant_id: uuid.UUID, form_id: uuid.UUID, file_id: uuid.UUID) -> dict: """Name and size of a file uploaded to this form — what a visitor's own form shows for it. Never the content.""" async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) row = ( await db.execute(select(UploadedFile).where(UploadedFile.tenant_id == tenant_id, UploadedFile.id == file_id, UploadedFile.public_form_id == form_id)) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="This file can't be found.") return {"id": str(row.id), "filename": row.filename, "content_type": row.content_type, "size_bytes": row.size_bytes} # ---- An applicant's status page (app/crud/application_review.py) --------------------------------------- def _lookup_fields(form) -> tuple[dict, dict]: """(the reference field, the email/phone field the code goes to) — 404 unless the form lets applicants look themselves up.""" review = application_review.review_settings(form) defs = {f.get("name"): f for f in form.field_definitions} ref, verify = defs.get(review.get("lookup_reference_field")), defs.get(review.get("lookup_verify_field")) if not (review.get("enabled") and review.get("status_lookup")) or ref is None or verify is None or verify.get("type") not in ("email", "phone"): raise HTTPException(status_code=404, detail="This form has no status page.") return ref, verify async def _by_reference(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID, field: str, reference: str) -> FormSubmission | None: stored = func.lower(func.trim(FormSubmission.field_values.op("->>")(field))) return ( await db.execute( select(FormSubmission) .where( FormSubmission.tenant_id == tenant_id, FormSubmission.form_id == form_id, FormSubmission.is_deleted.is_(False), FormSubmission.status == "submitted", stored == reference.strip().lower(), ) .order_by(FormSubmission.submitted_at.desc()) .limit(1) ) ).scalar_one_or_none() def _status_token(tenant_id: uuid.UUID, form_id: uuid.UUID, submission_id: uuid.UUID) -> str: claims = {"purpose": _STATUS_PURPOSE, "tid": str(tenant_id), "fid": str(form_id), "sid": str(submission_id), "exp": datetime.now(timezone.utc) + STATUS_TTL} return jwt.encode(claims, settings.jwt_secret, algorithm=settings.jwt_algorithm) async def _from_status_token(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID, token: str) -> FormSubmission: expired = HTTPException(status_code=401, detail="Your session has ended — look your application up again.") try: claims = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) except JWTError: raise expired from None if claims.get("purpose") != _STATUS_PURPOSE or claims.get("tid") != str(tenant_id) or claims.get("fid") != str(form_id): raise expired row = ( await db.execute( select(FormSubmission).where(FormSubmission.form_id == form_id, FormSubmission.id == uuid.UUID(claims["sid"]), FormSubmission.tenant_id == tenant_id) ) ).scalar_one_or_none() if row is None or row.is_deleted: raise HTTPException(status_code=404, detail=_NO_APPLICATION) return row @router.post("/{tenant_id}/{form_id}/status/send") async def send_status_code(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicStatusSend, request: Request) -> dict: """A code to the email / phone the application gave — the reference number alone never shows anything.""" throttle.check(f"form-status:{form_id}:{throttle.client_ip(request)}", *STATUS_LIMIT, "Too many lookups from here — wait a few minutes and try again.") async with AsyncSessionLocal() as db, db.begin(): tenant, form = await _open(db, tenant_id, form_id) ref_field, verify_field = _lookup_fields(form) submission = await _by_reference(db, tenant_id, form_id, ref_field["name"], payload.reference) target = (submission.field_values or {}).get(verify_field["name"]) if submission else None if submission is None or not target: raise HTTPException(status_code=404, detail=_NO_APPLICATION) try: verification_id, sent_to = await form_verification.send_code(db, tenant_id, tenant.name, form.form_id, form.name, verify_field, target) except form_verification.VerificationError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc return {"verification_id": str(verification_id), "sent_to": sent_to} @router.post("/{tenant_id}/{form_id}/status/confirm", response_model=PublicStatusToken) async def confirm_status_code(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicStatusConfirm) -> PublicStatusToken | JSONResponse: async with AsyncSessionLocal() as db, db.begin(): _tenant, form = await _open(db, tenant_id, form_id) ref_field, verify_field = _lookup_fields(form) submission = await _by_reference(db, tenant_id, form_id, ref_field["name"], payload.reference) if submission is None: raise HTTPException(status_code=404, detail=_NO_APPLICATION) try: token = await form_verification.confirm_code(db, tenant_id, form_id, payload.verification_id, payload.code) except form_verification.VerificationError as exc: # Answered rather than raised, so the wrong try is counted. return JSONResponse(status_code=exc.status_code, content={"detail": str(exc)}) target = form_verification.normalized_target(verify_field, (submission.field_values or {}).get(verify_field["name"])) if target is None or not form_verification._token_ok(token, tenant_id, form_id, verify_field["name"], target): raise HTTPException(status_code=404, detail=_NO_APPLICATION) return PublicStatusToken(token=_status_token(tenant_id, form_id, submission.id)) async def _status_read(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID, submission: FormSubmission) -> PublicStatusRead: version = await _get_form_version(db, tenant_id, submission.form_id, submission.form_version) _live_tenant, live = await _open(db, tenant_id, form_id) ref_field, _verify = _lookup_fields(live) rows = await application_review.history(db, tenant_id, submission.id) status = application_review.current(submission) history = [PublicStatusEvent(status=r.to_status, label=application_review.LABELS.get(r.to_status, r.to_status), note=r.note, at=r.occurred_at) for r in rows] if not history: history = [PublicStatusEvent(status="submitted", label=application_review.LABELS["submitted"], at=submission.submitted_at)] correcting = application_review.can_correct(submission) access = field_access.effective_access(version.field_definitions, set(), None, True) fields = [f for f in version.field_definitions if access.get(f.get("name")) != "hidden"] if correcting else [] return PublicStatusRead( reference=str((submission.field_values or {}).get(ref_field["name"]) or ""), form_name=version.name, status=status, label=application_review.LABELS.get(status, status), note=submission.review_note, submitted_at=submission.submitted_at, history=history, can_correct=correcting, correction_until=submission.correction_until if correcting else None, correction_fields=submission.correction_fields if correcting else None, field_definitions=fields, field_values=field_access.without_hidden(submission.field_values, access) if correcting else {}, ) @router.post("/{tenant_id}/{form_id}/status/view", response_model=PublicStatusRead) async def view_status(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicStatusView) -> PublicStatusRead: async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) submission = await _from_status_token(db, tenant_id, form_id, payload.token) return await _status_read(db, tenant_id, form_id, submission) @router.post("/{tenant_id}/{form_id}/status/resubmit", response_model=PublicStatusRead) async def resubmit_from_status(tenant_id: uuid.UUID, form_id: uuid.UUID, payload: PublicStatusResubmit, request: Request) -> PublicStatusRead: throttle.check(f"form-submit:{form_id}:{throttle.client_ip(request)}", *SUBMIT_LIMIT, "Too many submissions from here — wait a few minutes and try again.") async with AsyncSessionLocal() as db, db.begin(): await _open(db, tenant_id, form_id) submission = await _from_status_token(db, tenant_id, form_id, payload.token) anonymous = AuthContext(tenant_id=tenant_id, user_id=None, vendor_id=None) updated, _form = await correct_returned(db, anonymous, submission, payload.field_values, payload.verifications) return await _status_read(db, tenant_id, form_id, updated)