AID-H-029.002: Lifecycle-Stage Authorization Gate

Version 1.20261015 · Actionable sub-technique

Implement a fail-closed authorization gate at each AI lifecycle-stage boundary. Before any data asset is consumed by training, fine-tuning, evaluation, RAG indexing, inference context assembly, logging, memory write, or retraining, the gate evaluates the asset's data-use tags against the requested stage and context. Data lacking valid authorization for the target stage is denied and logged as a policy event. At the memory-write boundary, require a purpose-bound grant for the exact candidate in addition to eligibility evidence and stage authorization, and reject source-trust, behavioral-policy, or tool-permission changes through ordinary memory APIs, directing them to a separately authorized policy/configuration release process.

Pillars: data, app

Lifecycle: building, operation

Scope boundary

Owns the fail-closed allow or deny decision immediately before data crosses into training, fine-tuning, evaluation, indexing, inference context, logging, memory, or retraining, including enforcement of a purpose-bound, candidate-bound memory-write grant. It consumes authoritative tags, eligibility evidence, and write grants rather than creating them.

AID-H-029.001: AID-H-029.002 enforces lifecycle-stage admission; AID-H-029.001 defines the tags and policy facts used in that decision. The gate must bind its verdict to the exact asset, requested stage, and current tag version.

AID-M-002.004: AID-H-029.002 owns the runtime stage allow or deny decision; AID-M-002.004 supplies memory or knowledge-base write-eligibility evidence, trust tier, provenance, and promotion state. Eligibility evidence is an input and cannot silently become authorization.

AID-H-018.005: AID-H-029.002 decides whether the asset may be used in the requested AI stage; AID-H-018.005 decides whether a value may reach a specific runtime sink. Stage admission and sink authorization must both pass when both apply.

Implementation guidance

AID-H-029.002-G001

Deploy an OPA or Cedar authorization check at every lifecycle-stage ingress point and deny data that lacks stage-specific authorization.
Resolve trusted stage and purpose context

Before a stage reads or copies an asset, authenticate the workload and resolve the exact asset/version and current tags through the authoritative registry. Bind the tags to the held asset bytes and keep that binding stable through the read or copy. The stage comes from the ingress route; the purpose must be explicit and authorized for that stage. Never accept asset tags, requester identity or policy data directly from user/model request JSON. Stage authorization is separate from any additional caller/object access check.

Check every stage with the same expiry and purpose rule

This Rego v1 policy denies missing or malformed inputs. The stage allowlist comes from the verified deployment policy, not the request. An explicit null expiry is accepted only when the authoritative grant permits no time limit; missing, invalid or elapsed expiry denies. Withdrawal denies all stages. Keep the existing restricted-data prohibition for RAG indexing in addition to the exact stage/purpose grant.

# File: datause/stage_authorization.rego
package datause
import rego.v1

default allow := false

nonempty_text(value) if {
    is_string(value)
    count(value) > 0
}

expiry_valid if { input.asset.consent_expiry == null }
expiry_valid if {
    is_string(input.asset.consent_expiry)
    time.now_ns() < time.parse_rfc3339_ns(input.asset.consent_expiry)
}

stage_restrictions if { input.target_stage != "rag_indexing" }
stage_restrictions if {
    input.target_stage == "rag_indexing"
    input.asset.classification_tier != "restricted"
}

allow if {
    nonempty_text(input.asset.asset_id)
    nonempty_text(input.asset.asset_version)
    nonempty_text(input.asset.tag_version)
    nonempty_text(input.asset.classification_tier)
    nonempty_text(input.requester)
    nonempty_text(input.target_stage)
    nonempty_text(input.requested_purpose)
    is_array(data.datause_stage_policy.allowed_stages)
    input.target_stage in data.datause_stage_policy.allowed_stages
    is_array(input.asset.authorized_stages)
    input.target_stage in input.asset.authorized_stages
    # Bind purpose to this exact stage; do not cross-join independent lists.
    is_object(input.asset.authorized_purposes)
    purposes := input.asset.authorized_purposes[input.target_stage]
    is_array(purposes)
    input.requested_purpose in purposes
    input.asset.authorization_withdrawn == false
    expiry_valid
    stage_restrictions
}
Configure the approved stage vocabulary

Load this illustrative data document together with the Rego module through the verified OPA bundle path. Include only the lifecycle stages approved for this deployment and use the same exact vocabulary in authoritative tags and ingress routes. The example grant and date below are test data, not authorization or a default retention/consent period.

{
  "datause_stage_policy": {
    "allowed_stages": [
      "training",
      "fine_tuning",
      "evaluation",
      "rag_indexing",
      "inference_context",
      "logging",
      "memory_write",
      "retraining"
    ]
  }
}

