windy-git/api/tests/test_ept_and_throttle.py

"""Behavioral tests for EPT verification and EI throttling.

These sign real ES256 tokens with a locally-generated key and verify against a
locally-served key set, so they exercise the ACTUAL crypto path with no network
dependency and no reliance on Eternitas being reachable.

This file exists because the suite it joins was ~86 "does the source contain
this string" assertions and zero that ran the auth decision — which is how a
live impersonation bypass passed every test on 2026-08-13.
"""

from __future__ import annotations

import time

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import ec

from api.app import ept as ept_mod
from api.app.auth import BAND_MULTIPLIER
from api.app.config import Settings
from api.app.ept import EptInvalid, verify_ept
from api.app.throttle import ACTION_BASE, limit_for

ISSUER = "eternitas.ai"
KID = "test-key-1"


@pytest.fixture
def signing(monkeypatch):
    """A real EC keypair; point the verifier's JWKS lookup at its public half."""
    key = ec.generate_private_key(ec.SECP256R1())

    class _FakeJWK:
        def __init__(self, k):
            self.key = k

    class _FakeClient:
        def __init__(self, *a, **kw):
            pass

        def get_signing_key_from_jwt(self, token):
            header = jwt.get_unverified_header(token)
            if header.get("kid") != KID:
                raise Exception(f"unknown kid {header.get('kid')!r}")
            return _FakeJWK(key.public_key())

    monkeypatch.setattr(ept_mod, "_jwks_client", None)
    monkeypatch.setattr(ept_mod, "PyJWKClient", _FakeClient)
    return key


def _sign(key, claims, alg="ES256", kid=KID):
    return jwt.encode(claims, key, algorithm=alg, headers={"kid": kid, "typ": "EPT"})


def _claims(**over):
    c = {
        "sub": "ET26-TEST-0001",
        "iss": ISSUER,
        "iat": int(time.time()) - 10,
        "exp": int(time.time()) + 3600,
    }
    c.update(over)
    return c


# ---- the property that was broken ----------------------------------------
def test_genuine_ept_is_accepted(signing):
    v = verify_ept(_sign(signing, _claims()), "https://api.eternitas.ai")
    assert v.passport == "ET26-TEST-0001"


def test_passport_comes_from_sub_not_a_passport_claim(signing):
    """Real EPTs put the passport in `sub`. The pre-fix code read `passport` /
    `sub_passport`, which no genuine EPT carries — so real agents were never
    recognised and only forged tokens ever worked."""
    tok = _sign(signing, _claims(sub="ET26-REAL-9999", passport="ET26-LIES-0000"))
    assert verify_ept(tok, "https://api.eternitas.ai").passport == "ET26-REAL-9999"


def test_alg_none_is_refused(signing):
    """The exact 2026-08-13 exploit."""
    import base64
    import json as _j

    def seg(d):
        return base64.urlsafe_b64encode(_j.dumps(d).encode()).rstrip(b"=").decode()

    forged = f"{seg({'alg':'none','typ':'EPT','kid':KID})}.{seg(_claims())}."
    with pytest.raises(EptInvalid):
        verify_ept(forged, "https://api.eternitas.ai")


def test_signature_from_a_different_key_is_refused(signing):
    attacker = ec.generate_private_key(ec.SECP256R1())
    with pytest.raises(EptInvalid):
        verify_ept(_sign(attacker, _claims()), "https://api.eternitas.ai")


def test_tampered_payload_is_refused(signing):
    tok = _sign(signing, _claims())
    h, _p, s = tok.split(".")
    import base64
    import json as _j

    evil = base64.urlsafe_b64encode(
        _j.dumps(_claims(sub="ET26-EVIL-0000")).encode()
    ).rstrip(b"=").decode()
    with pytest.raises(EptInvalid):
        verify_ept(f"{h}.{evil}.{s}", "https://api.eternitas.ai")


