Scope boundary

Owns machine identity authentication at API, service-mesh, gateway, data-store, and MLOps boundaries, including issuer, audience, workload identity, credential freshness, proof, and route binding. It proves identity but does not by itself grant business-data authorization.

AID-H-004.001: AID-H-004.002 authenticates non-human workloads and services; AID-H-004.001 governs human and privileged users. The credential types, lifecycle, assurance methods, and review populations are different.

AID-H-004.003: AID-H-004.002 establishes the calling machine or workload identity; AID-H-004.003 protects the complete live agent-to-agent message and task protocol, including replay, sequence, context ownership, and per-request authorization. Authenticated transport identity is necessary but insufficient for secure agent interaction semantics.

AID-H-004.005: AID-H-004.002 authenticates a machine or workload at the channel or service boundary; AID-H-004.005 preserves applicable security-relevant message content, context, freshness, and replay state across the complete path. Authentication at one hop does not establish end-to-end message integrity after termination, forwarding, queuing, or asynchronous delivery.

Implementation guidance

• AID-H-004.002-G001: Use OAuth 2.0 client credentials flow for service-to-service authentication.
• AID-H-004.002-G002: Implement short-lived, securely managed API keys for external service access.
• AID-H-004.002-G003: Enforce mutual TLS (mTLS) for all internal API traffic to ensure both client and server are authenticated.
• AID-H-004.002-G004: Use cloud provider IAM roles (e.g., AWS IAM Roles for Service Accounts) for workloads running in the cloud.

AID-H-004.002-G001

Client credentials with bounded transport behavior

Register one confidential client per calling workload, grant only the API scope it needs, and keep the credential in a workload secret store. Before startup, a bootstrap identity signature-verifies and exports the exact token/JWKS endpoints, trust bundle, positive deadlines and decoded response/token caps, allowed JWT algorithms, issuer, and audience used below.

# File: client_service/auth.py
from __future__ import annotations

import json
import math
import os
import ssl
from urllib.parse import urlsplit, urlunsplit

import httpx


def positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be finite and positive")
    return value


def positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be positive")
    return value


def canonical_https(value: str) -> str:
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https" or not parsed.hostname
        or parsed.username is not None or parsed.password is not None
        or parsed.query or parsed.fragment or not parsed.path.startswith("/")
        or value != urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", ""))
    ):
        raise RuntimeError("token URL must be the exact canonical HTTPS endpoint")
    return value


def no_duplicates(pairs: list[tuple[str, object]]) -> dict:
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


TOKEN_URL = canonical_https(os.environ["IDP_TOKEN_URL"])
CLIENT_ID = os.environ["MY_APP_CLIENT_ID"]
CLIENT_SECRET = os.environ["MY_APP_CLIENT_SECRET"]
REQUESTED_SCOPE = os.environ["VERIFIED_H004_REQUESTED_SCOPE"]
TIMEOUT = positive_seconds("VERIFIED_H004_TOKEN_TIMEOUT_SECONDS")
MAX_BYTES = positive_int("VERIFIED_H004_TOKEN_RESPONSE_MAX_BYTES")


