agentic-mobile-control/tests/test_cloud_auth.py

from __future__ import annotations

from dataclasses import dataclass

import cloud.auth as auth_module
import pytest
from cloud.auth import (
    BearerCredential,
    ChainedAuthProvider,
    ChainedEnrollmentAuthProvider,
    ConfiguredBearerAuthProvider,
    ConfiguredEnrollmentTokenProvider,
    EnrollmentCredential,
    HostIdentityMismatchError,
    HostPrincipalRequiredError,
    NullAuthProvider,
    RepositoryHostAuthProvider,
    SelfServiceEnrollmentAuthProvider,
    digest_token,
)


@dataclass
class _Request:
    headers: dict[str, str]


def test_configured_bearer_auth_returns_identity_and_scopes() -> None:
    provider = ConfiguredBearerAuthProvider(
        [
            BearerCredential(
                principal_id="integrator-a",
                token="secret-a",
                scopes=frozenset({"tasks:submit", "tasks:read"}),
            )
        ]
    )

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer secret-a"})
    )

    assert principal is not None
    assert principal.id == "integrator-a"
    assert principal.scopes == frozenset({"tasks:submit", "tasks:read"})


def test_missing_malformed_or_invalid_bearer_token_is_rejected() -> None:
    provider = ConfiguredBearerAuthProvider(
        [BearerCredential(principal_id="integrator", token="valid-token")]
    )

    assert provider.authenticate(_Request(headers={})) is None
    assert (
        provider.authenticate(_Request(headers={"authorization": "Basic value"}))
        is None
    )
    assert (
        provider.authenticate(_Request(headers={"authorization": "Bearer invalid"}))
        is None
    )


def test_bearer_verification_compares_every_configured_digest(monkeypatch) -> None:
    comparisons: list[tuple[bytes, bytes]] = []
    original_compare_digest = auth_module.compare_digest

    def recording_compare_digest(left: bytes, right: bytes) -> bool:
        comparisons.append((left, right))
        return original_compare_digest(left, right)

    monkeypatch.setattr(auth_module, "compare_digest", recording_compare_digest)
    provider = ConfiguredBearerAuthProvider(
        [
            BearerCredential(principal_id="first", token="match"),
            BearerCredential(principal_id="second", token="other"),
        ]
    )

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer match"})
    )

    assert principal is not None
    assert principal.id == "first"
    assert len(comparisons) == 2
    assert all(len(left) == len(right) == 32 for left, right in comparisons)


def test_credential_representations_do_not_expose_token() -> None:
    credential = BearerCredential(principal_id="integrator", token="top-secret")
    provider = ConfiguredBearerAuthProvider([credential])

    assert "top-secret" not in repr(credential)
    assert "top-secret" not in repr(provider.__dict__)


def test_null_auth_provider_is_explicitly_unrestricted() -> None:
    principal = NullAuthProvider().authenticate(_Request(headers={}))

    assert principal is not None
    assert principal.id == "anonymous"
    assert principal.scopes == frozenset({"*"})


def test_host_credential_authenticates_as_one_bound_host() -> None:
    provider = ConfiguredBearerAuthProvider(
        [
            BearerCredential(
                principal_id="host-agent-a",
                token="host-secret",
                scopes=frozenset({"host:agent"}),
                host_id="host-a",
            )
        ]
    )

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer host-secret"})
    )

    assert principal is not None
    assert principal.host_id == "host-a"
    principal.require_host("host-a")


def test_host_principal_rejects_cross_host_operation() -> None:
    provider = ConfiguredBearerAuthProvider(
        [
            BearerCredential(
                principal_id="host-agent-a",
                token="host-secret",
                host_id="host-a",
            )
        ]
    )
    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer host-secret"})
    )

    assert principal is not None
    with pytest.raises(HostIdentityMismatchError):
        principal.require_host("host-b")