def test_expired_token_is_refused(signing):
    """Genuinely signed, genuinely expired — proves exp is enforced rather than
    the token merely failing to parse."""
    tok = _sign(signing, _claims(exp=int(time.time()) - 5, iat=int(time.time()) - 100))
    with pytest.raises(EptInvalid):
        verify_ept(tok, "https://api.eternitas.ai")


def test_wrong_issuer_is_refused(signing):
    """Correctly signed by a trusted key but claiming another issuer."""
    with pytest.raises(EptInvalid):
        verify_ept(_sign(signing, _claims(iss="evil.example.com")), "https://api.eternitas.ai")


def test_unknown_kid_is_refused(signing):
    with pytest.raises(EptInvalid):
        verify_ept(_sign(signing, _claims(), kid="attacker-key"), "https://api.eternitas.ai")


def test_missing_required_claims_are_refused(signing):
    for missing in ("sub", "exp"):
        c = _claims()
        c.pop(missing)
        with pytest.raises(EptInvalid):
            verify_ept(_sign(signing, c), "https://api.eternitas.ai")


def test_only_es256_is_ever_accepted():
    """Widening this list reopens algorithm confusion."""
    assert ept_mod.ALGORITHMS == ["ES256"]


# ---- the throttle that used to be dead code ------------------------------
def test_band_multiplier_is_actually_consumed():
    """BAND_MULTIPLIER was defined and read by nothing before this."""
    s = Settings()
    assert limit_for(s, "repo.create", "platinum") == s.rate_repo_creates_per_day * 10
    assert limit_for(s, "repo.create", "gold") == s.rate_repo_creates_per_day * 4
    assert limit_for(s, "repo.create", "standard") == s.rate_repo_creates_per_day
    assert limit_for(s, "repo.create", "watch") == s.rate_repo_creates_per_day // 2


def test_unknown_band_gets_standard_not_unlimited_and_not_zero():
    s = Settings()
    assert limit_for(s, "repo.create", "a-band-invented-tomorrow") == s.rate_repo_creates_per_day
    assert limit_for(s, "repo.create", None) == s.rate_repo_creates_per_day


def test_untrusted_band_is_read_only():
    assert BAND_MULTIPLIER["untrusted"] == 0


def test_every_throttled_action_has_a_configured_base():
    s = Settings()
    for action, field in ACTION_BASE.items():
        assert getattr(s, field) > 0, f"{action} has no positive base rate"


# ---- revocation enforced on the live trust path (not just the webhook) ----
def test_revoked_passport_is_refused_by_trust_decision():
    """A revoked passport returns HTTP 200, status=revoked, band=unproven,
    allowed=[] (verified live 2026-08-13). The decision must refuse it — the
    old code returned band 'unproven' and seated the agent."""
    from api.app.auth import PassportNotInGoodStanding, decide_trust

    revoked = {"status": "revoked", "band": "unproven", "allowed_actions": []}
    with pytest.raises(PassportNotInGoodStanding):
        decide_trust(revoked, "ET26-NJQT-QMR0")


def test_active_passport_is_accepted_by_trust_decision():
    from api.app.auth import decide_trust

    active = {"status": "active", "band": "gold", "allowed_actions": ["read", "send"]}
    band, actions = decide_trust(active, "ET26-1EF9-VJAN")
    assert band == "gold" and actions == ("read", "send")


def test_unknown_or_missing_status_fails_closed():
    from api.app.auth import PassportNotInGoodStanding, decide_trust

    for body in ({"band": "gold"}, {"status": "suspended"}, {"status": "frozen"},
                 {"status": ""}, {}):
        with pytest.raises(PassportNotInGoodStanding):
            decide_trust(body, "ET26-X")