{
  "input": {
    "asset": {
      "schema_version": "2.0",
      "asset_id": "dataset-example",
      "asset_version": "release-example",
      "tag_version": "tags-example",
      "classification_tier": "internal",
      "authorized_stages": [
        "evaluation",
        "rag_indexing"
      ],
      "authorized_purposes": {
        "evaluation": [
          "quality-evaluation"
        ],
        "rag_indexing": [
          "support-retrieval"
        ]
      },
      "authorization_withdrawn": false,
      "consent_expiry": "2030-01-01T00:00:00Z",
      "consent_type": "contract",
      "jurisdiction": "EU",
      "data_owner": "data-team"
    },
    "target_stage": "rag_indexing",
    "requested_purpose": "support-retrieval",
    "requester": "authenticated-indexer"
  }
}
Enforce the decision before consuming data
# File: datause/stage_authorization.py
import json
import math
from urllib.parse import urlsplit, urlunsplit
import requests

def _unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON response key")
        result[key] = value
    return result

def check_stage_authorization(opa_url, asset_tags, target_stage, requested_purpose,
                              requester, timeout_seconds: float,
                              max_response_bytes: int, ca_bundle: str,
                              client_cert: str, client_key: str):
    if (type(timeout_seconds) not in (int, float)
            or not math.isfinite(timeout_seconds) or timeout_seconds <= 0
            or type(max_response_bytes) is not int or max_response_bytes <= 0):
        raise ValueError("finite positive timeout and integer response-byte cap required")
    if any(not isinstance(x, str) or not x for x in (target_stage, requested_purpose, requester)):
        raise ValueError("stage, purpose and authenticated requester are required")
    if any(not isinstance(x, str) or not x for x in (ca_bundle, client_cert, client_key)):
        raise ValueError("explicit TLS trust and client certificate paths required")
    parsed = urlsplit(opa_url)
    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 parsed.path != "/v1/data/datause/allow"
            or opa_url != urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", ""))):
        raise ValueError("exact canonical HTTPS datause decision endpoint required")
    payload = {"input": {"asset": asset_tags, "target_stage": target_stage,
                         "requested_purpose": requested_purpose, "requester": requester}}
    with requests.post(opa_url, json=payload, timeout=timeout_seconds,
                       allow_redirects=False, verify=ca_bundle,
                       cert=(client_cert, client_key), stream=True) as response:
        if response.status_code != 200:
            raise RuntimeError("OPA did not return a successful decision")
        raw = bytearray()
        for chunk in response.iter_content(chunk_size=min(65536, max_response_bytes + 1)):
            raw.extend(chunk)
            if len(raw) > max_response_bytes:
                raise RuntimeError("OPA response exceeds the signed decoded-byte cap")
    body = json.loads(bytes(raw), object_pairs_hook=_unique_object)
    # OPA adds decision_id only when decision logging is enabled; allow nothing else.
    if (not isinstance(body, dict) or "result" not in body
            or set(body) - {"result", "decision_id"} or type(body["result"]) is not bool):
        raise RuntimeError("OPA decision schema differs")
    decision_id = body.get("decision_id")
    if "decision_id" in body and (not isinstance(decision_id, str) or not decision_id):
        raise RuntimeError("OPA decision_id must be a non-empty string when present")
    if body["result"] is not True:
        denied = PermissionError("data use denied for the requested stage and purpose")
        denied.decision_id = decision_id
        raise denied
    return decision_id

Call the hook at every training, fine-tuning, evaluation, indexing, inference-context, logging, memory-write or retraining ingress actually present. Read the pinned endpoint and bounded timeout/response settings from trusted deployment configuration. Verify the effective OPA bundle identity/digest and current registry tags independently. Network, TLS, redirect, parse or policy failures must abort the data operation; never catch an exception and continue. When OPA decision logging is enabled, the Data API adds a decision_id string to the response; the hook accepts only that optional field, returns it on allow and attaches it to the denial, so the AID-H-029.002-G002 audit event can reference the matching OPA decision-log entry. Metrics, provenance and other response fields appear only when requested through query parameters, which the endpoint check forbids. Do not cache an allow decision across authorization, tag or asset changes. Re-evaluate immediately before each admission; ongoing withdrawal propagation remains with AID-H-029.003.

Verify the complete ingress population

Independently replay and observe every eligible ingress, including allowed and wrong-purpose requests, purpose grants for a different stage, unsupported stages, withdrawal, expiry exactly at the boundary, expired RAG consent, missing or malformed expiry, missing tags, changed asset bindings and PDP outages. Verify that denied data was never read or copied and that required decision audit records exist. Fixture replay may override OPA time for determinism; production uses the PDP clock. A policy result alone does not prove that every data-plane route enforces it.

