app

tit.server.app

FastAPI application factory for tit.server.

CSPMiddleware

CSPMiddleware(app: ASGIApp)

Add Content-Security-Policy to every HTML response.

Source code in tit/server/app.py

def __init__(self, app: ASGIApp) -> None:
    self.app = app

allowed_hosts

allowed_hosts(settings: ServerSettings) -> list[str]

127.0.0.1/localhost plus --allow-host / TIT_ALLOW_HOSTS entries.

Source code in tit/server/app.py

def allowed_hosts(settings: ServerSettings) -> list[str]:
    """``127.0.0.1``/``localhost`` plus ``--allow-host`` / ``TIT_ALLOW_HOSTS`` entries."""
    hosts = list(DEFAULT_ALLOWED_HOSTS)
    for host in settings.allow_hosts:
        if host and host not in hosts:
            hosts.append(host)
    return hosts

lifespan async

lifespan(app: FastAPI) -> AsyncIterator[None]

Reconcile stranded jobs and clean up notebook kernels on shutdown.

Source code in tit/server/app.py

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Reconcile stranded jobs and clean up notebook kernels on shutdown."""
    # Startup reconciliation (tit/jobs/manager.py::_reconcile_all) runs when the job manager is
    # first constructed, and blocks until it is done. Doing that here, rather than lazily on the
    # first jobs request, is what the maintainer's rule ("a restart should not automatically keep
    # running jobs") needs: any runner or sibling container left over from the previous server
    # life is stopped and its job failed *before* this server answers anything. Best-effort: a
    # server started without a project directory has no store to reconcile.
    with contextlib.suppress(Exception):
        from tit.jobs.bootstrap import get_manager

        await asyncio.to_thread(get_manager, app)
    try:
        yield
    finally:
        # Every kernel this process started is this process's to end. A
        # notebook kernel is a full SimNIBS Python interpreter; leaking one
        # per server restart is how a container ends up out of memory.
        with contextlib.suppress(Exception):
            from tit.server.kernels import get_kernel_registry

            get_kernel_registry().shutdown_all()

create_app

create_app(settings: ServerSettings) -> FastAPI

Build the application; sets the PathManager singleton from settings.

Source code in tit/server/app.py

def create_app(settings: ServerSettings) -> FastAPI:
    """Build the application; sets the PathManager singleton from *settings*."""
    if settings.project_dir:
        get_path_manager(settings.project_dir)

    app = FastAPI(
        title="TI-Toolbox job server (tit.server)",
        version=tit.__version__,
        description=f"server API {SERVER_API}",
        docs_url=None,
        redoc_url=None,
        openapi_url=None,
        lifespan=lifespan,
    )
    app.state.settings = settings
    app.state.started_at = time.monotonic()
    # Live cookie sessions (random ids, never the token). In-memory only: a
    # restart forgets them all and the UI falls back to the launcher.
    app.state.sessions = set()
    psutil.cpu_percent(interval=None)  # prime: the first sample is always 0.0

    # Middleware: last added = outermost. TrustedHost runs before everything.
    app.add_middleware(CSPMiddleware)
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts(settings))

    @app.get(
        "/auth/session",
        summary="Token → cookie exchange (called by the launcher once)",
        status_code=303,
        responses={
            303: {"description": "cookie set, redirect to /"},
            401: {"description": "bad token"},
        },
        response_class=RedirectResponse,
    )
    def auth_session(token: str, request: Request) -> RedirectResponse:
        if not _token_matches(token, request.app.state.settings.token):
            raise HTTPException(status_code=401, detail="Unauthorized")
        session_id = new_session_id()
        request.app.state.sessions.add(session_id)
        response = RedirectResponse(url="/", status_code=303)
        response.set_cookie(
            COOKIE_NAME,
            session_id,
            httponly=True,
            samesite="strict",
            secure=False,
            path="/",
        )
        return response

    @app.post(
        "/auth/logout",
        summary="End the cookie session (cookie or Bearer auth required)",
        status_code=204,
        dependencies=[Depends(require_auth)],
        responses={
            204: {"description": "cookie cleared, session forgotten"},
            401: {"description": "Unauthorized"},
        },
        response_class=Response,
    )
    def auth_logout(request: Request) -> Response:
        request.app.state.sessions.discard(request.cookies.get(COOKIE_NAME))
        response = Response(status_code=204)
        response.delete_cookie(COOKIE_NAME, path="/", httponly=True, samesite="strict")
        return response

    protected = [Depends(require_auth)]
    for (
        module
    ) in iter_route_modules():  # auto-discovered; see tit/server/routes/__init__.py
        name = module.__name__.rsplit(".", 1)[-1]
        router = getattr(module, "router", None)
        if router is not None:
            if name in OPEN_MODULES:
                app.include_router(router)
            else:
                app.include_router(router, dependencies=protected)
        ws_router = getattr(module, "ws_router", None)
        if ws_router is not None:
            app.include_router(ws_router)  # authorises inside the handshake
    app.include_router(ws.router)  # /ws/system, authorises inside the handshake
    app.include_router(static.router)  # catch-all, must be last

    app.openapi = lambda: _custom_openapi(app)  # type: ignore[method-assign]
    return app