def get_access_token() -> str:
    context = ssl.create_default_context(cafile=os.environ["VERIFIED_H004_IDP_CA_BUNDLE"])
    with httpx.Client(verify=context, timeout=TIMEOUT, follow_redirects=False) as client:
        with client.stream(
            "POST", TOKEN_URL,
            data={"grant_type": "client_credentials", "scope": REQUESTED_SCOPE},
            auth=httpx.BasicAuth(CLIENT_ID, CLIENT_SECRET),
            headers={"Accept": "application/json"},
        ) as response:
            if response.status_code != 200:
                raise RuntimeError(f"identity provider returned HTTP {response.status_code}")
            if response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
                raise RuntimeError("identity provider returned an unexpected media type")
            raw = bytearray()
            for chunk in response.iter_bytes():
                raw.extend(chunk)
                if len(raw) > MAX_BYTES:
                    raise RuntimeError("token response exceeds the signed decoded-byte cap")
    payload = json.loads(bytes(raw), object_pairs_hook=no_duplicates, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
    allowed = {"access_token", "token_type", "expires_in", "scope"}
    if not isinstance(payload, dict) or not {"access_token", "token_type", "expires_in"}.issubset(payload) or not set(payload).issubset(allowed):
        raise RuntimeError("token response schema differs")
    token = payload["access_token"]
    if not isinstance(token, str) or not token or payload["token_type"].lower() != "bearer":
        raise RuntimeError("identity provider returned an invalid bearer token")
    if isinstance(payload["expires_in"], bool) or not isinstance(payload["expires_in"], int) or payload["expires_in"] <= 0:
        raise RuntimeError("identity provider returned an invalid lifetime")
    token.encode("ascii")
    return token

Verify JWT signature, issuer, audience, lifetime, and scope

# File: model_service/app.py
from __future__ import annotations

import json
import math
import os
import ssl
from urllib.parse import urlsplit, urlunsplit

import httpx
import jwt
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer


app = FastAPI()
bearer = HTTPBearer(auto_error=False)
ISSUER = os.environ["OIDC_ISSUER"]
AUDIENCE = os.environ["OIDC_AUDIENCE"]
JWKS_URL = os.environ["OIDC_JWKS_URL"]
ALGORITHMS = [item for item in os.environ["VERIFIED_H004_OIDC_JWT_ALGORITHMS"].split(",") if item]
SAFE_ALGORITHMS = {"RS256", "RS384", "RS512", "PS256", "PS384", "PS512", "ES256", "ES384", "ES512", "EdDSA"}
TIMEOUT = float(os.environ["VERIFIED_H004_JWKS_TIMEOUT_SECONDS"])
MAX_BYTES = int(os.environ["VERIFIED_H004_JWKS_RESPONSE_MAX_BYTES"])
TOKEN_MAX_BYTES = int(os.environ["VERIFIED_H004_ACCESS_TOKEN_MAX_BYTES"])
if not math.isfinite(TIMEOUT) or TIMEOUT <= 0 or MAX_BYTES <= 0 or TOKEN_MAX_BYTES <= 0 or not ALGORITHMS or not set(ALGORITHMS).issubset(SAFE_ALGORITHMS):
    raise RuntimeError("verified OIDC transport or algorithm policy is invalid")


def canonical_https(value: str) -> str:
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https" or not parsed.hostname
        or parsed.username is not None or parsed.password is not None
        or parsed.query or parsed.fragment or not parsed.path.startswith("/")
        or value != urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", ""))
    ):
        raise RuntimeError("JWKS URL must be the exact canonical HTTPS endpoint")
    return value


def no_duplicates(pairs: list[tuple[str, object]]) -> dict:
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def load_jwks() -> jwt.PyJWKSet:
    context = ssl.create_default_context(cafile=os.environ["VERIFIED_H004_IDP_CA_BUNDLE"])
    with httpx.Client(verify=context, timeout=TIMEOUT, follow_redirects=False) as client:
        with client.stream("GET", canonical_https(JWKS_URL), headers={"Accept": "application/jwk-set+json, application/json"}) as response:
            if response.status_code != 200:
                raise RuntimeError(f"JWKS endpoint returned HTTP {response.status_code}")
            media = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
            if media not in {"application/json", "application/jwk-set+json"}:
                raise RuntimeError("JWKS endpoint returned an unexpected media type")
            raw = bytearray()
            for chunk in response.iter_bytes():
                raw.extend(chunk)
                if len(raw) > MAX_BYTES:
                    raise RuntimeError("JWKS response exceeds the signed decoded-byte cap")
    value = json.loads(bytes(raw), object_pairs_hook=no_duplicates, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")))
    if not isinstance(value, dict):
        raise RuntimeError("JWKS root must be an object")
    return jwt.PyJWKSet.from_dict(value)


def validate_token(credentials: HTTPAuthorizationCredentials | None = Depends(bearer)) -> dict:
    if credentials is None or credentials.scheme.lower() != "bearer":
        raise HTTPException(status_code=401, detail="bearer token required")
    token = credentials.credentials
    if not isinstance(token, str) or len(token.encode("utf-8")) > TOKEN_MAX_BYTES:
        raise HTTPException(status_code=401, detail="invalid access token")
    try:
        header = jwt.get_unverified_header(token)
        if header.get("alg") not in ALGORITHMS or not isinstance(header.get("kid"), str):
            raise jwt.InvalidTokenError("unapproved header")
        keys = [key for key in load_jwks().keys if key.key_id == header["kid"]]
        if len(keys) != 1:
            raise jwt.InvalidTokenError("signing key is ambiguous or missing")
        claims = jwt.decode(
            token, keys[0].key, algorithms=ALGORITHMS,
            audience=AUDIENCE, issuer=ISSUER,
            options={"require": ["exp", "iat", "sub", "jti"]},
        )
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=503, detail="identity verification unavailable") from exc
    except (jwt.PyJWTError, ValueError, RuntimeError) as exc:
        raise HTTPException(status_code=401, detail="invalid access token") from exc
    scopes = set(str(claims.get("scope", "")).split())
    if os.environ["VERIFIED_H004_REQUIRED_SCOPE"] not in scopes:
        raise HTTPException(status_code=403, detail="required scope missing")
    return claims