@pytest.mark.asyncio
async def test_resolve_passport_raises_on_revoked_status_wiring(monkeypatch):
    """Proves the WIRING, not just the decision: resolve_passport must feed the
    real trust body through decide_trust and propagate the refusal. A validly
    minted EPT can outlive the passport by ~a year, so this is the path that
    stops a revoked-but-still-signed token."""

    from api.app import auth
    from api.app.auth import PassportNotInGoodStanding, resolve_passport
    from api.app.config import Settings

    class _Resp:
        status_code = 200

        def json(self):
            return {"status": "revoked", "band": "unproven", "allowed_actions": []}

    class _Client:
        def __init__(self, *a, **k):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def get(self, *a, **k):
            return _Resp()

    monkeypatch.setattr(auth.httpx, "AsyncClient", _Client)
    settings = Settings(eternitas_platform_api_key="x", eternitas_base_url="https://api.eternitas.ai")

    with pytest.raises(PassportNotInGoodStanding):
        await resolve_passport(settings, "ET26-NJQT-QMR0")


@pytest.mark.asyncio
async def test_resolve_passport_returns_band_on_active_wiring(monkeypatch):
    import httpx  # noqa: F401

    from api.app import auth
    from api.app.auth import resolve_passport
    from api.app.config import Settings

    class _Resp:
        status_code = 200

        def json(self):
            return {"status": "active", "band": "gold", "allowed_actions": ["read"]}

    class _Client:
        def __init__(self, *a, **k): pass
        async def __aenter__(self): return self
        async def __aexit__(self, *a): return False
        async def get(self, *a, **k): return _Resp()

    monkeypatch.setattr(auth.httpx, "AsyncClient", _Client)
    settings = Settings(eternitas_platform_api_key="x")
    band, actions = await resolve_passport(settings, "ET26-1EF9-VJAN")
    assert band == "gold" and actions == ("read",)


def test_routing_does_not_depend_on_signature_wellformedness():
    """An EPT-shaped token must route to the EPT verifier even when its
    signature is malformed. jwt.get_unverified_header() validates the whole
    token and rejects bad signature padding, which used to push such tokens to
    the human path — refused for the wrong reason, and readable as a human
    identity via `sub` whenever require_verified_jwt was off."""
    import base64
    import json as _j

    from api.app.ept import looks_like_ept

    def seg(d):
        return base64.urlsafe_b64encode(_j.dumps(d).encode()).rstrip(b"=").decode()

    for hdr in ({"alg": "none", "typ": "EPT"}, {"alg": "ES256", "typ": "EPT"},
                {"alg": "ES256"}):
        tok = f"{seg(hdr)}.{seg({'sub': 'ET26-X'})}.x"  # deliberately bad signature
        assert looks_like_ept(tok), f"{hdr} did not route to the EPT verifier"

    assert not looks_like_ept("not-a-token")
    assert not looks_like_ept("")


def test_only_actions_that_route_through_this_api_are_claimed_enforced():
    """git push never touches this API, so a push limit here would count zero
    forever and allow everything — a silent no-op wearing the costume of a
    control. Push limits must stay in NOT_ENFORCED_HERE until a Gitea-side hook
    reports pushes into agent_actions."""
    from api.app.throttle import ACTION_BASE, NOT_ENFORCED_HERE

    assert "push" not in ACTION_BASE
    assert "push.force" not in ACTION_BASE
    assert "push" in NOT_ENFORCED_HERE
    assert not set(ACTION_BASE) & set(NOT_ENFORCED_HERE)


@pytest.mark.asyncio
async def test_enforce_refuses_an_action_it_cannot_actually_limit():
    """Asking to throttle 'push' must raise, not silently allow."""
    from api.app.auth import ActorType, Caller
    from api.app.config import Settings
    from api.app.errors import RepairPointer
    from api.app.throttle import enforce

    caller = Caller(actor_type=ActorType.agent, passport="ET26-X", band="gold")
    with pytest.raises(RepairPointer) as exc:
        await enforce(None, Settings(), caller, "push")
    assert exc.value.code == "throttle_unknown_action"