stackpilot/backend/tests/test_route_authorization.py

"""The authorization matrix — StackPilot's most load-bearing test.

Authorization lives in the routers: each one picks ``require_admin`` or
``get_current_user`` per route, and nothing checks that the choice was right.
That is how 0.43.0 shipped a ``user`` role that could download the auth
database, every stack's ``.env`` and every ``.secrets/*`` file — a wrong default
on four routes, invisible in review.

So the policy is written down here instead of being implied by 171 individual
decisions:

    every route requires admin, unless it is listed in USER_READABLE or PUBLIC.

Adding a route that the read-only role can reach means adding it to the list,
which is the review moment this test exists to force. The check is static — it
reads FastAPI's dependency graph rather than calling the routes — because
calling all 171 would mean constructing valid bodies for each and would happily
fire ``POST /stacks/{id}/down`` at whatever Docker is around.

The dynamic tests at the bottom are the direct regression net for the specific
leaks that were found: they use a real non-admin token and assert 403.
"""
from __future__ import annotations

import pytest

# --------------------------------------------------------------------------- #
# The policy
# --------------------------------------------------------------------------- #

#: Reachable without any token. Everything needed to log in, plus the health
#: probe the container's HEALTHCHECK hits.
PUBLIC = {
    "GET /api/health",
    "GET /api/auth/needs-setup",
    "POST /api/auth/setup",
    "POST /api/auth/login",
    "POST /api/auth/refresh",
    # Only drops the refresh cookie. Requiring a valid token would mean you
    # cannot sign out once the session has already gone stale.
    "POST /api/auth/logout",
    # A Git forge has no StackPilot credentials to present, so this one cannot
    # be behind a bearer token. It is authorized instead by an HMAC over the
    # request body against a per-stack secret, and answers 404 — not 403 — to
    # anything unsigned, so it cannot be used to discover which stacks exist or
    # which are connected to a repository. See routers/git.py.
    "POST /api/git/webhook/{stack_id}",
}

#: Reachable by the read-only ``user`` role. Everything here has been checked
#: for whether it can return a credential:
#:
#: * ``GET /api/settings`` returns the interval and counts — webhook URLs (which
#:   carry tokens) come from the admin-only ``/api/settings/webhooks``.
#: * ``GET /api/stacks/{id}`` blanks out ``env`` for non-admins.
#: * ``GET /api/templates`` is metadata only; the detail route, which returns a
#:   template's env, is admin-only.
#: * The ``/api/editor/*`` routes are pure YAML transformations — they take YAML
#:   in and hand YAML back, touching nothing on disk.
USER_READABLE = {
    "GET /api/auth/me",
    # Mutating, but only ever on the caller's own account: it bumps their own
    # token_version to sign every one of their devices out.
    "POST /api/auth/logout-everywhere",
    "GET /api/dashboard/fleet",
    # Stacks: status, logs and the compose file. Not the .env, not the export.
    "GET /api/stacks",
    "GET /api/stacks/stats",
    "GET /api/stacks/updates",
    # The app-logo catalog and its images: public artwork from a public icon
    # set, keyed by app name. Nothing here is derived from this install.
    "GET /api/stacks/icons/search",
    "GET /api/stacks/icons/logo/{slug}",
    "GET /api/stacks/{stack_id}",
    "GET /api/stacks/{stack_id}/auto-update",
    # The stack's uploaded icon — an image the user already sees next to the
    # stack's name in the list. Uploading and resetting it stay admin-only.
    "GET /api/stacks/{stack_id}/icon",
    "GET /api/stacks/{stack_id}/logs",
    "GET /api/stacks/{stack_id}/services/{service}/logs",
    "POST /api/stacks/convert",
    # Editor helpers: stateless YAML rewriting.
    "POST /api/editor/validate",
    "POST /api/editor/services",
    "POST /api/editor/add-volume",
    "POST /api/editor/add-device",
    "POST /api/editor/remove-device",
    "POST /api/editor/set-gpu",
    "POST /api/editor/set-privileged",
    "POST /api/editor/set-resources",
    # Read-only inventory.
    "GET /api/containers/{container_id}",
    "GET /api/images",
    "GET /api/images/updates",
    # Vulnerability scan results: CVE ids and package versions for images this
    # host runs. Not a credential, and the read-only role can already see every
    # image tag and compose file, so the package list adds nothing it could not
    # work out. Reading is also the point — a monitoring script wants a
    # read-only token. Running a scan stays admin-only, because it spends real
    # CPU and spawns a container.
    "GET /api/images/scans",
    "GET /api/images/scan",
    "GET /api/images/scan-status",
    "GET /api/networks",
    "GET /api/networks/{network_id}",
    "GET /api/networks/{network_id}/containers",
    "GET /api/volumes",
    "GET /api/volumes/sizes",
    "GET /api/volumes/orphaned",
    "POST /api/volumes/generate-yaml",
    "POST /api/ports/conflicts",
    "GET /api/settings",
    "GET /api/system/info",
    "GET /api/system/gpus",
    "GET /api/system/devices",
    "GET /api/system/update",
    "GET /api/templates",
}


# --------------------------------------------------------------------------- #
# Reading the declared authorization off the routes
# --------------------------------------------------------------------------- #


def _declared_auth(route) -> str:
    """What a route actually requires, per its dependency graph."""
    from auth import get_current_user, require_admin

    found = set()

    def walk(dependant):
        for sub in dependant.dependencies:
            if sub.call is require_admin:
                found.add("admin")
            elif sub.call is get_current_user:
                found.add("user")
            walk(sub)

    walk(route.dependant)
    if "admin" in found:
        return "admin"
    if "user" in found:
        return "user"
    return "public"


