Artur Mukhamadiev
d77b52ae86
- tests/integration: frontend served by backend, runtime error mapping, backend + real pi image with scripted model/Confluence (shared example, variants, failure paths, cancellation/busy gate, isolation canaries, HTTP download ownership, 16 MiB prompt round trip), real OpenAI-compatible adapter + real image over a scripted transport, real Chrome against the real backend with a scripted runtime peer, backend crash/restart reconciliation, and an opt-in live model check (marker: live). - backend: map runtime terminal codes (model_output_limit, model_context_exceeded, query_timeout, connectivity_failed) to the contract's HTTP statuses; make the artifact 404 body identical for no-session, wrong-session, unknown and expired IDs. - Makefile, scripts/run-backend.sh, deploy/confluence-web.env.example, root README for the integrated application; integration pytest marker.
546 lines
21 KiB
Python
546 lines
21 KiB
Python
"""FastAPI application factory, endpoints, session handling, and security headers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
import secrets
|
|
import time
|
|
import urllib.parse
|
|
from typing import Any, Callable, Dict, Optional, Tuple
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, Response
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, ConfigDict
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from backend.artifacts import ArtifactStore
|
|
from backend.confluence import ConfluenceClient
|
|
from backend.containers import (
|
|
ContainerManager,
|
|
DockerContainerManager,
|
|
FakeContainerManager,
|
|
)
|
|
from backend.errors import (
|
|
AppError,
|
|
ArtifactNotFoundError,
|
|
InvalidInputError,
|
|
OriginDeniedError,
|
|
ConnectivityFailedError,
|
|
sanitize_message,
|
|
RequestTooLargeError,
|
|
)
|
|
from backend.model import FakeModelAdapter, ModelAdapter, OpenAIModelAdapter
|
|
from backend.runner import QueryRunner
|
|
from backend.settings import Settings, validate_confluence_url
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SESSION_COOKIE_NAME = "cw_session"
|
|
|
|
CSP_POLICY = (
|
|
"default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; "
|
|
"img-src 'none'; media-src 'none'; font-src 'self'; object-src 'none'; "
|
|
"frame-src 'none'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'"
|
|
)
|
|
|
|
FRONTEND_ASSET_DIRS = ("css", "js", "vendor")
|
|
|
|
MAX_VERIFY_BODY_BYTES = 128 * 1024
|
|
MAX_QUERY_BODY_BYTES = 6 * 16 * 1024 * 1024 + 64 * 1024
|
|
|
|
|
|
class AuthVerifyRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
url: str
|
|
pat: str
|
|
|
|
|
|
class QueryCredentials(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
url: str
|
|
pat: str
|
|
|
|
|
|
class QueryRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
prompt: str
|
|
credentials: QueryCredentials
|
|
|
|
|
|
class ServerSessionStore:
|
|
"""In-memory store of server-issued session IDs to prevent client-forged sessions."""
|
|
|
|
def __init__(self) -> None:
|
|
self._sessions: Dict[str, float] = {}
|
|
|
|
def is_valid(self, session_id: Optional[str]) -> bool:
|
|
if not session_id or not isinstance(session_id, str):
|
|
return False
|
|
if len(session_id) != 64: # secrets.token_hex(32) is 64 hex characters
|
|
return False
|
|
return session_id in self._sessions
|
|
|
|
def create_session(self) -> str:
|
|
session_id = secrets.token_hex(32)
|
|
self._sessions[session_id] = time.time()
|
|
return session_id
|
|
|
|
def touch(self, session_id: str) -> None:
|
|
if session_id in self._sessions:
|
|
self._sessions[session_id] = time.time()
|
|
|
|
def prune_stale(self, max_age_seconds: float = 86400.0) -> None:
|
|
now = time.time()
|
|
stale = [s for s, ts in self._sessions.items() if now - ts > max_age_seconds]
|
|
for s in stale:
|
|
self._sessions.pop(s, None)
|
|
|
|
|
|
def get_or_set_session_cookie(
|
|
request: Request, response: Response, session_store: ServerSessionStore
|
|
) -> str:
|
|
"""Read server-issued cw_session cookie or issue a new cryptographically random session ID."""
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if session_store.is_valid(session_id):
|
|
session_store.touch(session_id) # type: ignore[arg-type]
|
|
else:
|
|
session_id = session_store.create_session()
|
|
|
|
secure = request.url.scheme == "https"
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE_NAME,
|
|
value=session_id,
|
|
httponly=True,
|
|
samesite="strict",
|
|
path="/",
|
|
secure=secure,
|
|
)
|
|
return session_id
|
|
|
|
|
|
def _normalize_endpoint(scheme: str, hostname: Optional[str], port: Optional[int]) -> Tuple[str, str, int]:
|
|
s = scheme.lower()
|
|
h = (hostname or "").lower()
|
|
if port is not None:
|
|
p = port
|
|
else:
|
|
p = 443 if s == "https" else (80 if s == "http" else 0)
|
|
return (s, h, p)
|
|
|
|
|
|
def check_same_origin(request: Request) -> None:
|
|
"""Require matching Origin for state-changing POST requests."""
|
|
origin = request.headers.get("origin")
|
|
if not origin:
|
|
raise OriginDeniedError("Missing Origin header")
|
|
|
|
try:
|
|
parsed_origin = urllib.parse.urlsplit(origin)
|
|
except ValueError:
|
|
raise OriginDeniedError("Malformed Origin header")
|
|
if parsed_origin.scheme not in ("http", "https") or not parsed_origin.netloc or parsed_origin.username is not None or parsed_origin.password is not None or parsed_origin.path or parsed_origin.query or parsed_origin.fragment:
|
|
raise OriginDeniedError("Malformed Origin header")
|
|
|
|
try:
|
|
origin_norm = _normalize_endpoint(
|
|
parsed_origin.scheme,
|
|
parsed_origin.hostname,
|
|
parsed_origin.port,
|
|
)
|
|
req_norm = _normalize_endpoint(
|
|
request.url.scheme,
|
|
request.url.hostname,
|
|
request.url.port,
|
|
)
|
|
except ValueError:
|
|
raise OriginDeniedError("Invalid origin or destination")
|
|
|
|
if origin_norm != req_norm:
|
|
raise OriginDeniedError("Origin does not match request destination")
|
|
|
|
|
|
async def _periodic_maintenance(runner, store, session_store) -> None:
|
|
while True:
|
|
await asyncio.sleep(60)
|
|
if not runner._gate.locked():
|
|
try:
|
|
await runner.reconcile()
|
|
except Exception:
|
|
logger.warning("Container reconciliation failed; query admission remains closed")
|
|
try:
|
|
await asyncio.to_thread(store.expire_artifacts)
|
|
session_store.prune_stale()
|
|
except Exception:
|
|
logger.warning("Artifact or session maintenance failed")
|
|
|
|
|
|
class SecurityMiddleware:
|
|
"""ASGI middleware bounds incoming bytes without stealing disconnect events."""
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope["type"] != "http":
|
|
return await self.app(scope, receive, send)
|
|
request = Request(scope)
|
|
limit = MAX_VERIFY_BODY_BYTES if scope["path"] == "/api/v1/auth/verify" else MAX_QUERY_BODY_BYTES
|
|
total = 0
|
|
exceeded = False
|
|
|
|
async def bounded_receive():
|
|
nonlocal total, exceeded
|
|
message = await receive()
|
|
if message["type"] == "http.request":
|
|
total += len(message.get("body", b""))
|
|
if total > limit:
|
|
exceeded = True
|
|
raise RequestTooLargeError("Request body exceeds byte limit")
|
|
return message
|
|
|
|
async def secure_send(message):
|
|
if message["type"] == "http.response.start":
|
|
headers = list(message.get("headers", []))
|
|
if scope["path"].startswith("/api/v1/"):
|
|
headers = [(k, v) for k, v in headers if k.lower() != b"cache-control"]
|
|
headers.append((b"cache-control", b"no-store"))
|
|
else:
|
|
headers.extend([(b"content-security-policy", CSP_POLICY.encode()), (b"referrer-policy", b"no-referrer")])
|
|
if not any(k.lower() == b"x-content-type-options" for k, _ in headers):
|
|
headers.append((b"x-content-type-options", b"nosniff"))
|
|
message = {**message, "headers": headers}
|
|
await send(message)
|
|
|
|
if scope["method"] == "POST":
|
|
header = request.headers.get("content-length")
|
|
if header is not None:
|
|
try:
|
|
length = int(header)
|
|
if length < 0:
|
|
raise ValueError
|
|
except ValueError:
|
|
return await JSONResponse(status_code=400, content=InvalidInputError("Invalid Content-Length").to_envelope())(scope, receive, secure_send)
|
|
if length > limit:
|
|
return await JSONResponse(status_code=413, content=RequestTooLargeError().to_envelope())(scope, receive, secure_send)
|
|
# Bound before framework parsing: Starlette otherwise translates read exceptions to HTTP 400.
|
|
chunks = bytearray()
|
|
while True:
|
|
try:
|
|
message = await bounded_receive()
|
|
except RequestTooLargeError as exc:
|
|
return await JSONResponse(status_code=413, content=exc.to_envelope())(scope, receive, secure_send)
|
|
if message["type"] == "http.disconnect":
|
|
return
|
|
chunks.extend(message.get("body", b""))
|
|
if not message.get("more_body", False):
|
|
break
|
|
delivered = False
|
|
async def replay_receive():
|
|
nonlocal delivered
|
|
if not delivered:
|
|
delivered = True
|
|
return {"type": "http.request", "body": bytes(chunks), "more_body": False}
|
|
return await receive()
|
|
return await self.app(scope, replay_receive, secure_send)
|
|
return await self.app(scope, receive, secure_send)
|
|
|
|
|
|
class OwnedFileResponse(FileResponse):
|
|
def __init__(self, *args, store, artifact_id, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.store, self.artifact_id = store, artifact_id
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
try:
|
|
return await super().__call__(scope, receive, send)
|
|
finally:
|
|
await asyncio.to_thread(self.store.release_reader, self.artifact_id)
|
|
|
|
|
|
def create_app(
|
|
settings: Optional[Settings] = None,
|
|
container_manager: Optional[ContainerManager] = None,
|
|
artifact_store: Optional[ArtifactStore] = None,
|
|
model_adapter: Optional[ModelAdapter] = None,
|
|
confluence_client_factory: Optional[Callable[..., ConfluenceClient]] = None,
|
|
) -> FastAPI:
|
|
"""Create and configure FastAPI application with dependency injection."""
|
|
app_settings = settings or Settings.from_env()
|
|
store = artifact_store or ArtifactStore(app_settings.artifact_storage_dir)
|
|
|
|
# Container manager selection
|
|
if container_manager is not None:
|
|
mgr = container_manager
|
|
elif app_settings.dev_mode:
|
|
mgr = FakeContainerManager()
|
|
else:
|
|
mgr = DockerContainerManager(app_settings)
|
|
|
|
# Model adapter selection
|
|
if model_adapter is not None:
|
|
adapter = model_adapter
|
|
elif app_settings.dev_mode or app_settings.model_provider == "fake":
|
|
adapter = FakeModelAdapter()
|
|
elif app_settings.model_provider == "openai":
|
|
if not app_settings.model_api_key or not app_settings.model_api_key.strip():
|
|
raise RuntimeError("Missing model_api_key for OpenAI provider in production mode")
|
|
adapter = OpenAIModelAdapter(
|
|
api_key=app_settings.model_api_key,
|
|
model_name=app_settings.model_name,
|
|
endpoint=app_settings.model_endpoint,
|
|
context_window_tokens=app_settings.model_context_window_tokens,
|
|
max_output_tokens=app_settings.model_max_output_tokens,
|
|
timeout=app_settings.model_timeout_seconds,
|
|
)
|
|
else:
|
|
raise RuntimeError(f"Unknown model_provider '{app_settings.model_provider}' in production mode")
|
|
|
|
if confluence_client_factory is None and app_settings.dev_mode:
|
|
from backend.dev.fake_confluence import create_client
|
|
confluence_client_factory = create_client
|
|
|
|
runner = QueryRunner(
|
|
settings=app_settings,
|
|
container_manager=mgr,
|
|
artifact_store=store,
|
|
model_adapter=adapter,
|
|
confluence_client_factory=confluence_client_factory,
|
|
)
|
|
|
|
session_store = ServerSessionStore()
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def lifespan(fastapi_app: FastAPI):
|
|
maintenance_task = None
|
|
try:
|
|
await asyncio.to_thread(store.purge_all)
|
|
if isinstance(mgr, DockerContainerManager):
|
|
await mgr.verify_rootless()
|
|
await runner.reconcile()
|
|
maintenance_task = asyncio.create_task(_periodic_maintenance(runner, store, session_store))
|
|
yield
|
|
finally:
|
|
if maintenance_task is not None:
|
|
maintenance_task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await maintenance_task
|
|
await adapter.close()
|
|
|
|
app = FastAPI(title="Confluence Web Backend", docs_url=None, redoc_url=None, lifespan=lifespan)
|
|
|
|
# Bounded concurrency semaphore for auth verify
|
|
verify_semaphore = asyncio.Semaphore(10)
|
|
|
|
# Error envelope exception handlers
|
|
@app.exception_handler(AppError)
|
|
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=exc.to_envelope(),
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
|
|
# Sanitize framework errors: NEVER echo back request body or PAT
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"error": {"code": "invalid_input", "message": "Invalid request payload"}},
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(request: Request, exc: Any) -> JSONResponse:
|
|
code_map = {
|
|
400: "invalid_input",
|
|
403: "origin_denied",
|
|
404: "artifact_not_found",
|
|
405: "invalid_input",
|
|
409: "busy",
|
|
413: "request_too_large",
|
|
502: "upstream_failed",
|
|
504: "query_timeout",
|
|
500: "execution_failed",
|
|
}
|
|
if exc.status_code == 404:
|
|
if request.url.path.startswith("/api/v1/artifacts/"):
|
|
code = "artifact_not_found"
|
|
else:
|
|
code = "invalid_input"
|
|
elif exc.status_code == 405:
|
|
code = "invalid_input"
|
|
else:
|
|
code = code_map.get(exc.status_code, "execution_failed")
|
|
msg = str(exc.detail) if exc.detail else "HTTP error"
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={"error": {"code": code, "message": sanitize_message(msg)}},
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": {"code": "execution_failed", "message": "Internal execution failure"}},
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
app.add_middleware(SecurityMiddleware)
|
|
app.state.runner = runner
|
|
app.state.artifact_store = store
|
|
|
|
@app.get("/")
|
|
async def root_index(request: Request):
|
|
if app_settings.frontend_dist_dir and (app_settings.frontend_dist_dir / "index.html").exists():
|
|
index_path = app_settings.frontend_dist_dir / "index.html"
|
|
html_content = await asyncio.to_thread(index_path.read_text, encoding="utf-8")
|
|
resp = HTMLResponse(content=html_content, status_code=200)
|
|
else:
|
|
resp = JSONResponse(content={"status": "ok", "service": "confluence-web-backend"})
|
|
get_or_set_session_cookie(request, resp, session_store)
|
|
return resp
|
|
|
|
@app.post("/api/v1/auth/verify")
|
|
async def verify_auth(req: AuthVerifyRequest, request: Request):
|
|
check_same_origin(request)
|
|
|
|
if len(req.pat.encode("utf-8")) > 8192 or not req.pat.strip():
|
|
raise InvalidInputError("Invalid Confluence PAT")
|
|
try:
|
|
async with asyncio.timeout(15), verify_semaphore:
|
|
# Validate Confluence URL against approved origins
|
|
canonical_url = validate_confluence_url(req.url, app_settings.approved_confluence_origins)
|
|
|
|
# Request-scoped verification client
|
|
client_factory = confluence_client_factory or ConfluenceClient
|
|
client_kwargs: Dict[str, Any] = {}
|
|
if app_settings.confluence_proxy:
|
|
client_kwargs["proxy"] = app_settings.confluence_proxy
|
|
async with client_factory(
|
|
base_url=canonical_url,
|
|
pat=req.pat,
|
|
approved_origins=app_settings.approved_confluence_origins,
|
|
corporate_ca_path=app_settings.corporate_ca_path,
|
|
timeout=15.0,
|
|
**client_kwargs,
|
|
) as client:
|
|
await client.verify_auth()
|
|
except asyncio.TimeoutError as exc:
|
|
raise ConnectivityFailedError("Confluence verification timed out") from exc
|
|
|
|
resp = JSONResponse(content={"valid": True})
|
|
get_or_set_session_cookie(request, resp, session_store)
|
|
return resp
|
|
|
|
@app.post("/api/v1/query")
|
|
async def execute_query(req: QueryRequest, request: Request):
|
|
check_same_origin(request)
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if not session_store.is_valid(session_id):
|
|
session_id = session_store.create_session()
|
|
else:
|
|
session_store.touch(session_id) # type: ignore[arg-type]
|
|
|
|
# Execute query via runner
|
|
result = await runner.run(
|
|
prompt=req.prompt,
|
|
confluence_url=req.credentials.url,
|
|
confluence_pat=req.credentials.pat,
|
|
session_id=session_id,
|
|
is_disconnected=request.is_disconnected,
|
|
)
|
|
resp = JSONResponse(content=result)
|
|
# Ensure session cookie is set
|
|
secure = request.url.scheme == "https"
|
|
resp.set_cookie(
|
|
key=SESSION_COOKIE_NAME,
|
|
value=session_id,
|
|
httponly=True,
|
|
samesite="strict",
|
|
path="/",
|
|
secure=secure,
|
|
)
|
|
return resp
|
|
|
|
@app.get("/api/v1/artifacts/{artifact_id}")
|
|
async def download_artifact(artifact_id: str, request: Request):
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if not session_store.is_valid(session_id):
|
|
raise ArtifactNotFoundError("Artifact not found")
|
|
|
|
lookup_task = asyncio.create_task(asyncio.to_thread(store.get_artifact_for_download, artifact_id, session_id=session_id))
|
|
cancelled = False
|
|
while not lookup_task.done():
|
|
try:
|
|
await asyncio.shield(lookup_task)
|
|
except asyncio.CancelledError:
|
|
cancelled = True
|
|
res = lookup_task.result()
|
|
if cancelled:
|
|
if res:
|
|
await asyncio.to_thread(store.release_reader, artifact_id)
|
|
raise asyncio.CancelledError
|
|
if not res:
|
|
raise ArtifactNotFoundError("Artifact not found") # identical to unknown-ID and no-session responses
|
|
|
|
file_path, display_name, size_bytes = res
|
|
|
|
# RFC 6266 filename encoding
|
|
ascii_fallback = "".join(c if 0x20 <= ord(c) < 0x7F and c not in '"\\;' else '_' for c in display_name)
|
|
if not ascii_fallback.strip("_ "):
|
|
ascii_fallback = "artifact.bin"
|
|
|
|
encoded_name = urllib.parse.quote(display_name, encoding="utf-8")
|
|
content_disposition = f'attachment; filename="{ascii_fallback}"; filename*=UTF-8\'\'{encoded_name}'
|
|
|
|
headers = {
|
|
"Content-Disposition": content_disposition,
|
|
"Content-Type": "application/octet-stream",
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Cache-Control": "no-store",
|
|
}
|
|
return OwnedFileResponse(
|
|
store=store, artifact_id=artifact_id,
|
|
path=str(file_path),
|
|
headers=headers,
|
|
)
|
|
|
|
# Serve only the production frontend asset directories at the root origin.
|
|
# index.html references css/, js/ and vendor/ relatively; dev/, tests/ and
|
|
# documentation inside the frontend tree are never exposed.
|
|
if app_settings.frontend_dist_dir and app_settings.frontend_dist_dir.exists():
|
|
for asset_dir in FRONTEND_ASSET_DIRS:
|
|
directory = app_settings.frontend_dist_dir / asset_dir
|
|
if directory.is_dir():
|
|
app.mount(f"/{asset_dir}", StaticFiles(directory=str(directory)), name=f"frontend-{asset_dir}")
|
|
|
|
return app
|
|
|
|
|
|
_default_app: Optional[FastAPI] = None
|
|
|
|
|
|
def get_app() -> FastAPI:
|
|
"""Lazy default application instance avoiding module-import side-effects."""
|
|
global _default_app
|
|
if _default_app is None:
|
|
_default_app = create_app()
|
|
return _default_app
|
|
|
|
|
|
async def app(scope: Any, receive: Any, send: Any) -> None:
|
|
"""ASGI 3.0 entrypoint for uvicorn backend.app:app."""
|
|
application = get_app()
|
|
await application(scope, receive, send)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
app_settings = Settings.from_env()
|
|
uvicorn.run("backend.app:app", host=app_settings.bind_host, port=app_settings.bind_port)
|