AID-H-029.002-G002

Emit structured audit events for both allow and deny decisions so stage-boundary enforcement becomes explainable and reviewable.
Concept:

Every decision should become an auditable event that says which asset, which stage, which policy version, and which rule path produced the result.

Example audit event
{
  "timestamp": "2026-03-17T12:00:00Z",
  "asset_id": "asset-123",
  "target_stage": "training",
  "decision": "deny",
  "policy_version": "2026-03-17.1",
  "requester_id": "svc-train-pipeline",
  "reason": "asset not authorized for training"
}

Operational notes: Send deny events to the SIEM immediately. Keep allow events too, because they are the evidence that a stage admitted data for a legitimate reason.

Production implementation

At the PDP/enforcement adapter before returning the decision, emit a closed redacted event binding actor, asset/version, source/destination stage, tag/policy digest, decision and reason; write it under an external audit identity and fail closed when the required record cannot be persisted and read back.

Independent verification

A separately credentialed verifier independently reads back or replays decision-to-event one-to-one reconciliation, signature, sink sequence and redaction with allow, deny, PDP error, duplicate, missing-event and sink-outage fixtures, runs positive, negative, boundary, stale-policy, partial-population and dependency-failure fixtures, and reconciles the complete eligible population without trusting a component-supplied verdict.

AID-H-029.002-G003

Require a purpose-bound memory-write grant and keep ordinary facts or preferences from modifying behavioral and source-trust policy.
Step 1: Distinguish evidence eligibility from write authority

At the memory_write stage, AND the existing data-use stage decision with the M-002.004 eligibility result, normal subject/object access checks, and a grant for this exact candidate. Derive the operation class from the storage route and schema. Facts and typed preferences use bounded data namespaces; instructions that alter source ranking, trust status, future behavior, or tool permissions must not be admitted through the generic memory API. Route such changes to the separately authorized policy/configuration release process. A writer cannot choose “fact” in a request to bypass this boundary.

Step 2: Enforce the separate authority check

Resolve provenance and grants from authenticated immutable stores and verify signatures/version binding before calling this predicate. A client may provide only their record references. Use positive policy-bound expiry limits, current revocation readback, and the existing atomic consume/write transaction so replays cannot reuse a one-time grant. Missing records deny. The record types and number fields must pass strict schema validation; an unavailable registry is an error, never a default allow.

# memory_authority.py -- called with server-resolved records, never request verdicts
import hashlib

def memory_write_allowed(payload: bytes, candidate: dict, provenance: dict,
                         grant: dict, policy: dict, now: int) -> bool:
    try:
        digest = hashlib.sha256(payload).hexdigest()
        # operation_class is chosen by the storage route and schema, not the LLM.
        kind = candidate["operation_class"]
        if kind not in {"fact", "preference"}:
            return False  # trust/behavior/tool-policy changes use an admin release path
        if candidate["namespace"] != policy["memory_namespaces"][kind]:
            return False
        if grant["stage"] != "memory_write" or grant["operation_class"] != kind:
            return False
        if grant["candidate_sha256"] != digest or provenance["candidate_sha256"] != digest:
            return False
        for key in ("tenant_id", "subject_id", "task_id"):
            if not candidate[key] or candidate[key] != grant[key] or candidate[key] != provenance[key]:
                return False
        if grant["policy_version"] != policy["version"] or grant["revoked"] is not False:
            return False
        if not grant["issued_at"] <= now < grant["expires_at"]:
            return False
        if grant["purpose"] != policy["memory_purposes"][kind]:
            return False
        if grant["grant_kind"] not in {"explicit_memory_confirmation", "approved_memory_workflow"}:
            return False
        # A link-prefill submit receipt cannot be substituted for this grant.
        return grant["namespace"] == candidate["namespace"]
    except (KeyError, TypeError, ValueError):
        return False
Step 3: Preserve the boundary during retrieval

Store free-text facts as untrusted data with origin and eligibility tags, not as system/developer instructions. Prefer explicit typed preference keys (for example an allowed language enum), and never interpret a remembered domain name as a trusted-source rule. Apply the input gate and instruction/data separation when loading memory; string matching alone cannot reliably classify hidden behavioral instructions. A legitimate “remember my preferred language” action can receive a narrowly bound user confirmation or an approved workflow grant. Clicking an assistant link or confirming its prompt supplies neither. Test the exact same bytes under fact, preference, and trust-policy routes, and prove the latter cannot be written by the generic agent credential. Also test cross-user candidates, stale/revoked grants, and replacement of the candidate after approval.

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-029.002: Lifecycle-Stage Authorization Gate | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service