IATP

from agent_os.iatp import AgentIdentity, TrustRegistry

# Create a cryptographic identity for an agent
agent = AgentIdentity.create("finance-agent")

# Sign a message proving authenticity
signed_msg = agent.sign({
    "action": "approve_transaction",
    "amount": 50000,
    "recipient": "vendor-123"
})

# Any agent can verify the signature
is_valid = AgentIdentity.verify(signed_msg)
# is_valid: True

# Standalone installation
pip install agent-os-iatp

# With full kernel
pip install agent-os-kernel[iatp]

# With hardware security module support
pip install agent-os-iatp[hsm]

# With distributed registry support
pip install agent-os-iatp[distributed]

from agent_os.iatp import AgentIdentity

# Create a new identity with auto-generated keys
agent = AgentIdentity.create(
    agent_id="research-agent",
    metadata={
        "role": "researcher",
        "department": "R&D",
        "created_by": "system-admin"
    }
)

# Access identity properties
print(agent.agent_id)      # "research-agent"
print(agent.public_key)    # Base64-encoded public key
print(agent.fingerprint)   # SHA256 fingerprint for easy identification

# Export public key only (safe to share)
public_export = agent.export_public()
# {
#     "agent_id": "research-agent",
#     "public_key": "base64...",
#     "fingerprint": "sha256:abc123...",
#     "metadata": {...}
# }

# Export full identity (includes private key - KEEP SECURE!)
full_export = agent.export(
    passphrase="secure-passphrase-here",
    format="pem"  # or "jwk", "raw"
)

# Save to file
agent.save("./identities/research-agent.pem", passphrase="secret")

# Import from exported data
agent = AgentIdentity.from_export(
    export_data,
    passphrase="secure-passphrase-here"
)

# Load from file
agent = AgentIdentity.load(
    "./identities/research-agent.pem",
    passphrase="secret"
)

# Import public key only (for verification)
public_agent = AgentIdentity.from_public_key(
    agent_id="research-agent",
    public_key="base64-encoded-public-key..."
)

# Create identity with specific algorithm
agent = AgentIdentity.create(
    agent_id="compliance-agent",
    algorithm="ecdsa-p384"  # NIST-approved
)

from agent_os.iatp import AgentIdentity, SignedMessage

agent = AgentIdentity.create("trading-agent")

# Sign any serializable data
signed = agent.sign({
    "action": "execute_trade",
    "symbol": "AAPL",
    "quantity": 100,
    "price": 150.00,
    "timestamp": "2024-01-15T10:30:00Z"
})

# SignedMessage properties
print(signed.payload)      # Original message
print(signed.signature)    # Base64 signature
print(signed.signer_id)    # "trading-agent"
print(signed.public_key)   # Signer's public key
print(signed.timestamp)    # When it was signed
print(signed.algorithm)    # "ed25519"

# Method 1: Verify with embedded public key
is_valid = SignedMessage.verify(signed)

# Method 2: Verify against known identity
is_valid = agent.verify_signature(signed)

# Method 3: Verify against TrustRegistry
registry = TrustRegistry()
is_valid, trust_level = registry.verify(signed)

# Comprehensive verification with details
result = SignedMessage.verify_detailed(signed)
# {
#     "valid": True,
#     "signer_id": "trading-agent",
#     "fingerprint": "sha256:abc123...",
#     "signed_at": "2024-01-15T10:30:00Z",
#     "algorithm": "ed25519",
#     "payload_hash": "sha256:def456..."
# }

# Agent A initiates a request
agent_a = AgentIdentity.create("requester")
request = agent_a.sign({
    "type": "approval_request",
    "amount": 10000
})

# Agent B approves (chains to original)
agent_b = AgentIdentity.create("approver")
approval = agent_b.sign({
    "type": "approval",
    "original_request": request.to_dict(),
    "decision": "approved"
})

# Verify the entire chain
chain_valid = SignedMessage.verify_chain([request, approval])

from agent_os.iatp import TrustRegistry, TrustLevel

# In-memory registry (for development)
registry = TrustRegistry()

# Persistent registry (for production)
registry = TrustRegistry(
    backend="sqlite",
    path="./trust-registry.db"
)

