Harden production authorization and service auth

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0217e-8c4c-7383-be6b-f50a6e485306

tests/test_service_auth.py

@@ -0,0 +1,159 @@
+import base64
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from secrets_engine.errors import BackendError
+from secrets_engine.service_auth import (
+    KeyCapeServiceAuthConfig,
+    KeyCapeServiceAuthProvider,
+    preflight_service_jwt,
+)
+
+NOW = datetime(2026, 8, 23, 12, 0, tzinfo=timezone.utc)
+
+
+def _segment(value):
+    return base64.urlsafe_b64encode(
+        json.dumps(value, separators=(",", ":")).encode()
+    ).decode().rstrip("=")
+
+
+def _jwt(**overrides):
+    now = int(NOW.timestamp())
+    claims = {
+        "iss": "https://keycape.example.test",
+        "sub": "service:secrets-engine",
+        "aud": "secrets-engine-openbao",
+        "iat": now,
+        "exp": now + 900,
+        "principal_type": "service",
+        "tenant": "tenant:coulomb",
+        "roles": ["secrets-engine"],
+        "groups": [],
+        "scope": "openbao:login",
+        "assurance": {
+            "aal": "AAL1",
+            "method": "client_secret",
+            "mfa": False,
+            "source": "key-cape",
+        },
+    }
+    claims.update(overrides)
+    return f"{_segment({'alg': 'RS256', 'typ': 'JWT'})}.{_segment(claims)}.signature"
+
+
+def _config(secret_file=Path("/tmp/keycape-client-secret")):
+    return KeyCapeServiceAuthConfig(
+        token_url="https://keycape.example.test/token",
+        issuer="https://keycape.example.test",
+        client_secret_file=secret_file,
+    )
+
+
+def test_preflight_accepts_exact_service_contract_and_renewal_window():
+    token = _jwt()
+    result = preflight_service_jwt(token, _config(), now=NOW)
+    assert result.expires_at - result.issued_at == 900
+    assert not result.needs_renewal(NOW)
+    assert result.needs_renewal(datetime.fromtimestamp(result.expires_at - 180, timezone.utc))
+    assert token not in repr(result)
+
+
+@pytest.mark.parametrize(
+    ("override", "match"),
+    [
+        ({"iss": "https://attacker.test"}, "'iss'"),
+        ({"sub": "service:operator"}, "'sub'"),
+        ({"aud": "different"}, "'aud'"),
+        ({"tenant": "tenant:other"}, "'tenant'"),
+        ({"roles": ["admin"]}, "roles"),
+        ({"scope": "openbao:login admin"}, "scope"),
+        ({"exp": int(NOW.timestamp()) + 901}, "15-minute"),
+    ],
+)
+def test_preflight_rejects_claim_or_lifetime_drift(override, match):
+    with pytest.raises(BackendError, match=match):
+        preflight_service_jwt(_jwt(**override), _config(), now=NOW)
+
+
+class _Response:
+    status = 200
+
+    def __init__(self, payload):
+        self.payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self):
+        return json.dumps(self.payload).encode()
+
+
+def test_exchange_uses_basic_auth_and_never_places_secret_in_body(tmp_path):
+    secret_file = tmp_path / "client.secret"
+    secret_file.write_text("client-secret-value")
+    secret_file.chmod(0o600)
+    captured = {}
+
+    def transport(request, *, timeout):
+        captured["authorization"] = request.get_header("Authorization")
+        captured["body"] = request.data.decode()
+        captured["timeout"] = timeout
+        return _Response(
+            {
+                "access_token": _jwt(),
+                "token_type": "Bearer",
+                "expires_in": 900,
+            }
+        )
+
+    provider = KeyCapeServiceAuthProvider(_config(secret_file), transport=transport)
+    result = provider.exchange(now=NOW)
+    assert result.expires_at == int(NOW.timestamp()) + 900
+    assert captured["authorization"].startswith("Basic ")
+    assert "client-secret-value" not in captured["body"]
+    assert captured["body"] == "grant_type=client_credentials&scope=openbao%3Alogin"
+
+
+def test_exchange_rejects_refresh_or_id_tokens_without_exposing_secret(tmp_path):
+    secret = "do-not-leak-this-client-secret"
+    secret_file = tmp_path / "client.secret"
+    secret_file.write_text(secret)
+    secret_file.chmod(0o600)
+
+    def transport(_request, *, timeout):
+        return _Response(
+            {
+                "access_token": _jwt(),
+                "token_type": "Bearer",
+                "expires_in": 900,
+                "refresh_token": "also-secret",
+            }
+        )
+
+    provider = KeyCapeServiceAuthProvider(_config(secret_file), transport=transport)
+    with pytest.raises(BackendError, match="forbidden extra token") as error:
+        provider.exchange(now=NOW)
+    assert secret not in str(error.value)
+
+
+def test_provider_config_is_exact_and_has_no_bootstrap_fallback():
+    with pytest.raises(BackendError, match="HTTPS"):
+        KeyCapeServiceAuthConfig(
+            token_url="http://keycape.example.test/token",
+            issuer="https://keycape.example.test",
+            client_secret_file=Path("/tmp/unused"),
+        )
+    with pytest.raises(BackendError, match="accepted contract"):
+        KeyCapeServiceAuthConfig(
+            token_url="https://keycape.example.test/token",
+            issuer="https://keycape.example.test",
+            client_secret_file=Path("/tmp/unused"),
+            client_id="operator",
+        )