@app.get("/authz-check")
def authz_check(claims: dict = Depends(validate_token)) -> dict:
    return {"status": "authorized", "subject": claims["sub"], "audience": AUDIENCE}

Verification and evidence

Replay a valid client, wrong audience, wrong issuer, expired token, unknown signing key, missing scope, revoked client, and unavailable JWKS endpoint. Require fail-closed behavior and no model-handler invocation for every negative case. Preserve client/workload identity, grant and scope policy versions, token jti or a non-reversible token fingerprint, signing kid, issuer/audience readback, decision, trace ID, and explicit PASS/FAIL/ERROR verdict. Never log the token or client secret.

Inference-runtime authentication must cover all enabled inference and management routes, including alternate endpoints outside a product's built-in API-key prefixes. Use the complete method/path registry and no-credential handler-invocation regressions in AID-H-038.002; checking only /v1 is insufficient.

AID-H-004.002-G002

Production implementation: non-recoverable API-key verification

Use API keys only when OAuth, mTLS, or workload identity is unavailable. Generate 256-bit secrets; expose each once; store only a keyed digest plus metadata. Rotation and overlap windows come from signed provider and organizational policy.

# File: gateway/api_keys.py
from __future__ import annotations
import base64
import hashlib
import hmac
import secrets
from datetime import datetime, timezone

def create_key(store, *, audience: str, scopes: list[str], not_before: datetime, not_after: datetime, pepper: bytes) -> str:
    if not_before.tzinfo is None or not_after.tzinfo is None or not not_before < not_after:
        raise ValueError("key validity interval is invalid")
    key_id, secret = secrets.token_hex(16), secrets.token_bytes(32)
    digest = hmac.new(pepper, key_id.encode() + b":" + secret, hashlib.sha256).hexdigest()
    store.insert_unique({"key_id": key_id, "digest": digest, "audience": audience, "scopes": sorted(set(scopes)), "not_before": not_before.isoformat(), "not_after": not_after.isoformat(), "status": "active"})
    return key_id + "." + base64.urlsafe_b64encode(secret).decode().rstrip("=")