# Distributed registry (for multi-node deployments)
registry = TrustRegistry(
    backend="redis",
    url="redis://localhost:6379",
    namespace="prod-agents"
)

# Register an agent with a trust level
agent = AgentIdentity.create("data-processor")

registry.register(
    identity=agent,
    trust_level=TrustLevel.MEDIUM,
    capabilities=["read_data", "process_data"],
    expires_at="2025-01-01T00:00:00Z",  # Optional expiration
    metadata={
        "registered_by": "admin",
        "department": "engineering"
    }
)

# Bulk registration
registry.register_many([
    (agent1, TrustLevel.HIGH, ["admin"]),
    (agent2, TrustLevel.LOW, ["read_only"]),
    (agent3, TrustLevel.MEDIUM, ["read", "write"]),
])

# Check if an agent is registered and trusted
is_trusted = registry.is_trusted(agent.fingerprint)

# Get detailed trust information
trust_info = registry.get_trust_info(agent.fingerprint)
# {
#     "agent_id": "data-processor",
#     "trust_level": TrustLevel.MEDIUM,
#     "capabilities": ["read_data", "process_data"],
#     "registered_at": "2024-01-10T...",
#     "expires_at": "2025-01-01T...",
#     "revoked": False
# }

# Verify a signed message with trust context
result = registry.verify_message(signed_message)
# {
#     "signature_valid": True,
#     "signer_trusted": True,
#     "trust_level": TrustLevel.MEDIUM,
#     "has_capability": True  # If capability check requested
# }

# Revoke an agent's trust (e.g., compromised key)
registry.revoke(
    fingerprint=agent.fingerprint,
    reason="Key compromised",
    revoked_by="security-admin"
)

# Check revocation status
is_revoked = registry.is_revoked(agent.fingerprint)

# Get revocation details
revocation = registry.get_revocation(agent.fingerprint)
# {
#     "revoked_at": "2024-01-15T...",
#     "reason": "Key compromised",
#     "revoked_by": "security-admin"
# }

from agent_os.iatp import TrustLevel

# Compare trust levels
if agent_trust >= TrustLevel.MEDIUM:
    allow_write_access()

# Trust level gates
@registry.require_trust(TrustLevel.HIGH)
async def sensitive_operation(signed_request):
    # Only HIGH or ADMIN trust can execute this
    pass

# Dynamic trust requirements
required_level = TrustLevel.from_action("delete_user")  # Returns HIGH
if registry.get_trust_level(agent) >= required_level:
    proceed_with_action()

# Define trust policies
policy = TrustPolicy(
    default_level=TrustLevel.LOW,
    action_requirements={
        "read_*": TrustLevel.LOW,
        "write_*": TrustLevel.MEDIUM,
        "delete_*": TrustLevel.HIGH,
        "admin_*": TrustLevel.ADMIN
    },
    auto_elevate=False,  # Require explicit elevation
    require_mfa_above=TrustLevel.HIGH  # MFA for HIGH+
)

registry = TrustRegistry(policy=policy)

from agent_os.iatp import Capability, AgentIdentity

# Admin agent creates a capability
admin = AgentIdentity.load("admin.pem", passphrase="...")

capability = Capability.create(
    issuer=admin,
    subject="data-processor",  # Who receives this capability
    actions=["read_customer_data", "process_orders"],
    resources=["customers/*", "orders/*"],
    constraints={
        "max_records": 1000,
        "rate_limit": "100/minute"
    },
    valid_from="2024-01-01T00:00:00Z",
    valid_until="2024-12-31T23:59:59Z",
    delegatable=False  # Cannot be passed to other agents
)

# Agent attaches capability to request
agent = AgentIdentity.create("data-processor")

request = agent.sign({
    "action": "read_customer_data",
    "customer_id": "cust-123"
})

# Attach the capability as proof of authorization
request.attach_capability(capability)

# Verifier checks both signature AND capability
result = registry.verify_authorized(request)
# {
#     "signature_valid": True,
#     "capability_valid": True,
#     "action_authorized": True,
#     "constraints_satisfied": True
# }

