AID-H-004.001: User & Privileged Access Management

Version 1.20261015 · Actionable sub-technique

Focuses on securing access for human users, such as developers, data scientists, and system administrators, who manage and interact with AI systems. The goal is to enforce strong authentication and granular permissions for human identities.

Pillars: infra

Lifecycle: building, operation

Scope boundary

Owns authentication, least privilege, privileged-access management, recertification, and deprovisioning for human developers, operators, data scientists, reviewers, and administrators accessing AI systems.

AID-H-004.002: AID-H-004.001 governs human and privileged identities; AID-H-004.002 authenticates workloads, services, jobs, applications, and external machine integrations. Human MFA and PAM do not establish workload identity, while service credentials do not grant a person privileged access.

Implementation guidance

AID-H-004.001-G001

Enforce Multi-Factor Authentication (MFA) for all users accessing sensitive AI environments.
Before you begin

Apply AID-H-004.001-G001 only when human users, administrators, reviewers, operators, or emergency responders can access sensitive AI assets or control planes.

Production implementation: enforce trusted MFA claims at the AI boundary

Configure phishing-resistant MFA and step-up policies in the enterprise identity provider, then make the AI gateway independently enforce signed OIDC assurance claims. Load the exact endpoints, trust path, positive deadlines, and decoded response/token caps from the signature-verified identity policy used by this code. A password plus an application flag is not MFA evidence.

# File: iam/verify_human_session.py
from __future__ import annotations
import json
import math
import ssl
import time
from urllib.parse import urlsplit, urlunsplit

import httpx
import jwt

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 canonical_https_url(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("/")
    ):
        raise ValueError("identity URL must be an exact HTTPS endpoint")
    if value != urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", "")):
        raise ValueError("identity URL is not canonical")
    return value


