app/backend/tests/test_cloud.py

"""Remote access: off by default, and only ever as much as somebody granted.

The property worth pinning down is the one everything else rests on — a
portal's token is worth nothing here until somebody at this installation
enrolled it, and even then it grants exactly the rights of the local account
the portal identity holding it was mapped to. An identity nobody mapped gets
nothing, which is what makes deleting that local account a revocation.
"""

from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, Mock, patch

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi.testclient import TestClient
from sqlmodel import Session

from app.api.deps import decode_token, user_from_token
from app.cloud import config as cloud_config
from app.core.config import settings
from app.flow.panels import PanelDef, PanelsConfig, write_config
from app.models import User
from tests.utils.portal import ISSUER, portal_token


def test_portal_token_is_refused_when_not_enrolled(
    portal_key: rsa.RSAPrivateKey, tmp_path_factory: pytest.TempPathFactory
) -> None:
    """An installation nobody connected trusts no portal at all."""
    original = settings.CLOUD_CONFIG_FILE
    settings.CLOUD_CONFIG_FILE = tmp_path_factory.mktemp("empty") / "cloud.json"
    try:
        with pytest.raises(jwt.exceptions.InvalidTokenError):
            decode_token(portal_token(portal_key))
    finally:
        settings.CLOUD_CONFIG_FILE = original


def test_portal_token_resolves_by_portal_identity(
    enrolled: User, portal_key: rsa.RSAPrivateKey, db: Session
) -> None:
    """The token names a person on the portal; the mapping names them here."""
    claims = decode_token(portal_token(portal_key))
    # No local account of its own: the payload carries who they are on the
    # portal and nothing else, so an unmapped identity cannot fall back onto
    # whoever enrolled.
    assert "sub" not in claims
    assert claims["portal_sub"] == "portal-user-1"
    assert user_from_token(db, portal_token(portal_key)) == enrolled
    assert user_from_token(db, portal_token(portal_key, subject="nobody")) is None


def test_token_for_another_installation_is_refused(
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    portal_key: rsa.RSAPrivateKey,
) -> None:
    """The audience is this installation's id, so someone else's is worthless."""
    with pytest.raises(jwt.exceptions.InvalidTokenError):
        decode_token(portal_token(portal_key, audience=str(uuid.uuid4())))


def test_token_from_an_unpinned_key_is_refused(
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
) -> None:
    """A different portal, or a hijacked one, cannot sign for this installation."""
    impostor = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    with pytest.raises(jwt.exceptions.InvalidTokenError):
        decode_token(portal_token(impostor))


def test_non_proxy_scope_is_refused(
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    portal_key: rsa.RSAPrivateKey,
) -> None:
    with pytest.raises(jwt.exceptions.InvalidTokenError):
        decode_token(portal_token(portal_key, scope="session"))


def test_disconnecting_ends_remote_access(
    enrolled: User, portal_key: rsa.RSAPrivateKey
) -> None:
    """Deleting the config is the whole of the local revocation."""
    assert decode_token(portal_token(portal_key))["portal_sub"] == enrolled.portal_sub
    cloud_config.delete()
    with pytest.raises(jwt.exceptions.InvalidTokenError):
        decode_token(portal_token(portal_key))