# Create a delegatable capability
admin_cap = Capability.create(
    issuer=admin,
    subject="team-lead",
    actions=["read_*", "write_*"],
    delegatable=True,
    max_delegation_depth=2  # Can delegate up to 2 levels deep
)

# Team lead delegates subset to team member
team_lead = AgentIdentity.create("team-lead")
delegated_cap = admin_cap.delegate(
    delegator=team_lead,
    subject="team-member",
    actions=["read_*"],  # Can only delegate subset
    valid_until="2024-06-30T23:59:59Z"  # Cannot exceed parent
)

# Verify delegation chain
chain_valid = Capability.verify_delegation_chain(delegated_cap)

from agent_os.iatp import AgentIdentity, TrustRegistry

agent = AgentIdentity.load("agent.pem", passphrase="old-secret")

# Rotate to new key pair
new_agent = agent.rotate_keys(
    new_passphrase="new-secret",
    rotation_proof=True  # Signs new key with old key
)

# Update in registry (maintains trust level and history)
registry.rotate_identity(
    old_fingerprint=agent.fingerprint,
    new_identity=new_agent,
    reason="Scheduled rotation"
)

# Old key is automatically added to revocation list
# but messages signed before rotation remain valid

from agent_os.iatp import RotationPolicy

policy = RotationPolicy(
    max_key_age_days=90,
    warn_before_days=14,
    auto_rotate=True,
    notify_on_rotation=["security@company.com"],
    backup_old_keys=True,
    backup_retention_days=365
)

registry = TrustRegistry(rotation_policy=policy)

# Check rotation status
status = registry.check_rotation_status()
# [
#     {"agent_id": "agent-1", "days_until_rotation": 5, "status": "warning"},
#     {"agent_id": "agent-2", "days_until_rotation": 45, "status": "ok"},
# ]

# Immediate revocation (compromised key)
registry.revoke_immediately(
    fingerprint=compromised_agent.fingerprint,
    reason="Key compromised - security incident #1234",
    revoked_by="security-team",
    broadcast=True  # Notify all connected registries
)

# Revoke all capabilities issued by this agent
registry.revoke_capabilities_by_issuer(compromised_agent.fingerprint)

# Revoke with grace period (planned decommission)
registry.revoke_scheduled(
    fingerprint=old_agent.fingerprint,
    effective_at="2024-02-01T00:00:00Z",
    reason="Agent decommissioned"
)

# Get current CRL
crl = registry.get_revocation_list()

# Export CRL for offline verification
crl_data = registry.export_crl(format="json")

# Check against CRL
is_revoked = registry.check_crl(
    fingerprint=agent.fingerprint,
    crl=crl_data
)

# Subscribe to revocation updates
@registry.on_revocation
async def handle_revocation(event):
    print(f"Agent {event.agent_id} was revoked: {event.reason}")
    await invalidate_cached_sessions(event.fingerprint)

from agent_os.iatp import TrustRegistry, TrustFederation

# Create local registry
local_registry = TrustRegistry(
    node_id="us-east-1",
    backend="postgres",
    url="postgresql://..."
)

# Join a federation
federation = TrustFederation(
    registries=[
        "https://trust.us-east-1.example.com",
        "https://trust.eu-west-1.example.com",
        "https://trust.ap-south-1.example.com"
    ],
    consensus="majority",  # 2 of 3 must agree
    sync_interval_seconds=30
)

local_registry.join_federation(federation)

# Trust an external organization's registry
local_registry.add_trusted_issuer(
    issuer_id="partner-org",
    public_key="partner-public-key...",
    trust_level=TrustLevel.LOW,  # External agents get LOW by default
    capabilities=["read_shared_data"],
    namespace="partner:*"  # Prefix their agent IDs
)

# Verify message from partner's agent
result = local_registry.verify_message(partner_signed_message)
# {
#     "signature_valid": True,
#     "issuer": "partner-org",
#     "trust_level": TrustLevel.LOW,
#     "namespace": "partner:agent-1"
# }

# Enable gossip for revocation propagation
local_registry.enable_gossip(
    peers=[
        "registry-node-2:7946",
        "registry-node-3:7946"
    ],
    gossip_interval_ms=500,
    propagation_factor=3
)

# Revocations automatically propagate across network
# within seconds, even in large deployments

