app/backend/tests/test_cloud.py

"""Remote access: off by default, and scoped to the account that enabled it.

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 account that
did.
"""

from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

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

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.models import User

INSTALLATION_ID = "6f1c2d3e-4a5b-4c6d-8e9f-0a1b2c3d4e5f"
ISSUER = "https://hub.example.test"


@pytest.fixture
def portal_key() -> rsa.RSAPrivateKey:
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)


def _jwks(key: rsa.RSAPrivateKey) -> dict[str, Any]:
    jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(key.public_key()))
    jwk.update({"use": "sig", "alg": "RS256", "kid": "test-portal"})
    return {"keys": [jwk]}


def _portal_token(
    key: rsa.RSAPrivateKey,
    *,
    subject: str = "portal-user-1",
    audience: str = INSTALLATION_ID,
    issuer: str = ISSUER,
    scope: str = "proxy",
) -> str:
    now = datetime.now(UTC)
    pem = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    return jwt.encode(
        {
            "sub": subject,
            "iss": issuer,
            "aud": audience,
            "iat": now,
            "exp": now + timedelta(hours=1),
            "scope": scope,
        },
        pem,
        algorithm="RS256",
        headers={"kid": "test-portal"},
    )


@pytest.fixture
def enrolled(
    tmp_path_factory: pytest.TempPathFactory,
    portal_key: rsa.RSAPrivateKey,
    db: Session,
) -> Any:
    """Enrol this installation with a fake portal, then undo it."""
    local_user = db.exec(
        select(User).where(User.email == settings.FIRST_SUPERUSER)
    ).one()
    original = settings.CLOUD_CONFIG_FILE
    settings.CLOUD_CONFIG_FILE = (
        tmp_path_factory.mktemp(f"cloud-{uuid.uuid4().hex[:6]}") / "cloud.json"
    )
    cloud_config.save(
        cloud_config.CloudConfig(
            portal_url=ISSUER,
            ws_url=f"{ISSUER}/api/v1/tunnel/attach",
            installation_id=INSTALLATION_ID,
            token="installation-token",
            issuer=ISSUER,
            jwks=_jwks(portal_key),
            local_user_id=str(local_user.id),
            enrolled_at=datetime.now(UTC).isoformat(),
            portal_account=settings.FIRST_SUPERUSER,
        )
    )
    yield local_user
    cloud_config.delete()
    settings.CLOUD_CONFIG_FILE = original


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_to_the_enrolling_user(
    enrolled: User, portal_key: rsa.RSAPrivateKey, db: Session
) -> None:
    claims = decode_token(_portal_token(portal_key))
    assert claims["sub"] == str(enrolled.id)
    # Who they are on the portal is carried for the record, not for rights.
    assert claims["portal_sub"] == "portal-user-1"
    assert user_from_token(db, _portal_token(portal_key)) == enrolled


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))["sub"] == str(enrolled.id)
    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