Authentication Overview for Multi-Agent Systems¶

# 1. AUTHENTICATION: Verify who the agent is
def authenticate_agent(agent_card, signature):
    """Verify the agent is who they claim to be"""
    if not verify_signature(agent_card, signature):
        raise AuthenticationError("Invalid signature")

    if agent_card.is_expired():
        raise AuthenticationError("Card expired")

    return authenticated_agent_id  # Now we know WHO they are

# 2. AUTHORIZATION: Verify what they can do
def authorize_operation(agent_id, operation):
    """Check if agent has permission for this operation"""
    agent_capabilities = get_capabilities(agent_id)

    if operation not in agent_capabilities:
        raise AuthorizationError("Permission denied")

    return True  # They can perform this operation

class ZeroTrustValidator:
    """Every request is verified, regardless of source"""

    def handle_request(self, request, agent_card):
        # NEVER skip these checks

        # 1. Verify identity
        if not self.verify_identity(agent_card):
            return "Authentication failed"

        # 2. Verify authorization
        if not self.check_permissions(agent_card, request.operation):
            return "Authorization failed"

        # 3. Validate input
        if not self.validate_input(request.payload):
            return "Invalid input"

        # 4. Check rate limits
        if not self.check_rate_limit(agent_card.agent_id):
            return "Rate limit exceeded"

        # 5. Log everything
        self.audit_log(agent_card.agent_id, request.operation)

        # Only now process the request
        return self.process_request(request)

class WebOfTrust:
    def __init__(self):
        self.trust_relationships = {}  # agent_id -> set of trusted agents
        self.reputation_scores = {}    # agent_id -> score (0-1)

    def calculate_trust(self, source_agent, target_agent, max_hops=3):
        """Calculate trust through relationship chains"""

        # Direct trust
        if target_agent in self.trust_relationships.get(source_agent, set()):
            return 1.0  # Full trust

        # Transitive trust through intermediaries
        trust_chain = self.find_trust_path(source_agent, target_agent, max_hops)

        if trust_chain:
            # Trust decreases with chain length
            trust_score = 1.0
            for hop in trust_chain:
                trust_score *= self.reputation_scores.get(hop, 0.5)
            return trust_score

        return 0.0  # No trust relationship found

    def vouch_for_agent(self, voucher_id, target_id, strength=1.0):
        """One agent vouches for another"""
        if voucher_id not in self.trust_relationships:
            self.trust_relationships[voucher_id] = set()

        self.trust_relationships[voucher_id].add(target_id)

        # Update reputation score
        current_score = self.reputation_scores.get(target_id, 0.5)
        self.reputation_scores[target_id] = (current_score + strength) / 2

class PKIAuthenticator:
    def __init__(self, trusted_ca_certs):
        self.trusted_cas = trusted_ca_certs
        self.revoked_certs = self.fetch_crl()

    def verify_agent_certificate(self, agent_card):
        """Verify certificate chain back to trusted CA"""

        # 1. Get agent's certificate
        cert = self.get_certificate(agent_card.certificate_fingerprint)

        if not cert:
            return False, "Certificate not found"

        # 2. Check revocation
        if cert.fingerprint in self.revoked_certs:
            return False, "Certificate revoked"

        # 3. Verify certificate chain
        chain_valid, chain_error = self.verify_certificate_chain(
            cert,
            self.trusted_cas
        )

        if not chain_valid:
            return False, f"Chain invalid: {chain_error}"

        # 4. Verify cert matches public key in agent card
        if cert.public_key != agent_card.public_key:
            return False, "Public key mismatch"

        # 5. Check validity period
        if not cert.is_valid_at(datetime.now()):
            return False, "Certificate expired"

        return True, "Certificate valid"

    def verify_certificate_chain(self, cert, trusted_cas):
        """Walk the certificate chain to a trusted root"""
        current_cert = cert

        while True:
            # Is this cert issued by a trusted CA?
            if current_cert.issuer in [ca.subject for ca in trusted_cas]:
                return True, "Chain valid"

            # Get issuer's certificate
            issuer_cert = self.get_certificate_by_subject(current_cert.issuer)

            if not issuer_cert:
                return False, "Broken chain"

            # Verify signature
            if not self.verify_cert_signature(current_cert, issuer_cert):
                return False, "Invalid signature in chain"

            # Move up the chain
            current_cert = issuer_cert

class FederatedAuthenticator:
    def __init__(self):
        self.trusted_idps = {
            "org1": "https://idp.org1.com",
            "org2": "https://idp.org2.com"
        }
        self.trust_agreements = self.load_trust_agreements()

    def validate_agent_token(self, token):
        """Validate agent token from federated IdP"""

        # 1. Decode token (JWT)
        try:
            payload = jwt.decode(
                token,
                verify=False  # We'll verify the signature next
            )
        except:
            return False, "Invalid token format"

        # 2. Identify issuer
        issuer = payload.get("iss")

        if issuer not in self.trusted_idps.values():
            return False, "Untrusted issuer"

        # 3. Get issuer's public key
        issuer_public_key = self.fetch_idp_public_key(issuer)

        # 4. Verify signature
        try:
            jwt.decode(token, issuer_public_key, algorithms=["RS256"])
        except:
            return False, "Invalid signature"

        # 5. Check expiration
        exp = payload.get("exp")
        if datetime.now().timestamp() > exp:
            return False, "Token expired"

        # 6. Verify audience (is token meant for us?)
        aud = payload.get("aud")
        if aud != self.our_identifier:
            return False, "Token not for this recipient"

        # 7. Extract agent identity
        agent_id = payload.get("sub")

        return True, agent_id