# Monitor gossip health
health = local_registry.gossip_health()
# {
#     "connected_peers": 15,
#     "avg_propagation_ms": 120,
#     "last_sync": "2024-01-15T10:30:00Z"
# }

from agent_os import KernelSpace
                from agent_os.iatp import TrustRegistry, TrustLevel

# Initialize kernel with IATP integration
kernel = KernelSpace(
    trust_registry=TrustRegistry(backend="redis"),
    require_signatures=True,  # All agent actions must be signed
    min_trust_level=TrustLevel.LOW
)

# Define trust-aware policies
kernel.add_policy({
    "name": "financial-operations",
    "rules": [
        {
            "action": "transfer_funds",
            "require": {
                "trust_level": "HIGH",
                "capabilities": ["financial_admin"],
                "signature": "required"
            }
        },
        {
            "action": "view_balance",
            "require": {
                "trust_level": "LOW",
                "signature": "required"
            }
        }
    ]
})

from agent_os.iatp import AgentIdentity

# Agent with identity
agent = AgentIdentity.create("finance-bot")
kernel.register_agent(agent)

@kernel.action("transfer_funds")
async def transfer(request: SignedRequest):
    # Request is automatically verified by kernel
    # Trust level and capabilities are checked
    
    # Access verified identity info
    print(request.signer.agent_id)
    print(request.signer.trust_level)
    print(request.signer.capabilities)
    
    # Perform transfer
    return await execute_transfer(request.payload)

# All actions are automatically signed
result = await kernel.execute(
    agent=agent,
    action="transfer_funds",
    params={"amount": 1000, "to": "vendor-123"}
)

# IATP + EMK for complete audit trails
from agent_os.emk import Memory

kernel = KernelSpace(
    trust_registry=TrustRegistry(),
    memory=Memory(immutable=True)
)

# All signed actions are logged with cryptographic proof
# Query audit trail with signature verification
audit_log = await kernel.memory.query(
    filter={"action": "transfer_funds"},
    verify_signatures=True  # Re-verify all signatures
)

for entry in audit_log:
    print(f"{entry.timestamp}: {entry.signer} - {entry.action}")
    print(f"  Signature valid: {entry.signature_valid}")
    print(f"  Trust level at time: {entry.trust_level}")

from agent_os import KernelSpace
from agent_os.iatp import AgentIdentity, TrustRegistry, TrustLevel, Capability
from agent_os.amb import MessageBus
from agent_os.emk import Memory

# Initialize infrastructure
registry = TrustRegistry(backend="redis")
bus = MessageBus(backend="redis")
memory = Memory(backend="postgres")

kernel = KernelSpace(
    trust_registry=registry,
    message_bus=bus,
    memory=memory,
    policy="strict"
)

# Create agents with identities
coordinator = AgentIdentity.create("coordinator")
worker_1 = AgentIdentity.create("worker-1")
worker_2 = AgentIdentity.create("worker-2")

# Register with trust levels
registry.register(coordinator, TrustLevel.HIGH, ["coordinate", "delegate"])
registry.register(worker_1, TrustLevel.MEDIUM, ["process_data"])
registry.register(worker_2, TrustLevel.MEDIUM, ["process_data"])

# Coordinator delegates work with capability
@kernel.agent(coordinator)
async def coordinate_work(task):
    # Create capability for this specific task
    task_capability = Capability.create(
        issuer=coordinator,
        subject=worker_1.agent_id,
        actions=["process_data"],
        resources=[f"task:{task.id}/*"],
        valid_until=task.deadline
    )
    
    # Send signed task with capability
    await bus.publish(
        topic="tasks",
        message=coordinator.sign({
            "task_id": task.id,
            "data": task.data,
            "capability": task_capability.to_dict()
        })
    )

# Worker processes with verified identity
@kernel.agent(worker_1)
async def process_work(signed_message):
    # Kernel automatically verifies signature and capability
    task = signed_message.payload
    
    result = await process_data(task["data"])
    
    # Sign and return result
    return worker_1.sign({
        "task_id": task["task_id"],
        "result": result,
        "processed_by": worker_1.agent_id
    })

# Run the multi-agent system
await kernel.run()