def test_public_principal_cannot_act_as_host() -> None:
    provider = ConfiguredBearerAuthProvider(
        [BearerCredential(principal_id="integrator", token="public-secret")]
    )
    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer public-secret"})
    )

    assert principal is not None
    with pytest.raises(HostPrincipalRequiredError):
        principal.require_host("host-a")


def test_authentication_failure_does_not_log_bearer_secret(caplog) -> None:
    provider = ConfiguredBearerAuthProvider(
        [BearerCredential(principal_id="integrator", token="valid-secret")]
    )

    assert (
        provider.authenticate(
            _Request(headers={"authorization": "Bearer invalid-secret"})
        )
        is None
    )
    assert "invalid-secret" not in caplog.text
    assert "valid-secret" not in caplog.text


def test_enrollment_token_provider_returns_digest_without_exposing_secret() -> None:
    credential = EnrollmentCredential(
        principal_id="installer-a",
        token="one-time-enrollment-secret",
    )
    provider = ConfiguredEnrollmentTokenProvider([credential])

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer one-time-enrollment-secret"})
    )

    assert principal is not None
    assert principal.id == "installer-a"
    assert principal.token_digest == digest_token("one-time-enrollment-secret")
    assert "one-time-enrollment-secret" not in repr(credential)
    assert "one-time-enrollment-secret" not in repr(provider.__dict__)


def test_repository_host_auth_and_chain_preserve_host_scope() -> None:
    class Repository:
        def authenticate_enrolled_host(self, credential_digest: str) -> str | None:
            if credential_digest == digest_token("dynamic-host-secret"):
                return "host-managed"
            return None

    configured = ConfiguredBearerAuthProvider(
        [BearerCredential(principal_id="sdk", token="sdk-secret")]
    )
    provider = ChainedAuthProvider(
        [configured, RepositoryHostAuthProvider(Repository())]  # type: ignore[arg-type]
    )

    public_principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer sdk-secret"})
    )
    host_principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer dynamic-host-secret"})
    )

    assert public_principal is not None
    assert public_principal.id == "sdk"
    assert host_principal is not None
    assert host_principal.host_id == "host-managed"
    assert host_principal.scopes == frozenset()


def test_self_service_enabled_with_no_token_enrolls_via_self_service() -> None:
    provider = ChainedEnrollmentAuthProvider(
        [ConfiguredEnrollmentTokenProvider(()), SelfServiceEnrollmentAuthProvider()]
    )

    principal = provider.authenticate(_Request(headers={}))

    assert principal is not None
    assert principal.id == "self-service"
    assert principal.token_digest is None


def test_self_service_disabled_with_no_token_is_unauthenticated() -> None:
    provider = ChainedEnrollmentAuthProvider([ConfiguredEnrollmentTokenProvider(())])

    assert provider.authenticate(_Request(headers={})) is None


def test_self_service_enabled_with_valid_configured_token_uses_token_bound_path() -> (
    None
):
    credential = EnrollmentCredential(
        principal_id="installer-a",
        token="one-time-enrollment-secret",
    )
    provider = ChainedEnrollmentAuthProvider(
        [
            ConfiguredEnrollmentTokenProvider([credential]),
            SelfServiceEnrollmentAuthProvider(),
        ]
    )

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer one-time-enrollment-secret"})
    )

    assert principal is not None
    assert principal.id == "installer-a"
    assert principal.token_digest == digest_token("one-time-enrollment-secret")


def test_self_service_enabled_with_unknown_token_falls_through_to_self_service() -> (
    None
):
    credential = EnrollmentCredential(
        principal_id="installer-a",
        token="one-time-enrollment-secret",
    )
    provider = ChainedEnrollmentAuthProvider(
        [
            ConfiguredEnrollmentTokenProvider([credential]),
            SelfServiceEnrollmentAuthProvider(),
        ]
    )

    principal = provider.authenticate(
        _Request(headers={"authorization": "Bearer unknown-secret"})
    )

    assert principal is not None
    assert principal.id == "self-service"
    assert principal.token_digest is None