from pathlib import Path from fastapi import Depends, FastAPI, Request from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.middleware.sessions import SessionMiddleware from .crypto import SecretBox from .db import Database from .fileconfig import (MASTER_KEY_FILE, SESSION_SECRET_FILE, FileConfig, load_file_config, load_master_key, load_session_secret) from .agent_runs import configured_secrets from .health import build_checks, run_checks from .logsetup import setup_logging from .ratelimit import FailureLimiter from .runtime import Runtime from .settings import SettingsStore from .users import UserStore from .web import (account, agent_pages, alerts, auth, baseline_pages, matrix_section, mutes_pages, pages, prompts_pages, prompts_test_pages, retention_pages, routing_pages, selftest_pages, telegram_section) from .web.deps import LoginRequired, redirect, templates, verify_csrf STATIC_DIR = Path(__file__).parent / "web" / "static" SESSION_COOKIE = "checkbot_session" SESSION_MAX_AGE_SECONDS = 8 * 3600 SECURITY_HEADERS = { "Content-Security-Policy": ("default-src 'self'; base-uri 'self'; form-action 'self'; " "frame-ancestors 'none'"), "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "Referrer-Policy": "same-origin", } def create_app(config: FileConfig | None = None) -> FastAPI: config = config or load_file_config() box = SecretBox(load_master_key(config.config_dir / MASTER_KEY_FILE)) session_secret = load_session_secret(config.config_dir / SESSION_SECRET_FILE) db = Database(config.db_path) store = SettingsStore(db, box) setup_logging(config.log_level, lambda: configured_secrets(store)) runtime = Runtime(db, store) app = FastAPI(title="checkbot", dependencies=[Depends(verify_csrf)], lifespan=runtime.lifespan, docs_url=None, redoc_url=None, openapi_url=None) app.state.db = db app.state.store = store app.state.runtime = runtime app.state.users = UserStore(db, box) app.state.limiter = FailureLimiter() app.state.totp_pending = {} app.add_middleware( SessionMiddleware, secret_key=session_secret, session_cookie=SESSION_COOKIE, max_age=SESSION_MAX_AGE_SECONDS, same_site="strict", https_only=config.secure_cookies, ) @app.middleware("http") async def security_headers(request: Request, call_next): response = await call_next(request) for name, value in SECURITY_HEADERS.items(): response.headers.setdefault(name, value) return response @app.exception_handler(LoginRequired) async def login_required(request: Request, exc: LoginRequired): return redirect("/login") @app.exception_handler(StarletteHTTPException) async def http_error(request: Request, exc: StarletteHTTPException): context = {"user": None, "csrf": "", "flashes": [], "status": exc.status_code, "detail": exc.detail} return templates.TemplateResponse(request, "error.html", context, status_code=exc.status_code) @app.get("/health") async def health() -> JSONResponse: result = await run_checks(build_checks(app.state.store, runtime.status, runtime.selftest_repo)) return JSONResponse(result, status_code=200 if result["ok"] else 503) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") app.include_router(auth.router) app.include_router(account.router) app.include_router(telegram_section.router) app.include_router(matrix_section.router) app.include_router(baseline_pages.router) app.include_router(routing_pages.router) app.include_router(mutes_pages.router) app.include_router(alerts.router) app.include_router(agent_pages.router) app.include_router(prompts_pages.router) app.include_router(prompts_test_pages.router) app.include_router(retention_pages.router) app.include_router(selftest_pages.router) app.include_router(pages.router) return app