def authenticate_with_shared_secret(agent_id, message, hmac_signature):
    """Verify message using shared secret"""

    # Get shared secret for this agent
    shared_secret = get_secret_for_agent(agent_id)

    # Calculate expected HMAC
    expected_hmac = hmac.new(
        shared_secret.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()

    # Compare (constant-time to prevent timing attacks)
    return hmac.compare_digest(hmac_signature, expected_hmac)

def authenticate_with_signature(agent_card, message, signature):
    """Verify message signature using agent's public key"""

    # Get agent's public key
    public_key = load_public_key(agent_card.public_key)

    # Verify signature
    try:
        public_key.verify(
            base64.b64decode(signature),
            message.encode(),
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        return True
    except:
        return False

def authenticate_with_certificate(agent_card, peer_connection):
    """Verify agent using TLS certificate"""

    # 1. Get peer certificate from TLS connection
    peer_cert = peer_connection.getpeercert()

    # 2. Extract certificate details
    cert_subject = peer_cert['subject']
    cert_fingerprint = hashlib.sha256(
        peer_cert.encode()
    ).hexdigest()

    # 3. Verify certificate matches agent card
    if cert_fingerprint != agent_card.certificate_fingerprint:
        return False, "Certificate mismatch"

    # 4. Verify certificate chain (done by TLS, but double-check)
    if not verify_certificate_chain(peer_cert):
        return False, "Invalid certificate chain"

    # 5. Check revocation
    if is_certificate_revoked(cert_fingerprint):
        return False, "Certificate revoked"

    return True, "Certificate valid"

class TokenAuthenticator:
    def __init__(self):
        self.active_tokens = {}  # token -> agent_id
        self.token_expiry = {}   # token -> expiration_time

    def issue_token(self, agent_id, duration_minutes=60):
        """Issue authentication token"""
        token = secrets.token_urlsafe(32)
        expiry = datetime.now() + timedelta(minutes=duration_minutes)

        self.active_tokens[token] = agent_id
        self.token_expiry[token] = expiry

        return {
            "token": token,
            "expires_at": expiry.isoformat(),
            "token_type": "Bearer"
        }

    def validate_token(self, token):
        """Validate authentication token"""

        # Check token exists
        if token not in self.active_tokens:
            return False, None, "Invalid token"

        # Check expiration
        if datetime.now() > self.token_expiry[token]:
            # Clean up expired token
            agent_id = self.active_tokens.pop(token)
            del self.token_expiry[token]
            return False, None, "Token expired"

        agent_id = self.active_tokens[token]
        return True, agent_id, "Token valid"

    def revoke_token(self, token):
        """Manually revoke token (e.g., on logout)"""
        if token in self.active_tokens:
            del self.active_tokens[token]
            del self.token_expiry[token]

# Layer multiple authentication mechanisms
def authenticate_agent(agent_card, connection, request):
    # Layer 1: TLS certificate
    if not verify_tls_certificate(connection):
        return False

    # Layer 2: Agent card signature
    if not verify_agent_card_signature(agent_card):
        return False

    # Layer 3: Request token
    if not verify_request_token(request.token):
        return False

    # All layers passed
    return True

# Grant minimum necessary permissions
agent_permissions = {
    "read_public_data": True,
    "read_private_data": False,  # Not needed
    "write_data": False,          # Not needed
    "admin": False                # Definitely not needed
}

def authenticate_agent(agent_card):
    try:
        result = verify_signature(agent_card)
        return result
    except Exception as e:
        # On error, DENY (don't default to allow)
        log_error(f"Authentication error: {e}")
        return False  # Fail closed

def authenticate_agent(agent_card):
    result = verify_signature(agent_card)

    # Log all authentication attempts
    audit_log.record({
        "event": "authentication_attempt",
        "agent_id": agent_card.agent_id,
        "timestamp": datetime.now(),
        "result": "success" if result else "failure",
        "source_ip": get_source_ip()
    })

    return result

# ❌ WRONG: Trust claimed identity
def handle_request(request):
    agent_id = request.headers["X-Agent-ID"]  # Attacker controls this!
    return process_for_agent(agent_id)

# ✅ CORRECT: Verify identity cryptographically
def handle_request(request, agent_card):
    if not verify_agent_card_signature(agent_card):
        raise AuthenticationError()

    agent_id = agent_card.agent_id  # Verified via signature
    return process_for_agent(agent_id)

# ❌ WRONG: Cache forever
authentication_cache = {}

def is_authenticated(agent_id):
    if agent_id in authentication_cache:
        return True  # Could be stale!
    # ... verify ...

# ✅ CORRECT: Time-limited cache
def is_authenticated(agent_id):
    cached = authentication_cache.get(agent_id)

    if cached and datetime.now() < cached['expires']:
        return True

    # Re-verify
    result = verify_agent(agent_id)
    authentication_cache[agent_id] = {
        'authenticated': result,
        'expires': datetime.now() + timedelta(minutes=5)
    }
    return result

# ❌ WRONG: Ignore expiration
def verify_agent_card(agent_card):
    return verify_signature(agent_card)  # Signature might be valid but card expired!

# ✅ CORRECT: Check expiration
def verify_agent_card(agent_card):
    if agent_card.is_expired():
        return False

    return verify_signature(agent_card)