bootnode/api/tests/test_fleet_admin_gate.py

"""Adversarial tests for the server-side admin gate on the luxd fleet overview.

Proves the ONE cross-tenant infra route is deny-by-default on BOTH surfaces
(the standalone ``bootnode.fleet_app`` and the mounted ``api/fleets`` router):

    unauthenticated        -> 401
    invalid token          -> 401
    authenticated non-admin -> 403
    admin (role)           -> 200
    admin (bootstrap email allowlist, e.g. the sole initial superuser) -> 200

The gate is exercised for real end-to-end via HTTP; only the IAM token
verification (JWKS network call) and the luxd RPC collector are stubbed.
"""

import os

# Base/SQLite so the mounted router's db.session engine imports without Postgres.
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///:memory:")
os.environ.setdefault("KV_URL", "redis://localhost:6379/0")

import pytest
from fastapi import FastAPI, HTTPException, status
from fastapi.testclient import TestClient

import bootnode.fleet_app as fleet_app
from bootnode.core.chains.lux_fleet import LuxFleetOverview
from bootnode.core.iam import IAMUser, get_iam_client, require_admin

OVERVIEW_PATH = "/v1/fleets/lux/overview"


class _FakeIAM:
    """Stand-in IAMClient — maps a presented bearer token to a fixed decision,
    with no JWKS/network. ``verify_token`` is only reached when a credential is
    actually presented (missing credentials are rejected earlier, at 401)."""

    def __init__(self, user: IAMUser | None = None, raise_401: bool = False):
        self._user = user
        self._raise = raise_401

    async def verify_token(self, token: str) -> IAMUser:
        if self._raise:
            raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token")
        assert self._user is not None
        return self._user


def _user(email: str = "[email protected]", roles: list[str] | None = None) -> IAMUser:
    return IAMUser(id="u1", name="U", email=email, org="lux", roles=roles or [])


async def _empty_overview(selected=None) -> LuxFleetOverview:
    return LuxFleetOverview()


# --------------------------------------------------------------------------- #
# Fixtures: one TestClient per (app, iam-decision), with clean overrides.
# --------------------------------------------------------------------------- #

def _client_for(app: FastAPI, iam: _FakeIAM) -> TestClient:
    app.dependency_overrides[get_iam_client] = lambda: iam
    return TestClient(app)


@pytest.fixture
def standalone(monkeypatch):
    # Never touch a real luxd fleet from the gate tests.
    monkeypatch.setattr(fleet_app, "get_overview", _empty_overview)
    yield fleet_app.app
    fleet_app.app.dependency_overrides.clear()


def _import_fleets_router():
    """Import ``bootnode.api.fleets`` WITHOUT executing ``bootnode.api/__init__``.

    The package __init__ eagerly wires ~24 production routers (web3, bcrypt,
    mpc, …); the fleets submodule itself only needs deps/db/iam. We register a
    lightweight ``bootnode.api`` package object (correct ``__path__``) so the
    submodule loads directly — the same route object the app serves, minus the
    heavy tail.
    """
    import importlib
    import pathlib
    import sys
    import types

    import bootnode
    if "bootnode.api" not in sys.modules:
        pkg = types.ModuleType("bootnode.api")
        pkg.__path__ = [str(pathlib.Path(bootnode.__file__).parent / "api")]
        sys.modules["bootnode.api"] = pkg
    return importlib.import_module("bootnode.api.fleets")


@pytest.fixture
def mounted(monkeypatch):
    fleets = _import_fleets_router()
    monkeypatch.setattr(fleets, "get_lux_overview", _empty_overview)
    app = FastAPI()
    app.include_router(fleets.router, prefix="/v1/fleets")
    yield app
    app.dependency_overrides.clear()


# Run the identical adversarial matrix against BOTH surfaces.
_APPS = ["standalone", "mounted"]


def _app(name, standalone, mounted):
    return standalone if name == "standalone" else mounted


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

@pytest.mark.parametrize("app_name", _APPS)
def test_unauthenticated_is_401(app_name, standalone, mounted):
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(_user()))  # iam unused: no creds presented
    r = client.get(OVERVIEW_PATH)
    assert r.status_code == 401, r.text
    assert "networks" not in r.text  # no fleet data leaked in the body


@pytest.mark.parametrize("app_name", _APPS)
def test_invalid_token_is_401(app_name, standalone, mounted):
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(raise_401=True))
    r = client.get(OVERVIEW_PATH, headers={"Authorization": "Bearer garbage"})
    assert r.status_code == 401, r.text


@pytest.mark.parametrize("app_name", _APPS)
def test_authenticated_non_admin_is_403(app_name, standalone, mounted):
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(_user(roles=[])))  # a normal customer
    r = client.get(OVERVIEW_PATH, headers={"Authorization": "Bearer good"})
    assert r.status_code == 403, r.text
    assert "networks" not in r.text  # denied before any collection


@pytest.mark.parametrize("app_name", _APPS)
def test_admin_role_is_200(app_name, standalone, mounted):
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(_user(roles=["admin"])))
    r = client.get(OVERVIEW_PATH, headers={"Authorization": "Bearer good"})
    assert r.status_code == 200, r.text
    assert "networks" in r.json()


@pytest.mark.parametrize("app_name", _APPS)
def test_bootstrap_allowlist_admin_is_200(app_name, standalone, mounted, monkeypatch):
    # Sole initial superuser reaches the view before the IAM admin role is wired.
    monkeypatch.setenv("FLEET_ADMIN_EMAILS", "[email protected]")
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(_user(email="[email protected]", roles=[])))
    r = client.get(OVERVIEW_PATH, headers={"Authorization": "Bearer good"})
    assert r.status_code == 200, r.text


@pytest.mark.parametrize("app_name", _APPS)
def test_non_allowlisted_email_still_403(app_name, standalone, mounted, monkeypatch):
    monkeypatch.setenv("FLEET_ADMIN_EMAILS", "[email protected]")
    app = _app(app_name, standalone, mounted)
    client = _client_for(app, _FakeIAM(_user(email="[email protected]", roles=[])))
    r = client.get(OVERVIEW_PATH, headers={"Authorization": "Bearer good"})
    assert r.status_code == 403, r.text


def test_standalone_health_stays_open(standalone):
    # Liveness must remain unauthenticated (no gate regression on /health).
    client = TestClient(standalone)
    assert client.get("/health").status_code == 200


# --------------------------------------------------------------------------- #
# Gate primitive, tested directly (defense in depth, no HTTP layer).
# --------------------------------------------------------------------------- #

@pytest.mark.asyncio
async def test_require_admin_primitive_role():
    admin = _user(roles=["admin"])
    assert await require_admin(admin) is admin
    with pytest.raises(HTTPException) as exc:
        await require_admin(_user(roles=[]))
    assert exc.value.status_code == 403


@pytest.mark.asyncio
async def test_require_admin_primitive_allowlist(monkeypatch):
    monkeypatch.setenv("FLEET_ADMIN_EMAILS", "[email protected], [email protected]")
    ok = _user(email="[email protected]", roles=[])  # case-insensitive match
    assert await require_admin(ok) is ok