Tyler King
f3ffeb38ae
Extracts agent identity registration behind AgentRegistrar protocol. Three implementations: KeycloakRegistrar — Keycloak Admin REST API (existing, refactored) EntraRegistrar — Microsoft Entra Agent ID platform (NEW) StubRegistrar — dev mode without real IdP Driver selection via AGENT_REGISTRAR env var: auto — prefers Entra if configured, Keycloak fallback, stub default keycloak — explicit Keycloak entra — explicit Entra Agent ID Entra integration: Registers agent as Entra app + service principal via Graph API Tags with delegation metadata (agent_type, delegator, governed:true) Client secret TTL matches delegation expiry Deletes application on revocation (cascades to SP) Uses msal for token acquisition Future: native Agent ID Blueprint API when GA Files: registrar.py — AgentRegistrar protocol + AgentCredentials dataclass registrar_keycloak.py — refactored from keycloak.py registrar_entra.py — NEW Entra Graph API driver registrar_stub.py — dev mode stub registrar_factory.py — driver selection factory delegation.py — updated to use registrar abstraction settings.py — added Entra config + agent_registrar field All 7 smoke tests pass with stub registrar. Implements GCAP-SPEC-LLM-PRINCIPAL-BROKER-0001 §4.1-§4.3. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
"""Stub registrar for development without a real IdP."""
|
|
|
|
import logging
|
|
|
|
from .registrar import AgentCredentials
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StubRegistrar:
|
|
"""AgentRegistrar stub — returns dev credentials without calling any IdP."""
|
|
|
|
async def register_agent(
|
|
self,
|
|
delegation_id: str,
|
|
agent_type: str,
|
|
delegator_id: str,
|
|
display_name: str,
|
|
expires_at: str,
|
|
metadata: dict | None = None,
|
|
) -> AgentCredentials:
|
|
logger.info("Stub registrar: register %s for delegation %s", agent_type, delegation_id)
|
|
return AgentCredentials(
|
|
client_id=f"stub-agent-{agent_type}-{delegation_id}",
|
|
client_secret=f"stub-secret-{delegation_id}",
|
|
agent_display_name=display_name,
|
|
idp_backend="stub",
|
|
)
|
|
|
|
async def delete_agent(self, client_id: str) -> bool:
|
|
logger.info("Stub registrar: delete %s", client_id)
|
|
return True
|
|
|
|
async def get_agent_token(self, client_id: str) -> str | None:
|
|
return f"stub-token-{client_id}"
|