inkwell/src/thoughtsync/app.py

from __future__ import annotations

import asyncio
import logging
import mimetypes
import os
import secrets
from contextlib import suppress
from datetime import timedelta

from quart import Quart, jsonify, send_from_directory
from quart.sessions import SecureCookieSessionInterface

from . import __version__
from .auth import bp as auth_bp
from .client_dist import advertisement as client_advertisement, bp as client_bp
from .config import Config
from .db import session_scope
from .labels import bp as labels_bp
from .notes import bp as notes_bp
from .proxy import is_https
from .retention import run_sweeper
from .saved_filters import bp as saved_filters_bp
from .settings import get_public_config, get_setting, load_or_create_secret_key, refresh_live
from .settings_api import bp as settings_bp
from .sync import bp as sync_bp, protocol_advertisement

# Without this, `logger.info` from this package goes nowhere: hypercorn configures its
# own access/error loggers and leaves the root logger at WARNING, so the credential
# events in auth.py would be invisible in `docker compose logs` — which is exactly
# where they are meant to be read until an audit table exists (task 2939).
#
# `force=False` (the default) so a host that has already configured logging keeps its
# own setup; LOG_LEVEL lets an operator turn it up without a code change.
logging.basicConfig(
    level=os.environ.get("THOUGHTSYNC_LOG_LEVEL", "INFO").upper(),
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")

# `.webmanifest` isn't in every base image's mime map; register it so the PWA
# manifest is served as application/manifest+json instead of octet-stream.
mimetypes.add_type("application/manifest+json", ".webmanifest")


class _AutoSecureSessionInterface(SecureCookieSessionInterface):
    """Mark the session cookie `Secure` whenever the request arrived over HTTPS —
    directly, or via a TLS-terminating reverse proxy that sets X-Forwarded-Proto.

    Auto-detecting per request (rather than a fixed SESSION_COOKIE_SECURE flag)
    hardens the cookie on HTTPS deployments without breaking a plain-HTTP install on
    a trusted LAN, where a hard-forced Secure flag would stop the browser from ever
    sending the cookie back — i.e. silently break login. No configuration required.
    """

    def get_cookie_secure(self, app: Quart) -> bool:
        return is_https()


def create_app() -> Quart:
    # static_folder=None: the SPA catch-all below owns static serving.
    app = Quart(__name__, static_folder=None)
    # Ephemeral/env secret so the app (and DB-free unit tests) construct without a
    # database. before_serving swaps in the real, DB-persisted key before serving.
    app.secret_key = Config.secret_key_env() or secrets.token_urlsafe(48)
    app.config["APP_VERSION"] = os.environ.get("APP_VERSION", __version__)
    app.config["SESSION_COOKIE_HTTPONLY"] = True
    app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
    # Auto-mark the session cookie Secure on HTTPS requests (see the interface above).
    app.session_interface = _AutoSecureSessionInterface()
    app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=30)
    # Hard request-body ceiling (any-file attachments, import zips, sync push). The
    # per-file attachment limit is the DB-backed `max_attachment_mb` setting, enforced
    # in the upload handler; this must stay >= the largest value that allows.
    app.config["MAX_CONTENT_LENGTH"] = 64 * 1024 * 1024

    app.register_blueprint(auth_bp)
    app.register_blueprint(notes_bp)
    app.register_blueprint(labels_bp)
    app.register_blueprint(settings_bp)
    app.register_blueprint(sync_bp)
    app.register_blueprint(saved_filters_bp)
    app.register_blueprint(client_bp)

    @app.before_serving
    async def _bootstrap() -> None:
        # Load (or generate + persist) the real signing secret and the live session
        # lifetime from the DB, before any request is served.
        async with session_scope() as db:
            app.secret_key = await load_or_create_secret_key(db)
            try:
                days = int(await get_setting(db, "session_ttl_days"))
                app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=days)
            except (ValueError, TypeError, KeyError):
                pass
            # The security settings the throttle and the proxy trust read on hot
            # paths. Cached rather than queried per request; until this runs they
            # hold their registry defaults, which is the correct behaviour for a
            # server that has not finished starting.
            await refresh_live(db)
        # Expire old trash in the background (retention.py). One task per process is
        # correct because the image serves with a single hypercorn worker (Dockerfile);
        # if that ever gains `--workers`, this needs a lock so N workers don't each
        # sweep. Duplicate sweeps would be harmless but wasteful — a purged row is
        # skipped by `purged_at IS NULL` — so this is about load, not correctness.
        app.config["TRASH_SWEEPER"] = asyncio.create_task(run_sweeper())

    @app.after_serving
    async def _shutdown() -> None:
        task = app.config.get("TRASH_SWEEPER")
        if task is not None:
            task.cancel()
            # Await the cancellation so shutdown doesn't race a sweep mid-transaction.
            with suppress(asyncio.CancelledError):
                await task

    @app.after_request
    async def _security_headers(response):
        """Headers a publicly-reachable instance should be sending.

        None of these change how the app behaves for a legitimate caller; they narrow
        what a browser will do if something else goes wrong.

        The CSP is the substantive one. `script-src 'self'` means that even if some
        future path did manage to reflect user text into the page, the browser would
        refuse to run it — the app has no inline scripts and no third-party scripts,
        so nothing legitimate is given up. The exceptions are honest ones:
          - `style-src 'unsafe-inline'` — Vue writes inline styles itself (`v-show`
            toggling display, TransitionGroup's FLIP setting transforms). Inline
            STYLE is not an execution primitive the way inline script is.
          - `img-src https: http:` — link previews render the remote og:image of
            whatever was linked, which is an arbitrary host by definition. Both
            schemes, because a LAN install is served over http and would otherwise
            lose every preview image; on an https instance the browser blocks the
            http ones as mixed content anyway, so naming it concedes nothing.
          - `blob:`/`data:` — attachment previews and the desktop's blob URI scheme.

        `frame-ancestors 'none'` replaces the older X-Frame-Options and is what stops
        the app being framed for clickjacking; `form-action 'self'` stops a form from
        being pointed at another origin.
        """
        response.headers.setdefault(
            "Content-Security-Policy",
            "default-src 'self'; "
            "base-uri 'self'; "
            "object-src 'none'; "
            "frame-ancestors 'none'; "
            "form-action 'self'; "
            "script-src 'self'; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: blob: https: http:; "
            "font-src 'self' data:; "
            "media-src 'self' blob:; "
            "connect-src 'self'",
        )
        # Content-type sniffing turns a file we said was text into whatever the bytes
        # look like. The attachment route already sets this; every other response
        # deserves it too.
        response.headers.setdefault("X-Content-Type-Options", "nosniff")
        # Note titles and label names end up in the URL of a search or a label lens,
        # and a full Referer would hand them to any site a link preview points at.
        response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
        # Nothing here uses a camera, a microphone or a location, so nothing embedded
        # in a page should be able to ask on its behalf.
        response.headers.setdefault(
            "Permissions-Policy", "camera=(), microphone=(), geolocation=(), interest-cohort=()"
        )
        # HSTS only where the request already arrived over TLS — the same detection
        # the session cookie uses. Sending it on a plain-HTTP LAN install would tell
        # the browser to refuse the only scheme that install serves.
        #
        # Scoped to this host: no `includeSubDomains` and no `preload`, both of which
        # commit domains this app does not own. A browser still remembers the policy
        # for up to a year after the header stops being sent, which is the point of
        # it — worth knowing before putting a hostname behind TLS temporarily.
        if is_https():
            response.headers.setdefault("Strict-Transport-Security", "max-age=31536000")
        return response

    @app.get("/api/health")
    async def health():
        return jsonify({"status": "ok", "version": app.config["APP_VERSION"]})

    @app.get("/api/config")
    async def public_config():
        # Public: the login/register screen reads site name + whether signups are open.
        async with session_scope() as db:
            data = await get_public_config(db)
        data["version"] = app.config["APP_VERSION"]
        # The sync-protocol handshake (M10.6). A native client reads this BEFORE
        # linking — while it still has no token and possibly no account — to decide
        # whether it can talk to this server, and which optional features to offer.
        data.update(protocol_advertisement())
        # Which Android client this server can hand out, if any. Absent rather than
        # null when it has none, so the web UI hides the download instead of
        # offering a button that 404s.
        data.update(client_advertisement())
        return jsonify(data)

    @app.get("/", defaults={"path": ""})
    @app.get("/<path:path>")
    async def spa(path: str):
        if path.startswith("api/"):
            return jsonify({"error": "not found"}), 404
        candidate = os.path.join(STATIC_DIR, path)
        if path and os.path.isfile(candidate):
            return await send_from_directory(STATIC_DIR, path)
        index = os.path.join(STATIC_DIR, "index.html")
        if os.path.isfile(index):
            return await send_from_directory(STATIC_DIR, "index.html")
        return jsonify({"error": "frontend not built"}), 404

    return app