"""Private Core Forms submit route. Public callers use Dashboard/BFF or installed-app BFF. Not a capability invoke. Not /api/contact. """ from __future__ import annotations import hmac from uuid import UUID from fastapi import APIRouter, Depends, Request from fastapi.responses import JSONResponse from sqlalchemy.ext.asyncio import AsyncSession from src.core.database import get_db from src.core.dependencies import get_productive_access_policy from src.core.security.encryption import decrypt_value from src.config import get_settings from src.domain.auth.models import User from src.domain.auth.router import get_current_user_optional from src.domain.billing.productive_access_policy import ProductiveAccessPolicy from src.domain.edge_functions.repository import ProjectFunctionTokenRepository from src.domain.forms.anti_abuse import InMemoryAttemptLimiter from src.domain.forms.cloud_directory import ( cloud_form_file_lookup, hydrate_form_file_directory, ) from src.domain.forms.composition import OverlayFormRuntimeStore from src.domain.forms.composition_store import load_composition_form_store from src.domain.forms.files import InMemoryFormFileDirectory from src.domain.forms.persistence import ( FormPersistenceContext, FormPersistenceError, InMemoryFormSubmissionStore, persist_form_submission_async, ) from src.integrations.cloud_control_client import get_cloud_client_for_workspace from src.domain.forms.public import ( FormPublicResult, FormPublicSubmitRequest, InMemoryFormRuntimeStore, SilentCapabilityInvoker, UnwiredChallengeVerifier, client_identity_from_request, parse_public_submit_request, submit_form, ) from src.domain.projects.repository import ProjectRepository forms_router = APIRouter( prefix="/api/projects/{project_id}/forms", tags=["forms"], ) runtime_store = InMemoryFormRuntimeStore() file_directory = InMemoryFormFileDirectory() attempt_limiter = InMemoryAttemptLimiter() capability_invoker = SilentCapabilityInvoker() challenge_verifier = UnwiredChallengeVerifier() def _bearer_token(request: Request) -> str: header = request.headers.get("authorization") or "" if header.lower().startswith("bearer "): return header[7:].strip() return "" async def _project_service_token_ok( db: AsyncSession, project_id: UUID, token: str ) -> bool: if not token: return False row = await ProjectFunctionTokenRepository(db).get(project_id) if row is None: return False try: expected = decrypt_value(row.token_enc) except Exception: return False if not expected or len(expected) != len(token): return False return hmac.compare_digest(expected, token) def _cloud_internal_token_ok(token: str) -> bool: """Allow the Cloud BFF to relay public form transport to Core. Cloud is the authenticated public facade for W4P. It forwards the fixed internal token, while Core remains authoritative for the project lookup, form definition, billing policy, and persistence. """ configured = (get_settings().cloud_internal_token or "").strip() return bool(configured and token and hmac.compare_digest(configured, token)) @forms_router.post("/{form_key}/submissions") async def submit_project_form( request: Request, form_key: str, project_id: UUID, db: AsyncSession = Depends(get_db), user: User | None = Depends(get_current_user_optional), policy: ProductiveAccessPolicy = Depends(get_productive_access_policy), ) -> JSONResponse: token = _bearer_token(request) service_ok = await _project_service_token_ok(db, project_id, token) or _cloud_internal_token_ok(token) if not service_ok: if user is None: return JSONResponse( status_code=401, content={"ok": False, "error": "form_unavailable"}, ) project = await ProjectRepository(db).get_by_id(project_id) if not project: return JSONResponse( status_code=404, content={"ok": False, "error": "form_unavailable"}, ) await policy.require_for_workspace(project.workspace_id, user.id) try: payload = await request.json() except Exception: return JSONResponse( status_code=400, content={"ok": False, "error": "invalid_request"}, ) parsed = parse_public_submit_request(payload) if isinstance(parsed, FormPublicResult): return JSONResponse(status_code=parsed.http_status, content=parsed.body) assert isinstance(parsed, FormPublicSubmitRequest) identity = client_identity_from_request( forwarded=request.headers.get("cf-connecting-ip") or request.headers.get("x-real-ip"), peer=request.client.host if request.client else None, ) composition = await load_composition_form_store(db, project_id) runtimes = OverlayFormRuntimeStore(runtime_store, composition) files = file_directory runtime = runtimes.get(str(project_id), form_key) if runtime is not None and any(field.type == "file" for field in runtime.definition.fields): files = await hydrate_form_file_directory( project_id=str(project_id), definition=runtime.definition, values=parsed.values, lookup=cloud_form_file_lookup, ) # The synchronous runtime contract still receives the deterministic # adapter used by domain tests and post-submit actions. Core persistence # is authoritative: the accepted envelope is immediately written to the # durable repository before this route returns success. request_store = InMemoryFormSubmissionStore() result = submit_form( project_id=str(project_id), form_key=form_key, values=parsed.values, challenge=parsed.challenge, client_identity=identity, runtimes=runtimes, invoker=capability_invoker, submissions=request_store, files=files, verifier=challenge_verifier, limiter=attempt_limiter, persist_required=True, ) if result.http_status == 200 and result.submission_id: accepted_row = request_store.get( FormPersistenceContext(project_id=str(project_id)), result.submission_id ) if accepted_row is None: return JSONResponse( status_code=500, content={"ok": False, "error": "submission_failed"}, ) try: await persist_form_submission_async( db, FormPersistenceContext(project_id=str(project_id)), accepted_row.accepted, ) except FormPersistenceError: return JSONResponse( status_code=500, content={"ok": False, "error": "submission_failed"}, ) return JSONResponse(status_code=result.http_status, content=result.body) @forms_router.post("/{form_key}/files") async def upload_project_form_file( request: Request, form_key: str, project_id: UUID, db: AsyncSession = Depends(get_db), ) -> JSONResponse: """Upload a private application file for a public form. The installed-app BFF authenticates with the project function token. The browser never receives Cloud credentials, and the file is still validated against the server-owned form definition before it is accepted. """ token = _bearer_token(request) if not (await _project_service_token_ok(db, project_id, token) or _cloud_internal_token_ok(token)): return JSONResponse(status_code=401, content={"ok": False, "error": "form_unavailable"}) composition = await load_composition_form_store(db, project_id) runtime = OverlayFormRuntimeStore(runtime_store, composition).get(str(project_id), form_key) if runtime is None or not any(field.type == "file" for field in runtime.definition.fields): return JSONResponse(status_code=404, content={"ok": False, "error": "form_unavailable"}) try: body = await request.json() except Exception: return JSONResponse(status_code=400, content={"ok": False, "error": "invalid_request"}) if not isinstance(body, dict): return JSONResponse(status_code=400, content={"ok": False, "error": "invalid_request"}) field_key = body.get("field_key") if not isinstance(field_key, str) or not any( field.type == "file" and field.key == field_key for field in runtime.definition.fields ): return JSONResponse(status_code=400, content={"ok": False, "error": "invalid_file_field"}) filename = body.get("filename") content_type = body.get("content_type") or "application/octet-stream" content_base64 = body.get("content_base64") if not isinstance(filename, str) or not filename.strip() or not isinstance(content_base64, str) or not content_base64: return JSONResponse(status_code=400, content={"ok": False, "error": "invalid_request"}) # Keep the JSON transport bounded; large uploads should use the separate # resumable storage surface rather than a public form request. if len(content_base64) > 14_000_000: return JSONResponse(status_code=413, content={"ok": False, "error": "file_too_large"}) project = await ProjectRepository(db).get_by_id(project_id) if not project: return JSONResponse(status_code=404, content={"ok": False, "error": "form_unavailable"}) cloud = await get_cloud_client_for_workspace(project.workspace_id, db) uploaded = await cloud.upload_application_file( project_id=str(project_id), filename=filename.strip()[:255], content_type=str(content_type)[:160], content_base64=content_base64, ) if not uploaded.get("ok") or not uploaded.get("file_id"): return JSONResponse(status_code=503, content={"ok": False, "error": "upload_failed"}) return JSONResponse(status_code=201, content={ "ok": True, "file_id": uploaded["file_id"], "filename": uploaded.get("filename", filename.strip()[:255]), "content_type": uploaded.get("content_type", content_type), "size_bytes": uploaded.get("size_bytes"), })