def test_status_reports_not_enrolled(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    response = client.get(
        f"{settings.API_V1_STR}/cloud/status", headers=superuser_token_headers
    )
    assert response.status_code == 200
    assert response.json()["enrolled"] is False


def test_enrolling_needs_a_superuser(
    client: TestClient, normal_user_token_headers: dict[str, str]
) -> None:
    """Remote access is an installation-wide grant, not a personal setting."""
    response = client.post(
        f"{settings.API_V1_STR}/cloud/enroll",
        headers=normal_user_token_headers,
        json={"portal_url": ISSUER, "claim_code": "ABCD-EFGH"},
    )
    assert response.status_code == 403


def test_a_panel_scoped_portal_token_reaches_only_its_panel(
    client: TestClient,
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    portal_key: rsa.RSAPrivateKey,
    superuser_token_headers: dict[str, str],
) -> None:
    """A screen paired through the portal is bounded here, not there.

    The portal names the panel; everything about what that means is this
    installation's, which is the whole reason it may mint one at all.
    """
    write_config(
        PanelsConfig(
            panels=[
                PanelDef(id="hallway", dashboards=["kitchen"]),
                PanelDef(id="workshop", dashboards=["bench"]),
            ]
        )
    )
    for name in ("kitchen", "bench"):
        created = client.post(
            f"{settings.API_V1_STR}/dashboards/{name}", headers=superuser_token_headers
        )
        assert created.status_code in (200, 201, 409), created.text

    token = portal_token(portal_key, subject="hallway", scope="panel")
    headers = {"Authorization": f"Bearer {token}"}

    assert (
        client.get(f"{settings.API_V1_STR}/panels/hallway", headers=headers).status_code
        == 200
    )
    assert (
        client.get(
            f"{settings.API_V1_STR}/dashboards/kitchen", headers=headers
        ).status_code
        == 200
    )
    # Another panel's dashboard, the panel list, and a draft are all refused.
    assert (
        client.get(
            f"{settings.API_V1_STR}/dashboards/bench", headers=headers
        ).status_code
        == 403
    )
    assert (
        client.get(f"{settings.API_V1_STR}/panels/", headers=headers).status_code == 403
    )
    assert (
        client.get(f"{settings.API_V1_STR}/flows/", headers=headers).status_code == 403
    )

    # Deleting the panel is how the screen is retired, whoever minted its token.
    write_config(PanelsConfig(panels=[PanelDef(id="workshop", dashboards=["bench"])]))
    assert (
        client.get(f"{settings.API_V1_STR}/panels/hallway", headers=headers).status_code
        == 401
    )


def test_a_panel_token_naming_no_panel_is_refused(
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    portal_key: rsa.RSAPrivateKey,
) -> None:
    """It would otherwise fall through to the account it borrows."""
    with pytest.raises(jwt.exceptions.InvalidTokenError):
        decode_token(portal_token(portal_key, subject="", scope="panel"))


def test_adding_a_remote_user_maps_and_revokes(
    client: TestClient,
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    portal_key: rsa.RSAPrivateKey,
    superuser_token_headers: dict[str, str],
    db: Session,
) -> None:
    """Admitting somebody makes an ordinary local user; deleting it ends them."""
    portal_reply = Mock(
        status_code=200,
        json=Mock(
            return_value={"user_id": "portal-user-9", "email": "remote@example.com"}
        ),
    )
    with patch("app.api.routes.cloud.httpx.post", return_value=portal_reply) as post:
        added = client.post(
            f"{settings.API_V1_STR}/cloud/users",
            headers=superuser_token_headers,
            json={"code": "ABCD-EFGH"},
        )
    assert added.status_code == 200, added.text
    assert post.call_args.kwargs["headers"]["Authorization"].startswith("Bearer ")
    body = added.json()
    assert body["email"] == "remote@example.com"
    assert body["portal_sub"] == "portal-user-9"
    # Never a superuser: a remote user must not be able to admit anyone else.
    assert body["is_superuser"] is False

    theirs = portal_token(portal_key, subject="portal-user-9")
    resolved = user_from_token(db, theirs)
    assert resolved is not None and resolved.email == "remote@example.com"

    with patch("app.api.routes.cloud.httpx.post", return_value=portal_reply):
        again = client.post(
            f"{settings.API_V1_STR}/cloud/users",
            headers=superuser_token_headers,
            json={"code": "ABCD-EFGH"},
        )
    assert again.status_code == 409

    with patch(
        "app.api.routes.cloud.httpx.delete", return_value=Mock(status_code=200)
    ) as delete:
        removed = client.delete(
            f"{settings.API_V1_STR}/users/{body['id']}", headers=superuser_token_headers
        )
    assert removed.status_code == 200, removed.text
    assert delete.call_args.args[0].endswith("/installation-members/portal-user-9")
    db.expire_all()
    assert user_from_token(db, theirs) is None


def test_adding_a_remote_user_needs_a_superuser(
    client: TestClient,
    enrolled: User,  # noqa: ARG001 (fixture installs the enrolment)
    normal_user_token_headers: dict[str, str],
) -> None:
    """Widening who can reach this installation stays a superuser's decision."""
    response = client.post(
        f"{settings.API_V1_STR}/cloud/users",
        headers=normal_user_token_headers,
        json={"code": "ABCD-EFGH"},
    )
    assert response.status_code == 403


def test_enrolling_against_a_portal_without_an_owner_is_refused(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    tmp_path_factory: pytest.TempPathFactory,
) -> None:
    """A portal too old to name the owner would leave nobody mapped here."""
    original = settings.CLOUD_CONFIG_FILE
    settings.CLOUD_CONFIG_FILE = tmp_path_factory.mktemp("old-portal") / "cloud.json"
    reply = Mock(
        status_code=200,
        json=Mock(
            return_value={
                "installation_id": str(uuid.uuid4()),
                "installation_token": "t",
                "ws_url": f"{ISSUER}/api/v1/tunnel/attach",
                "issuer": ISSUER,
                "jwks": {"keys": []},
            }
        ),
    )
    try:
        with patch("app.api.routes.cloud.httpx.AsyncClient") as client_cls:
            client_cls.return_value.__aenter__.return_value.post = AsyncMock(
                return_value=reply
            )
            response = client.post(
                f"{settings.API_V1_STR}/cloud/enroll",
                headers=superuser_token_headers,
                json={"portal_url": ISSUER, "claim_code": "ABCD-EFGH"},
            )
        assert response.status_code == 502
        # Nothing was written: an enrolment nobody can act as is not one to keep.
        assert not settings.CLOUD_CONFIG_FILE.exists()
    finally:
        settings.CLOUD_CONFIG_FILE = original