def _expected_auth(key: str) -> str:
    if key in PUBLIC:
        return "public"
    if key in USER_READABLE:
        return "user"
    return "admin"


def _all_routes(app):
    from fastapi.routing import APIRoute

    out = []
    for route in app.routes:
        if not isinstance(route, APIRoute):
            continue
        for method in sorted(route.methods - {"HEAD", "OPTIONS"}):
            out.append((f"{method} {route.path}", route))
    return sorted(out, key=lambda r: r[0])


@pytest.fixture(scope="session")
def routes(app):
    return _all_routes(app)


# --------------------------------------------------------------------------- #
# The matrix
# --------------------------------------------------------------------------- #


def test_every_route_matches_the_declared_policy(routes):
    """Each route requires exactly what PUBLIC / USER_READABLE say it should.

    A new route that nobody classified defaults to "admin" — which is the safe
    direction. What this catches is the dangerous one: a route written with
    ``get_current_user`` that was never weighed against "can this return a
    credential".
    """
    wrong = []
    for key, route in routes:
        actual, expected = _declared_auth(route), _expected_auth(key)
        if actual != expected:
            wrong.append(f"  {key}\n      declared={actual}  expected={expected}")
    assert not wrong, (
        "Route authorization does not match the policy in this file.\n\n"
        + "\n".join(wrong)
        + "\n\nIf the route is genuinely safe for the read-only role, add it to "
        "USER_READABLE with a note on why it cannot return a credential. "
        "Otherwise give it require_admin."
    )


def test_no_unlisted_route_is_reachable_without_a_token(routes):
    """Only the login/health surface may skip authentication entirely."""
    unauthenticated = {key for key, route in routes if _declared_auth(route) == "public"}
    assert unauthenticated == PUBLIC


def test_policy_lists_have_no_stale_entries(routes):
    """Keep the lists honest when routes get renamed or removed."""
    known = {key for key, _ in routes}
    assert not (PUBLIC - known), f"PUBLIC lists routes that no longer exist: {PUBLIC - known}"
    assert not (USER_READABLE - known), (
        f"USER_READABLE lists routes that no longer exist: {USER_READABLE - known}"
    )


def test_everything_touching_the_filesystem_requires_admin(routes):
    """The file browser, host-path picker and stack export, as one rule.

    These reach whatever the backend container can see — which includes every
    ``.env`` and ``.secrets/*``. Reads are no less sensitive than writes here,
    which is the mistake this test exists to prevent a repeat of.
    """
    offenders = [
        key
        for key, route in routes
        if ("/files" in key or "/host/paths" in key or key.endswith("/export"))
        and _declared_auth(route) != "admin"
    ]
    assert not offenders, f"Filesystem routes must be admin-only: {offenders}"


# --------------------------------------------------------------------------- #
# Regression tests for the leaks that were actually found (F1, F3)
# --------------------------------------------------------------------------- #

#: Routes a ``user`` token could reach before 0.44.0, each of which handed out
#: credentials. A 403 here is the whole point.
LEAKED_BEFORE_0_44 = [
    ("GET", "/api/files/list?path=/opt"),
    ("GET", "/api/files/read?path=/opt/stacks/x/.env"),
    ("GET", "/api/files/download?path=/opt/stacks/x/.env"),
    ("GET", "/api/host/paths?path=/opt"),
    ("GET", "/api/audit"),
    ("GET", "/api/stacks/anything/export"),
    ("GET", "/api/templates/jellyfin"),
]


@pytest.mark.parametrize("method,url", LEAKED_BEFORE_0_44)
def test_read_only_role_is_refused(as_user, method, url):
    assert as_user.request(method, url).status_code == 403, (
        f"{method} {url} is reachable by the read-only role again"
    )


@pytest.mark.parametrize("method,url", LEAKED_BEFORE_0_44)
def test_admin_is_not_refused(as_admin, method, url):
    """The same routes must still work for admins.

    Anything but 403 passes: without a Docker daemon or the referenced paths
    these legitimately answer 400/404/502, and this test is about the
    authorization layer, not the handler.
    """
    assert as_admin.request(method, url).status_code != 403


def test_missing_token_is_401_not_403(client):
    assert client.get("/api/files/read?path=/etc/hostname").status_code == 401


# --------------------------------------------------------------------------- #
# The .env is withheld from the read-only role (F1, via the stacks API)
# --------------------------------------------------------------------------- #


@pytest.fixture
def stack_with_env(app):
    """A real stack folder plus its DB row, so GET /api/stacks/{id} resolves."""
    import os

    from sqlmodel import Session

    from database import engine
    from models.stack import Stack
    from services import compose_service

    stack_id = "authz-fixture"
    directory = compose_service.stack_dir(stack_id)
    os.makedirs(directory, exist_ok=True)
    compose_service.write_compose(stack_id, "services:\n  app:\n    image: alpine\n")
    compose_service.write_env(stack_id, "DB_PASSWORD=super-secret-value\n")
    with Session(engine) as session:
        if not session.get(Stack, stack_id):
            session.add(Stack(id=stack_id, name="authz fixture"))
            session.commit()
    return stack_id


def test_stack_detail_withholds_env_from_the_read_only_role(as_user, stack_with_env):
    body = as_user.get(f"/api/stacks/{stack_with_env}").json()
    assert body["env"] == ""
    assert "super-secret-value" not in str(body)
    # The compose file is still there — the read-only role keeps a useful view.
    assert "image: alpine" in body["yaml"]


def test_stack_detail_gives_admins_the_env(as_admin, stack_with_env):
    body = as_admin.get(f"/api/stacks/{stack_with_env}").json()
    assert "super-secret-value" in body["env"]