secrets-engine/tests/test_service_auth.py

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",
        )