bot-bottle/tests/unit/test_egress_multitenant.py

"""Unit: fail-closed per-client egress resolution — config + context (PRD 0070)."""

from __future__ import annotations

import unittest

from bot_bottle.egress_addon_core import (
    DENY_RESOLVER_ERROR,
    DENY_UNATTRIBUTED,
    DENY_UNPARSEABLE,
    decide,
    resolve_client_config,
    resolve_client_context,
)
from bot_bottle.policy_resolver import PolicyResolveError


class _FakeResolver:
    def __init__(self, result: str | None = None, raises: bool = False) -> None:
        self._result = result
        self._raises = raises
        self.calls: list[tuple[str, str]] = []

    def resolve(self, source_ip: str, identity_token: str = "") -> str | None:
        self.calls.append((source_ip, identity_token))
        if self._raises:
            raise PolicyResolveError("orchestrator down")
        return self._result


class TestResolveClientConfig(unittest.TestCase):
    def test_valid_policy_is_parsed(self) -> None:
        cfg = resolve_client_config(
            _FakeResolver(result="routes:\n  - host: example.com\n"), "10.243.0.1"
        )
        self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))

    def test_unattributed_none_denies_all(self) -> None:
        cfg = resolve_client_config(_FakeResolver(result=None), "10.243.0.9")
        self.assertEqual((), cfg.routes)  # no routes → default-deny

    def test_empty_policy_denies_all(self) -> None:
        self.assertEqual((), resolve_client_config(_FakeResolver(result=""), "10.243.0.1").routes)

    def test_resolver_error_denies_all(self) -> None:
        # Orchestrator unreachable/errored must never widen egress.
        self.assertEqual((), resolve_client_config(_FakeResolver(raises=True), "10.243.0.1").routes)

    def test_unparseable_policy_denies_all(self) -> None:
        cfg = resolve_client_config(_FakeResolver(result="routes: notalist\n"), "10.243.0.1")
        self.assertEqual((), cfg.routes)

    def test_forwards_source_ip_and_token(self) -> None:
        r = _FakeResolver(result=None)
        resolve_client_config(r, "10.243.0.1", "tok")
        self.assertEqual(("10.243.0.1", "tok"), r.calls[0])


class _FakeContextResolver:
    def __init__(
        self, policy: str | None = None, bottle_id: str | None = None,
        raises: bool = False, tokens: dict[str, str] | None = None,
    ) -> None:
        self._policy = policy
        self._bottle_id = bottle_id
        self._raises = raises
        self._tokens = tokens or {}

    def resolve_policy_and_bottle_id(
        self, source_ip: str, identity_token: str = "",
    ) -> tuple[str | None, str | None, dict[str, str]]:
        if self._raises:
            raise PolicyResolveError("orchestrator down")
        return self._policy, self._bottle_id, self._tokens


class TestResolveClientContext(unittest.TestCase):
    def test_returns_config_bottle_id_and_tokens(self) -> None:
        cfg, slug, tokens = resolve_client_context(
            _FakeContextResolver(
                policy="routes:\n  - host: example.com\n", bottle_id="b1",
                tokens={"EGRESS_TOKEN_0": "sekret"},
            ),
            "10.243.0.1",
        )
        self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
        self.assertEqual("b1", slug)
        self.assertEqual({"EGRESS_TOKEN_0": "sekret"}, tokens)  # for auth injection

    def test_unattributed_denies_and_empty_slug(self) -> None:
        cfg, slug, tokens = resolve_client_context(
            _FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
        )
        self.assertEqual((), cfg.routes)
        self.assertEqual("", slug)  # no bottle → supervise unavailable
        self.assertEqual({}, tokens)

    def test_resolver_error_denies_empty_slug_no_tokens(self) -> None:
        cfg, slug, tokens = resolve_client_context(
            _FakeContextResolver(raises=True), "10.243.0.1",
        )
        self.assertEqual((), cfg.routes)
        self.assertEqual("", slug)
        self.assertEqual({}, tokens)

    def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
        # A bad policy denies egress, but the bottle is still attributed (its
        # supervise queue is keyed by the id, independent of route parsing).
        cfg, slug, _tokens = resolve_client_context(
            _FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
        )
        self.assertEqual((), cfg.routes)
        self.assertEqual("b2", slug)


if __name__ == "__main__":
    unittest.main()


class TestDenyReasonNamesTheRealFault(unittest.TestCase):
    """A deny-all must not masquerade as a missing allowlist entry.

    Regression: an unregistered bottle resolves no policy, so *every* host is
    denied — but the block message said `host X is not in the allowlist`,
    which reads as a config problem and sends the operator hunting for a route
    that was never missing. The structural reason wins over that wording.
    """

    def _reason(self, resolver: object, host: str = "chatgpt.com") -> str:
        cfg = resolve_client_config(resolver, "10.243.0.1")  # type: ignore[arg-type]
        return decide(cfg.routes, host, "/v1/x", {}, deny_reason=cfg.deny_reason).reason

    def test_unattributed_says_unregistered_not_allowlist(self) -> None:
        reason = self._reason(_FakeResolver(result=None))
        self.assertEqual(DENY_UNATTRIBUTED, reason)
        # The misleading claim is the one that must be gone: the host was
        # never "not in the allowlist" — there was no allowlist at all.
        self.assertNotIn("is not in the bottle's egress.routes allowlist", reason)
        self.assertIn("not registered", reason)

    def test_resolver_error_says_orchestrator_unreachable(self) -> None:
        self.assertEqual(DENY_RESOLVER_ERROR, self._reason(_FakeResolver(raises=True)))

    def test_unparseable_policy_says_so(self) -> None:
        self.assertEqual(
            DENY_UNPARSEABLE, self._reason(_FakeResolver(result="routes: notalist\n")))

    def test_a_real_allowlist_miss_keeps_the_allowlist_wording(self) -> None:
        """The message only changes for structural deny-alls — a loaded policy
        that genuinely lacks the host still points at the allowlist."""
        reason = self._reason(_FakeResolver(result='routes:\n  - host: "api.example.com"\n'))
        self.assertIn("is not in the bottle's egress.routes allowlist", reason)
        self.assertIn("chatgpt.com", reason)

    def test_allowed_host_is_still_forwarded(self) -> None:
        cfg = resolve_client_config(
            _FakeResolver(result='routes:\n  - host: "api.example.com"\n'), "10.243.0.1")
        decision = decide(
            cfg.routes, "api.example.com", "/v1/x", {}, deny_reason=cfg.deny_reason)
        self.assertEqual("forward", decision.action)

    def test_a_parsed_policy_carries_no_deny_reason(self) -> None:
        cfg = resolve_client_config(
            _FakeResolver(result='routes:\n  - host: "api.example.com"\n'), "10.243.0.1")
        self.assertEqual("", cfg.deny_reason)