def verify_key(presented: str, *, audience: str, required_scope: str, store, pepper: bytes) -> str:
    key_id, encoded = presented.split(".", 1)
    secret = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
    record, now = store.get_consistent(key_id), datetime.now(timezone.utc)
    if record is None or record["status"] != "active" or not datetime.fromisoformat(record["not_before"]) <= now < datetime.fromisoformat(record["not_after"]) or record["audience"] != audience or required_scope not in record["scopes"]:
        raise PermissionError("API key metadata denied")
    observed = hmac.new(pepper, key_id.encode() + b":" + secret, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(observed, record["digest"]):
        raise PermissionError("API key secret differs")
    return key_id

Action: Keep the pepper in HSM/KMS-backed secret storage, rate-limit by key/principal, log only key ID, revoke old and new keys on suspected compromise, and verify provider-side effective state. Missing expiry, recoverable storage, excess scope, stale overlap, or readback mismatch denies use.

AID-H-004.002-G003

Before you begin

Apply AID-H-004.002-G003 only when a workload, service, API client, job, pod, function, or external integration crosses an authenticated service boundary.

Production implementation: Istio stable API with identity authorization

apiVersion: security.istio.io/v1
kind: PeerAuthentication
metadata:
  name: strict-mtls
  namespace: ai-services
spec:
  mtls:
    mode: STRICT
---
apiVersion: security.istio.io/v1
kind: AuthorizationPolicy
metadata:
  name: inference-callers
  namespace: ai-services
spec:
  selector:
    matchLabels:
      app: inference-api
  action: ALLOW
  rules:
    - from:
        - source:
            principals:
              - cluster.local/ns/ai-clients/sa/approved-agent
      to:
        - operation:
            methods: ["POST"]
            paths: ["/v1/inference"]

Verification: Read back effective PeerAuthentication, AuthorizationPolicy, proxy certificate/SPIFFE identity, trust domain, and proxy config for every selected pod. Test approved identity, unapproved service account, plaintext, wrong trust domain, expired certificate, alternate method/path, bypass path, and newly rolled pod. mTLS authenticates; authorization separately limits the route.

Keep the action safely bounded

deny issuance or request delivery on identity, audience, scope, route, freshness, proof, trust, or effective-policy mismatch.

AID-H-004.002-G004

Before you begin

Apply AID-H-004.002-G004 only when a workload, service, API client, job, pod, function, or external integration crosses an authenticated service boundary.

Production implementation: EKS Pod Identity without static cloud keys

This example uses current EKS Pod Identity. The cluster must run the Pod Identity Agent and the signed workload inventory binds cluster, namespace, service account, role, and image digest.

variable "cluster_name" { type = string }
variable "namespace" { type = string }
variable "service_account_name" { type = string }
variable "model_bucket_arn" { type = string }
variable "workload_image" {
  type = string
  validation {
    condition     = can(regex("@sha256:[a-f0-9]{64}$", var.workload_image))
    error_message = "workload_image must be digest pinned"
  }
}
data "aws_iam_policy_document" "pod_identity_trust" {
  statement {
    actions = ["sts:AssumeRole", "sts:TagSession"]
    principals {
      type        = "Service"
      identifiers = ["pods.eks.amazonaws.com"]
    }
  }
}
resource "aws_iam_role" "inference" {
  name               = join("-", ["aidefend-inference", var.cluster_name])
  assume_role_policy = data.aws_iam_policy_document.pod_identity_trust.json
}
data "aws_iam_policy_document" "model_read" {
  statement {
    actions   = ["s3:GetObject"]
    resources = [join("", [var.model_bucket_arn, "/approved-models/*"])]
  }
}
resource "aws_iam_role_policy" "model_read" {
  role   = aws_iam_role.inference.id
  policy = data.aws_iam_policy_document.model_read.json
}
resource "aws_eks_pod_identity_association" "inference" {
  cluster_name    = var.cluster_name
  namespace       = var.namespace
  service_account = var.service_account_name
  role_arn        = aws_iam_role.inference.arn
}

Verification: Read back association and IAM policy from AWS, verify the Kubernetes service account and every selected pod, and call STS GetCallerIdentity plus an allowed model read from the workload. A different namespace/service account must fail; static AWS keys and node-role fallback must be absent. Bind cluster UID, association ID, role/policy digests, pod/service-account UIDs, agent version, and image digest.

Keep the action safely bounded

deny issuance or request delivery on identity, audience, scope, route, freshness, proof, trust, or effective-policy mismatch.

Open-source tools

• oauth2-proxy
• SPIRE
• Istio
• Linkerd
• OpenBao (dynamic workload credentials and PKI)

Source-available / open-weight tools

• HashiCorp Vault Community (BUSL-1.1; source-available)

Commercial / hosted tools

• Kong Gateway Enterprise
• Kong Konnect
• Google Cloud Apigee API Management
• MuleSoft Anypoint Platform
• AWS Secrets Manager
• Azure Key Vault

Threat mappings

MITRE ATLAS

• AML.T0040 AI Model Inference API Access
• AML.T0122 Exploitation of Remote Services (unauthenticated workloads cannot reach AI API or data-store handlers)
• AML.T0132 Misconfigured or Publicly Exposed AI Services (service authentication rejects unauthenticated access across enabled runtime routes)

MAESTRO

• Agent Impersonation (L7) (service authentication prevents agent impersonation)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI03:2026 Identity and Privilege Abuse (service auth enforces identity verification)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls
• AISubtech-3.1.2 Trusted Agent Spoofing (Authenticates the explicitly approved agent workload and route)
• AITech-14.1 Unauthorized Access
• AITech-16.1 Eavesdropping
• AITech-3.1 Masquerading / Obfuscation / Impersonation (Authenticates the explicitly approved agent workload and route)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.11: Model Inference API Access
• Agents - Core 13.9: Identity Spoofing & Impersonation