beer_data/src/flask/app_factory.py

"""Application factory for the beer_data web app.

The application is deliberately small: a handful of pages, one Mongo database,
no build step for the front end.  Everything server-side lives in blueprints
under ``src/flask/views/`` and the domain maths in ``src/flask/brewing.py``.
"""

from __future__ import annotations

import os
from pathlib import Path

from flask import (Flask, g, redirect, request, session, url_for)
from flask_wtf import CSRFProtect

__all__ = ["create_app"]


#: The Content-Security-Policy the app sends with every response.
#:
#: The app serves only its own vendored assets — CSS, JavaScript, fonts and
#: the logo are all in ``static/`` — so a strict ``'self'`` policy is
#: enforceable.  There is no ``'unsafe-inline'`` anywhere, which is why the
#: templates carry no inline scripts or style attributes.
CSP = (
    "default-src 'self'; "
    "script-src 'self'; "
    "style-src 'self'; "
    "img-src 'self' data:; "
    "font-src 'self'; "
    "connect-src 'self'; "
    "form-action 'self'; "
    "object-src 'none'; "
    "frame-ancestors 'none'; "
    "base-uri 'self'"
)


class ImproperlyConfigured(RuntimeError):
    """Raised when the app is missing configuration it cannot safely default.

    Werkzeug 3 dropped its own ``ImproperlyConfigured``, and Flask has never
    exported one, so the app defines its own.  It is a ``RuntimeError`` so it
    is not mistaken for an HTTP error the app should try to render a page for.
    """


def _require_secret_key() -> str:
    """Return the Flask secret key, refusing to start without a real one.

    A known default secret key means anyone can forge a session cookie and
    log in as the operator, so a missing or placeholder value must stop the
    app rather than silently fall back.  The one exception is the test
    suite, which supplies its own key through the config.
    """
    secret = os.getenv("SECRET_KEY", "").strip()
    if not secret:
        raise ImproperlyConfigured(
            "SECRET_KEY is not set. Export a random value, e.g. "
            "`export SECRET_KEY=$(python -c 'import secrets; print(secrets.token_hex(32))')`."
        )
    if secret.lower() in {"change-me-in-the-deployment", "dev-secret-change-me"}:
        raise ImproperlyConfigured(
            "SECRET_KEY is still the placeholder value. Set a real secret key."
        )
    return secret


def create_app(config: dict | None = None) -> Flask:
    """Build and configure the Flask application."""
    app = Flask(
        __name__,
        template_folder="templates",
        static_folder="static",
    )

    app.config.update(
        SECRET_KEY=_require_secret_key(),
        MONGO_HOST=os.getenv("MONGO_HOST", "127.0.0.1"),
        MONGO_USER=os.getenv("MONGO_USER", "root"),
        MONGO_PASS=os.getenv("MONGO_PASS", ""),
        SESSION_COOKIE_HTTPONLY=True,
        SESSION_COOKIE_SAMESITE="Lax",
        # Flask-WTF needs this for the login form to render and validate.
        WTF_CSRF_ENABLED=True,
    )
    if config:
        app.config.update(config)

    # The pages must not be cached by the browser.

    # A phone held on to a cached copy of the brewing page and showed
    # the old inline advice instead of the button and the pop-out, so
    # the change was not visible until the cache was cleared by hand.
    @app.after_request
    def _no_store(response):
        if response.mimetype == "text/html":
            response.headers["Cache-Control"] = "no-store, must-revalidate"
            response.headers["Pragma"] = "no-cache"
        return response

    csrf = CSRFProtect(app)

    # ------------------------------------------------------------------ #
    # Request-scoped repository
    # ------------------------------------------------------------------ #
    from mongo.repository import BeerRepository

    def get_repo() -> BeerRepository:
        """One Mongo client per request, closed when the request ends."""
        if "repo" not in g:
            # A client can be injected via config (used by the test suite);
            # otherwise build one from the environment.
            injected = app.config.get("MONGO_CLIENT")
            g.repo = BeerRepository(client=injected) if injected else BeerRepository()
        return g.repo

    app.extensions["beer_repo"] = get_repo

    @app.teardown_appcontext
    def _close_repo(_exc=None):
        repo = g.pop("repo", None)
        # An injected client (the test suite) is owned by the caller — do not
        # close it, or the next request in the same test gets a dead client.
        if repo is not None and app.config.get("MONGO_CLIENT") is None:
            repo.client.close()

    # ------------------------------------------------------------------ #
    # Auth helpers
    # ------------------------------------------------------------------ #
    def logged_in() -> bool:
        return bool(session.get("logged_in"))

    @app.context_processor
    def inject_globals():
        return {
            "logged_in": logged_in(),
            "current_path": request.path,
            "app_name": "Brew Log",
        }

    # ------------------------------------------------------------------ #
    # Error handlers
    # ------------------------------------------------------------------ #
    @app.after_request
    def _set_security_headers(response):
        """Add the standard security headers.

        The app serves its own vendored JavaScript and CSS and loads nothing
        from a CDN, so a strict Content-Security-Policy of 'self' is enough
        and is genuinely enforceable.
        """
        response.headers.setdefault("X-Content-Type-Options", "nosniff")
        response.headers.setdefault("X-Frame-Options", "DENY")
        response.headers.setdefault("Referrer-Policy", "same-origin")
        response.headers.setdefault("Content-Security-Policy", CSP)
        return response

    @app.errorhandler(404)
    def _not_found(_exc):
        return render_template("error.html", code=404,
                               message="That page does not exist."), 404

    @app.errorhandler(500)
    def _server_error(_exc):  # pragma: no cover - defensive
        return render_template("error.html", code=500,
                               message="Something broke on our side."), 500

    # ------------------------------------------------------------------ #
    # Blueprints
    # ------------------------------------------------------------------ #
    from views.analysis import bp as analysis_bp
    from views.auth import bp as auth_bp
    from views.brews import bp as brews_bp

    app.register_blueprint(auth_bp)
    app.register_blueprint(brews_bp)
    app.register_blueprint(analysis_bp)

    @app.route("/healthz")
    def healthz():
        """Liveness probe.  Never touches Mongo — it answers "is the process
        up", not "is the app working".  A database outage must not cause the
        kubelet to restart a perfectly healthy process."""
        return {"status": "ok"}, 200

    @app.route("/readyz")
    def readyz():
        """Readiness probe.  Checks that Mongo is actually reachable, so a
        pod with no database is taken out of the Service's endpoints instead
        of serving errors."""
        try:
            repo = get_repo()
            repo.client.admin.command("ping")
        except Exception as exc:  # noqa: BLE001 - any failure means not ready
            return {"status": "not ready", "error": str(exc)}, 503
        return {"status": "ready"}, 200

    @app.route("/")
    def index():
        if logged_in():
            return redirect(url_for("brews.index"))
        return redirect(url_for("auth.login"))

    return app


def render_template(*args, **kwargs):  # noqa: D401 - re-export for views
    """Small shim so view modules can import ``render_template`` from one place."""
    from flask import render_template as _rt

    return _rt(*args, **kwargs)