def bounded_json(response: httpx.Response, maximum_bytes: int) -> dict:
    if response.status_code != 200:
        raise PermissionError(f"identity 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 PermissionError("identity endpoint returned an unexpected media type")
    body = bytearray()
    for chunk in response.iter_bytes():
        body.extend(chunk)
        if len(body) > maximum_bytes:
            raise PermissionError("identity response exceeds the signed decoded-byte cap")
    value = json.loads(bytes(body), 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 PermissionError("identity response root must be an object")
    return value


def verify_session(token: str, policy: dict) -> dict:
    timeout_seconds = float(policy["identity_http_timeout_seconds"])
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("signed identity_http_timeout_seconds must be finite and positive")
    maximum_bytes = policy["identity_http_response_max_bytes"]
    token_maximum_bytes = policy["session_token_max_bytes"]
    if (
        isinstance(maximum_bytes, bool) or not isinstance(maximum_bytes, int) or maximum_bytes <= 0
        or isinstance(token_maximum_bytes, bool) or not isinstance(token_maximum_bytes, int) or token_maximum_bytes <= 0
        or not isinstance(token, str) or not token or len(token.encode("utf-8")) > token_maximum_bytes
    ):
        raise ValueError("signed identity byte bounds or token are invalid")
    discovery_url = canonical_https_url(policy["discovery_url"])
    jwks_url = canonical_https_url(policy["jwks_uri"])
    context = ssl.create_default_context(cafile=policy["identity_ca_bundle_path"])
    client_cert = policy.get("identity_client_cert_path")
    client_key = policy.get("identity_client_key_path")
    if bool(client_cert) != bool(client_key):
        raise ValueError("identity client certificate and key must be configured together")
    if client_cert:
        context.load_cert_chain(certfile=client_cert, keyfile=client_key)
    with httpx.Client(verify=context, timeout=timeout_seconds, follow_redirects=False) as client:
        with client.stream("GET", discovery_url, headers={"Accept": "application/json"}) as response:
            metadata = bounded_json(response, maximum_bytes)
        if metadata.get("issuer") != policy["issuer"] or metadata.get("jwks_uri") != jwks_url:
            raise PermissionError("OIDC metadata differs from signed policy")
        with client.stream("GET", jwks_url, headers={"Accept": "application/jwk-set+json, application/json"}) as response:
            jwks = jwt.PyJWKSet.from_dict(bounded_json(response, maximum_bytes))
    header = jwt.get_unverified_header(token)
    keys = [key for key in jwks.keys if key.key_id == header.get("kid")]
    if len(keys) != 1 or header.get("alg") not in policy["algorithms"]:
        raise PermissionError("session signing key or algorithm is not approved")
    claims = jwt.decode(token, keys[0].key, algorithms=policy["algorithms"], issuer=policy["issuer"], audience=policy["audience"], options={"require": ["exp", "iat", "sub", "jti", "acr", "amr", "auth_time"]})
    if claims["acr"] not in policy["allowed_acr_values"] or not set(policy["required_amr_values"]).issubset(set(claims["amr"])):
        raise PermissionError("authentication assurance is insufficient")
    if int(time.time()) - int(claims["auth_time"]) > policy["maximum_authentication_age_seconds"]:
        raise PermissionError("step-up authentication is stale")
    return {"subject": claims["sub"], "session_id": claims["jti"], "acr": claims["acr"], "amr": claims["amr"], "auth_time": claims["auth_time"], "expires_at": claims["exp"], "policy_version": policy["policy_version"]}

Verification: Read back IdP policy and enrolled-factor population through its administrative API; test valid WebAuthn/passkey step-up plus password-only, stale, wrong-audience, revoked-session, and recovery flows. Emergency access uses a separately governed PAM/break-glass path and never weakens normal policy.

AID-H-004.001-G002

Implement Role-Based Access Control (RBAC) to grant permissions based on job function.
Production implementation: centralized deny-by-default RBAC decision

Keep roles and resource scopes in version-controlled policy and require every AI API, model registry, feature store, and administrative action to call the same fail-closed authorization service. A bootstrap identity must signature-verify and export the exact decision URL, trust paths, positive deadline, and decoded response cap used below.

package ai.authz
default allow := false
permissions := {
  "data_scientist": {"training_job:create", "experiment:read"},
  "mlops_engineer": {"model:promote", "deployment:update"},
  "auditor": {"evidence:read", "model:read"},
  "application_user": {"inference:invoke"},
}
allow if {
  input.subject.active == true
  input.resource.environment == input.subject.environment
  some role in input.subject.roles
  input.action in permissions[role]
  input.resource.id in input.subject.resource_ids
  not input.resource.break_glass_only
}

# File: iam/authorize.py
import json
import math
import os
import ssl
from urllib.parse import urlsplit, urlunsplit

import httpx

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

RBAC_DECISION_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H004_RBAC_DECISION_TIMEOUT_SECONDS")
RBAC_DECISION_MAX_BYTES = int(os.environ["VERIFIED_H004_RBAC_DECISION_MAX_BYTES"])
if RBAC_DECISION_MAX_BYTES <= 0:
    raise RuntimeError("VERIFIED_H004_RBAC_DECISION_MAX_BYTES must be positive")


def canonical_url(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("RBAC decision 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 authorize(decision_input: dict) -> dict:
    context = ssl.create_default_context(cafile=os.environ["VERIFIED_H004_RBAC_CA_BUNDLE"])
    context.load_cert_chain(
        certfile=os.environ["VERIFIED_H004_RBAC_CLIENT_CERT"],
        keyfile=os.environ["VERIFIED_H004_RBAC_CLIENT_KEY"],
    )
    with httpx.Client(verify=context, timeout=RBAC_DECISION_TIMEOUT_SECONDS, follow_redirects=False) as client:
        with client.stream(
            "POST", canonical_url(os.environ["VERIFIED_H004_RBAC_DECISION_URL"]),
            json={"input": decision_input},
            headers={"Accept": "application/json"},
        ) as response:
            if response.status_code != 200:
                raise PermissionError(f"RBAC service returned HTTP {response.status_code}")
            if response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
                raise PermissionError("RBAC service returned an unexpected media type")
            raw = bytearray()
            for chunk in response.iter_bytes():
                raw.extend(chunk)
                if len(raw) > RBAC_DECISION_MAX_BYTES:
                    raise PermissionError("RBAC response exceeds the signed decoded-byte cap")
    body = 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(body, dict) or set(body) != {"result", "decision_id"}
        or body["result"] is not True
        or not isinstance(body["decision_id"], str) or not body["decision_id"]
    ):
        raise PermissionError("RBAC denied or decision evidence is invalid")
    return {"decision_id": body["decision_id"], "allowed": True}

Action: Reject direct grants outside approved roles, wildcard resources/actions, cross-environment inheritance, dormant identities, and ownerless roles. Independently enumerate effective permissions and compare them to the signed entitlement manifest; test every role/action pair plus cross-tenant and privilege-escalation negatives.

AID-H-004.001-G003

Utilize Privileged Access Management (PAM) solutions for administrators to control and audit high-risk actions.
Concept:

PAM systems provide a secure, audited gateway for users who need temporary, elevated access to critical systems. Instead of giving an administrator a permanent high-privilege role, they 'check out' the role through the PAM tool, with all actions being logged and session recordings enabled.

Configure Just-in-Time (JIT) Access

Use a PAM tool or a cloud-native equivalent to grant temporary, time-bound access. For example, AWS IAM Identity Center allows users to request access to a permission set for a specified duration.

# JIT privileged-access runbook

1.  **User Request:** An MLOps engineer needs to debug a production model server.
    They access the PAM portal (e.g., HashiCorp Boundary, CyberArk, or AWS IAM Identity Center).

2.  **Authentication:** The user authenticates to the PAM portal with their standard credentials and MFA.

3.  **Request Access:** The user requests access to the 'Production-ML-Debugger' role for a duration of '1 hour' and provides a justification (e.g., 'Investigating issue TICKET-123').

4.  **Approval (Optional):** The request can trigger an approval workflow, requiring a manager's sign-off.

5.  **Access Granted:** The PAM system dynamically grants the user a temporary session with the requested permissions. This might be an SSH session brokered by the PAM tool or temporary cloud credentials.

6.  **Auditing:** All commands executed and screens viewed during the session are recorded by the PAM tool.

7.  **Access Revoked:** After 1 hour, the session is automatically terminated, and the temporary credentials expire.

Action: For the most sensitive roles (like MLOps administrators with production access), integrate a PAM solution. Require that all privileged access is temporary, justified, and audited. Eliminate permanent, standing privileged access.

Keep the action safely bounded

deny or revoke access on absent strong authentication, unjustified entitlement, expired approval, orphan or stale identity, or effective readback mismatch.

AID-H-004.001-G004

Conduct periodic access recertification reviews for AI systems and privileged roles.
Production implementation: complete entitlement recertification

Export effective entitlements from the enforcement plane, not a hand-written spreadsheet. The signed review policy maps every resource/role to an accountable owner and defines the review window by privilege class.

# File: iam_reviews/close_review.py
from __future__ import annotations
import hashlib
import json
from pathlib import Path

def close_review(entitlements_path: Path, decisions_path: Path, *, review_id: str, policy: dict) -> dict:
    entitlements = json.loads(entitlements_path.read_text(encoding="utf-8"))
    decisions = json.loads(decisions_path.read_text(encoding="utf-8"))
    by_id = {item["entitlement_id"]: item for item in entitlements}
    if len(by_id) != len(entitlements) or not by_id:
        raise ValueError("entitlement population is empty or duplicated")
    decision_by_id = {item["entitlement_id"]: item for item in decisions}
    if len(decision_by_id) != len(decisions) or set(decision_by_id) != set(by_id):
        raise PermissionError("review decision population differs")
    revocations = []
    for entitlement_id, entitlement in by_id.items():
        decision = decision_by_id[entitlement_id]
        expected_owner = policy["owners_by_resource"][entitlement["resource_id"]]
        if decision["review_id"] != review_id or decision["reviewer_id"] != expected_owner or decision["decision"] not in {"retain", "revoke"} or not decision["ticket_id"]:
            raise PermissionError("reviewer, decision, or ticket binding differs")
        if decision["decision"] == "revoke":
            revocations.append({"entitlement_id": entitlement_id, "principal_id": entitlement["principal_id"], "resource_id": entitlement["resource_id"], "role": entitlement["role"], "ticket_id": decision["ticket_id"]})
    return {"schema_version": "aidefend.access-recertification.v1", "review_id": review_id, "population_count": len(by_id), "decision_count": len(decision_by_id), "revocation_count": len(revocations), "revocations": revocations, "entitlements_sha256": hashlib.sha256(entitlements_path.read_bytes()).hexdigest(), "decisions_sha256": hashlib.sha256(decisions_path.read_bytes()).hexdigest(), "policy_version": policy["policy_version"]}

Action: Sign the entitlement export and reviewer decisions, apply every revoke through the authoritative IAM API, then independently re-enumerate effective access until the revoked grants disappear. Silent carry-forward, ownerless roles, overdue privileged reviews, missing decisions, or unapplied revocations deny renewal.

Before you begin

Apply AID-H-004.001-G004 only when human users, administrators, reviewers, operators, or emergency responders can access sensitive AI assets or control planes.

AID-H-004.001-G005

Automatically disable and de-provision inactive or stale accounts and credentials.
Production implementation: policy-bound AWS access-key quarantine

This concrete adapter handles AWS IAM access keys; parallel inventory adapters must cover the enterprise IdP, service accounts, notebooks, PATs, and external credentials. Thresholds and quarantine duration come from the signature-verified policy by principal/credential class.

# File: iam_audits/quarantine_aws_keys.py
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import boto3

def inventory_and_quarantine(policy: dict, approved_exceptions: dict[str, dict]) -> list[dict]:
    iam = boto3.client("iam")
    actions = []
    user_paginator = iam.get_paginator("list_users")
    for user_page in user_paginator.paginate():
        for user in user_page["Users"]:
            user_name = user["UserName"]
            key_paginator = iam.get_paginator("list_access_keys")
            for key_page in key_paginator.paginate(UserName=user_name):
                for key in key_page["AccessKeyMetadata"]:
                    key_id = key["AccessKeyId"]
                    observed = iam.get_access_key_last_used(AccessKeyId=key_id)["AccessKeyLastUsed"].get("LastUsedDate")
                    threshold = timedelta(seconds=policy["aws_iam_access_key"]["maximum_inactive_seconds"])
                    stale = observed is None or observed < datetime.now(timezone.utc) - threshold
                    exception = approved_exceptions.get(key_id)
                    excepted = exception is not None and datetime.fromisoformat(exception["not_after"].replace("Z", "+00:00")) > datetime.now(timezone.utc) and exception["principal_id"] == user["UserId"]
                    action = "none"
                    if stale and not excepted and key["Status"] == "Active":
                        iam.update_access_key(UserName=user_name, AccessKeyId=key_id, Status="Inactive")
                        action = "quarantined"
                    actions.append({"principal_id": user["UserId"], "user_name": user_name, "credential_id": key_id, "last_used": observed.isoformat() if observed else None, "previous_status": key["Status"], "stale": stale, "exception_id": exception["exception_id"] if excepted else None, "action": action})
    return actions

Action: Record the complete inventory and each mutation, notify owner/manager through policy-configured routes, and delete only after the signed quarantine interval and a second approval. Independently read back every key status and run negative authentication tests. Pagination, API, time-source, exception-validation, or readback failure yields ERROR and blocks a false cleanup PASS.

Before you begin

Apply AID-H-004.001-G005 only when human users, administrators, reviewers, operators, or emergency responders can access sensitive AI assets or control planes.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-004.001: User & Privileged Access Management | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service