# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-lease-reclamation-and-jit-scoping-boundaries-must-an-instance-manage-ephemeral-credentials-to-prevent-static-secret-sprawl-and-privilege-escalation.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-lease-reclamation-and-jit-scoping-boundaries-must-an-instance-manage-ephemeral-credentials-to-prevent-static-secret-sprawl-and-privilege-escalation.warrant.json
# Markdown SHA-256:   6ebbe06af6eaeb2c6e03c50c643a4b72eb5c4e92b51fb38cf0d1e26af7fb83e4
# Code AST SHA-256:   099d70767077b6e9ba96da7287aff04a2399700bc18336a961b3666c2f5ca108
# Verified Classes:   CredentialLeaseExpiredError, EphemeralCredential, JITCredentialLeaseManager
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import time
import threading
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Callable, Iterator, TypeVar

T = TypeVar("T")


class CredentialLeaseExpiredError(PermissionError):
    '''Raised when an ephemeral credential lease has expired or reached its safety buffer.'''
    pass


@dataclass(frozen=True)
class EphemeralCredential:
    client_id: str
    secret_token: str
    expires_at_monotonic: float
    scopes: tuple[str, ...]

    def is_valid(self, safety_buffer_sec: float = 30.0) -> bool:
        '''Returns True only if the current monotonic time is safely before expiration.'''
        return (time.monotonic() + safety_buffer_sec) < self.expires_at_monotonic


class JITCredentialLeaseManager:
    '''
    Manages short-lived, JIT-scoped ephemeral credentials with:
    1. Minimum-privilege scope enforcement.
    2. Automatic expiration buffering.
    3. Thread-safe lease refreshing.
    4. Deterministic revocation and memory clearing hooks.
    '''

    def __init__(
        self,
        lease_issuer_fn: Callable[[tuple[str, ...]], tuple[str, str, float]],
        *,
        max_ttl_sec: float = 300.0,
        safety_buffer_sec: float = 30.0,
    ) -> None:
        self._issuer_fn = lease_issuer_fn
        self.max_ttl_sec = max_ttl_sec
        self.safety_buffer_sec = safety_buffer_sec
        self._lock = threading.Lock()
        self._current_credential: EphemeralCredential | None = None

    def _acquire_fresh_lease(self, required_scopes: tuple[str, ...]) -> EphemeralCredential:
        now = time.monotonic()
        # Mock issuer returns `(client_id, secret_token, ttl_sec)`
        client_id, token, ttl = self._issuer_fn(required_scopes)
        if ttl > self.max_ttl_sec:
            ttl = self.max_ttl_sec
        return EphemeralCredential(
            client_id=client_id,
            secret_token=token,
            expires_at_monotonic=now + ttl,
            scopes=required_scopes,
        )

    def get_valid_credential(self, required_scopes: tuple[str, ...]) -> EphemeralCredential:
        with self._lock:
            if self._current_credential is None or not self._current_credential.is_valid(self.safety_buffer_sec):
                self._current_credential = self._acquire_fresh_lease(required_scopes)
            return self._current_credential

    @contextmanager
    def scoped_session(self, required_scopes: tuple[str, ...]) -> Iterator[EphemeralCredential]:
        '''
        Context manager ensuring credentials are acquired fresh, scoped strictly,
        and cleared/revoked upon exit.
        '''
        cred = self.get_valid_credential(required_scopes)
        try:
            yield cred
        finally:
            # Deterministic lease cleanup / zeroing hook
            with self._lock:
                if self._current_credential == cred:
                    self._current_credential = None

    def verify_lease_invariants(self, requested_scopes: tuple[str, ...]) -> bool:
        if not requested_scopes or any(not s.strip() for s in requested_scopes):
            raise ValueError("JIT credential acquisition requires at least one non-empty scope")
        if self.safety_buffer_sec < 0.0:
            raise ValueError("safety_buffer_sec cannot be negative")
        assert len(requested_scopes) >= 1 and self.safety_buffer_sec >= 0.0
        return True