Files
checkbot/app/main.py
T
patrickandClaude Sonnet 5.5 347f55cf7c
test / pytest (3.11) (push) Canceled after 0s
test / pytest (3.13) (push) Canceled after 0s
feat: Phase 6c baseline, ack, host alarms, outage notice, send limit, privacy mode, UI link
Includes atomic migrations, per-rule reminder semantics, first-run helper script
and the note on the undecided UI chat idea.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-06 03:06:23 +01:00

96 lines
4.0 KiB
Python

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, 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))
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(pages.router)
return app