zelos.credential_injector

 1"""
 2Credential Injector — Injects credentials into Task.constraints at dispatch time.
 3
 4v1.1.0: Called by ExecutionEngine.dispatch() before Agent execution.
 5Ensures credentials never appear in EventBus events (only in memory).
 6"""
 7
 8from .credential_store import CredentialStore
 9
10
11class CredentialNotFoundError(Exception):
12    """Credential not available for the requested agent."""
13    pass
14
15
16class CredentialExpiredError(Exception):
17    """Credential has expired."""
18    pass
19
20
21class CredentialInjector:
22    """Injects credentials into Task.constraints at dispatch time.
23
24    Design principles:
25    - Credentials are injected in-memory only, not persisted to EventBus
26    - Agent A never sees Agent B's credentials
27    - Expired credentials trigger Task failure for retry evaluation
28    """
29
30    def __init__(self, store: CredentialStore | None = None):
31        self._store = store
32
33    @property
34    def store(self) -> CredentialStore | None:
35        return self._store
36
37    def set_store(self, store: CredentialStore) -> None:
38        self._store = store
39
40    def inject(self, task, agent_id: str, required_credentials: list[str]) -> dict:
41        """Inject credentials into Task.constraints.
42
43        Args:
44            task: The Task being dispatched.
45            agent_id: The agent receiving the task.
46            required_credentials: List of credential names this agent needs.
47
48        Returns:
49            dict of credential_name -> Credential.
50
51        Raises:
52            CredentialNotFoundError: if a required credential is missing.
53            CredentialExpiredError: if a required credential has expired.
54        """
55        if not self._store or not required_credentials:
56            return {}
57
58        credentials = {}
59        for cred_name in required_credentials:
60            cred = self._store.get(cred_name, agent_id)
61            if cred is None:
62                raise CredentialNotFoundError(
63                    f"Credential '{cred_name}' not available for agent '{agent_id}'"
64                )
65            if not self._store.validate(cred_name, agent_id):
66                # Try refresh (for OAuth2 etc.)
67                refreshed = self._store.refresh(cred_name, agent_id)
68                if refreshed and not refreshed.is_expired():
69                    cred = refreshed
70                else:
71                    raise CredentialExpiredError(
72                        f"Credential '{cred_name}' has expired for agent '{agent_id}'"
73                    )
74            credentials[cred_name] = cred
75
76        # Inject into task.constraints (in-memory only)
77        task.constraints = task.constraints or {}
78        task.constraints["credential_refs"] = {
79            name: cred.to_dict() for name, cred in credentials.items()
80        }
81
82        return credentials