95 lines
3.9 KiB
Python
95 lines
3.9 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, 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(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
|