Using SPIFFE/SPIRE Identity

import os
import json

# ── SPIRE identity (read once at startup) ─────────────────────
_spire_identity: dict[str, str] = {}
_SVID_DIR = os.environ.get("SPIFFE_SVID_DIR", "/spiffe")
_JWT_SVID_PATH = os.path.join(_SVID_DIR, "jwt_svid.token")
_X509_SVID_PATH = os.path.join(_SVID_DIR, "svid.pem")

def _load_spire_identity() -> dict[str, str]:
    """Read SPIRE SVID files and extract identity claims."""
    identity: dict[str, str] = {}
    try:
        if os.path.isfile(_JWT_SVID_PATH):
            with open(_JWT_SVID_PATH) as f:
                jwt_token = f.read().strip()
            # Decode payload without verification (local file, trusted)
            parts = jwt_token.split(".")
            if len(parts) >= 2:
                import base64 as _b64
                padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
                payload = json.loads(_b64.urlsafe_b64decode(padded))
                identity["spiffe.id"] = payload.get("sub", "")
                identity["spiffe.audience"] = str(payload.get("aud", ""))
                identity["spiffe.issuer"] = payload.get("iss", "")
                identity["spiffe.expiry"] = str(payload.get("exp", ""))
            identity["spiffe.jwt_svid"] = jwt_token[:80] + "..."
    except Exception as exc:
        print(f"[spire] Failed to read JWT SVID: {exc}", flush=True)
    try:
        if os.path.isfile(_X509_SVID_PATH):
            with open(_X509_SVID_PATH) as f:
                cert_pem = f.read().strip()
            identity["spiffe.x509_svid"] = cert_pem[:120] + "..."
    except Exception as exc:
        print(f"[spire] Failed to read X.509 SVID: {exc}", flush=True)
    return identity

_spire_identity = _load_spire_identity()
if _spire_identity:
    print(f"[spire] Identity loaded: "
          f"{_spire_identity.get('spiffe.id', 'unknown')}", flush=True)

import os
import json
from crewai import Agent, Task, Crew, Process

# ── Load SPIRE identity ──────────────────────────────────────
def _load_spire_identity() -> dict[str, str]:
    svid_dir = os.environ.get("SPIFFE_SVID_DIR", "/spiffe")
    jwt_path = os.path.join(svid_dir, "jwt_svid.token")
    identity = {}
    if os.path.isfile(jwt_path):
        with open(jwt_path) as f:
            token = f.read().strip()
        parts = token.split(".")
        if len(parts) >= 2:
            import base64
            padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(padded))
            identity["spiffe.id"] = payload.get("sub", "")
            identity["spiffe.audience"] = str(payload.get("aud", ""))
    return identity

spire_identity = _load_spire_identity()
if spire_identity:
    print(f"[spire] CrewAI agent identity: {spire_identity.get('spiffe.id')}")

# ── Define agents and tasks ──────────────────────────────────
researcher = Agent(
    role="Researcher",
    goal="Find accurate information on the given topic",
    backstory="You are an expert research analyst.",
    verbose=True,
)

crew = Crew(
    agents=[researcher],
    tasks=[Task(
        description="Research {topic}",
        expected_output="A summary of findings",
        agent=researcher,
    )],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"topic": "zero-trust agent identity"})

import os
import json
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# ── Load SPIRE identity ──────────────────────────────────────
def _load_spire_identity() -> dict[str, str]:
    svid_dir = os.environ.get("SPIFFE_SVID_DIR", "/spiffe")
    jwt_path = os.path.join(svid_dir, "jwt_svid.token")
    identity = {}
    if os.path.isfile(jwt_path):
        with open(jwt_path) as f:
            token = f.read().strip()
        parts = token.split(".")
        if len(parts) >= 2:
            import base64
            padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(padded))
            identity["spiffe.id"] = payload.get("sub", "")
            identity["spiffe.audience"] = str(payload.get("aud", ""))
    return identity

spire_identity = _load_spire_identity()
if spire_identity:
    print(f"[spire] AutoGen agent identity: {spire_identity.get('spiffe.id')}")

# ── Define agents ─────────────────────────────────────────────
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message="You are a planning agent. Break tasks into steps.",
)

executor = AssistantAgent(
    name="executor",
    model_client=model_client,
    system_message="You execute the plan. Say TERMINATE when done.",
)

team = RoundRobinGroupChat(
    participants=[planner, executor],
    termination_condition=TextMentionTermination("TERMINATE"),
    max_turns=6,
)

import asyncio
result = asyncio.run(
    team.run(task="Explain SPIFFE workload identity for AI agents")
)

import os
import json
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI

# ── Load SPIRE identity ──────────────────────────────────────
def _load_spire_identity() -> dict[str, str]:
    svid_dir = os.environ.get("SPIFFE_SVID_DIR", "/spiffe")
    jwt_path = os.path.join(svid_dir, "jwt_svid.token")
    identity = {}
    if os.path.isfile(jwt_path):
        with open(jwt_path) as f:
            token = f.read().strip()
        parts = token.split(".")
        if len(parts) >= 2:
            import base64
            padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(padded))
            identity["spiffe.id"] = payload.get("sub", "")
            identity["spiffe.audience"] = str(payload.get("aud", ""))
    return identity

spire_identity = _load_spire_identity()
if spire_identity:
    print(f"[spire] LlamaIndex agent identity: {spire_identity.get('spiffe.id')}")

# ── Build a RAG pipeline ─────────────────────────────────────
Settings.llm = OpenAI(model="gpt-4o-mini")

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

response = query_engine.query(
    "What are the key concepts in SPIFFE workload identity?"
)
print(response)

import os
import json
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# ── Load SPIRE identity ──────────────────────────────────────
def _load_spire_identity() -> dict[str, str]:
    svid_dir = os.environ.get("SPIFFE_SVID_DIR", "/spiffe")
    jwt_path = os.path.join(svid_dir, "jwt_svid.token")
    identity = {}
    if os.path.isfile(jwt_path):
        with open(jwt_path) as f:
            token = f.read().strip()
        parts = token.split(".")
        if len(parts) >= 2:
            import base64
            padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(padded))
            identity["spiffe.id"] = payload.get("sub", "")
            identity["spiffe.audience"] = str(payload.get("aud", ""))
    return identity

spire_identity = _load_spire_identity()
if spire_identity:
    print(f"[spire] ADK agent identity: {spire_identity.get('spiffe.id')}")

# ── Define an ADK agent ──────────────────────────────────────
def check_identity() -> dict:
    """Return the agent's SPIFFE workload identity."""
    return spire_identity or {"error": "No SPIRE identity available"}

agent = Agent(
    name="identity_agent",
    model="gemini-2.0-flash",
    description="An agent that can inspect its own workload identity",
    instruction="Help users understand SPIFFE workload identity. "
                "Use the check_identity tool to show the agent's identity.",
    tools=[check_identity],
)

session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name="identity_app",
                session_service=session_service)

session = session_service.create_session(
    app_name="identity_app", user_id="user1"
)

message = types.Content(
    role="user",
    parts=[types.Part(text="What is your workload identity?")],
)

import asyncio

async def run():
    async for event in runner.run_async(
        user_id="user1", session_id=session.id, new_message=message
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

asyncio.run(run())