Scope boundary

Owns dataset-format routing, dangerous-feature denial, the parser-side admission decision, and evidence binding for preview, conversion, sampling, indexing, and RAG-ingestion parsers. It consumes but does not own or re-score isolation strength, sandbox ephemerality, egress denial, detonation, safe remote fetch, model-artifact scanning, generic output sanitization, MCP response handling, or agentic value-to-sink authorization. A bounded dataset preview broker is a conditional implementation method here, not an independently scoreable release control.

AID-H-002.001: AID-H-002.001 sanitizes the records that may influence training or evaluation; AID-H-002.009 admits the dataset container and parser result against hostile format features. A dataset can contain benign records yet exploit its parser, and a parser-safe artifact can still contain poisoned or privacy-sensitive records.

AID-H-002.003: AID-H-002.003 safely decodes and sanitizes complete image, audio, video, document, and OCR artifacts; AID-H-002.009 governs dataset containers, reference descriptors, and the parser result. A dataset whose records include media may require both controls: parser admission for the container and media admission for each extracted artifact.

AID-H-003.009: AID-H-003.009 statically scans models, checkpoints, and adapters for unsafe serialization and embedded code; AID-H-002.009 admits untrusted dataset and structured-data artifacts against parser-specific features. Route each object by its actual format and loader; a model-artifact scan does not make a dataset parser safe, and parser admission does not approve model loading.

AID-H-006.002: AID-H-006.002 owns sink-specific release of model-generated text, markup, URLs, structured payloads, and token streams; AID-H-002.009 owns dataset-parser admission and its conditional bounded-row preview method. For a mixed response, apply each control to its own segment and assign one primary release-gate result to the final bytes rather than double-scoring the same publication.

AID-H-034.005: AID-H-034.005 owns MCP server tool, resource, and elicitation output boundaries. The bounded preview method here applies to an ordinary dataset preview or sampling endpoint, not MCP protocol output.

AID-H-028.004: AID-H-028.004 owns parsing, rendering, and execution surfaces for MCP client responses. The bounded preview method here rebuilds dataset rows before an ordinary preview or sampling API publishes them.

AID-H-018.005: AID-H-018.005 owns value-to-sink capability and data-flow enforcement for agentic actions. This control owns the narrower upstream dataset-parser admission and does not authorize any later agentic sink.

AID-H-019.001: AID-H-019.001 owns URL normalization, destination validation, redirect handling, and SSRF-safe fetch when a remote URL is staged. This control consumes the fetched exact object and digest; it does not re-score remote-fetch safety.

AID-I-001.002: AID-I-001.002 owns and verifies the policy-selected microVM or low-level isolation boundary. This control verifies and binds its receipt before admitting a parse result; it does not create a second isolation PASS.

AID-I-001.003: AID-I-001.003 owns per-run sandbox freshness and teardown. This control consumes its run-bound receipt and fails closed when policy-required evidence is absent or unverifiable.

AID-I-001.004: AID-I-001.004 owns sandbox egress policy and enforcement. This control consumes its receipt; egress denial does not inspect the legitimate parser-result channel.

AID-I-001.005: AID-I-001.005 owns safe detonation and raw observation integrity. Parser admission may consume its evidence for hostile-format tests but does not re-score detonation.

Warning

Profile-dependent compatibility and bounded-preview residual disclosure

Format lockdown can reject legitimate datasets that rely on external storage, virtual layouts, or references, and stronger isolation can add latency or cost. Select bounds and profiles from versioned organizational policy and measure the actual platform. The optional preview broker blocks raw, oversized, metadata-bearing, and out-of-schema publication, but it cannot distinguish a legitimate short value from a short schema-valid secret and is not a confidentiality guarantee. Keep the parser secretless and fail closed rather than silently falling back to an unsafe route.

Implementation guidance

• AID-H-002.009-G001: Pin the accepted parser routes and structurally reject dangerous dataset-format features before admitting a parser result.
• AID-H-002.009-G002: Use a version-and-digest-bound input handoff and parse the same opened descriptor in a secretless, no-network worker.
• AID-H-002.009-G003: Do not evaluate artifact-controlled templates or invoke code-bearing loaders on the dataset-parser path.
• AID-H-002.009-G004: Independently verify parser admission over the complete policy-required benign, negative, and regression case population without re-scoring consumed isolation controls.
• AID-H-002.009-G005: For bounded preview or sampling endpoints, use a small independent broker to validate, rebuild, and publish the exact policy-bounded parser result, then run specific negative tests without creating a second control outcome.

AID-H-002.009-G001

Step 1: Pin the route

Use a versioned policy that binds verified content type to the parser image, parser and native-library versions, allowed feature set, and maximum link and object populations. File extensions are hints only. Unsupported type, malformed policy, or policy drift is a deny. Authenticate the policy where the deployment's assurance profile requires it.

Step 2: Inspect HDF5 inside the confined worker

Opening HDF5 invokes native code, so this inspection is not a pre-sandbox safety check. With h5py 3.11 or later, use visititems_links(); ordinary visit() and visititems() do not enumerate soft or external links. Reject Dataset.external because it is external raw storage, and reject virtual datasets and soft or external links on the default route. Supply max_links and max_objects from the versioned policy so a compact hostile file cannot force unbounded traversal. The callbacks return a stop value and raise only after h5py unwinds, avoiding exception corruption across the native visitor boundary.

import h5py


def _positive_integer(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{field} must be a positive integer")
    return value


def enforce_hdf5_features(
    file_object,
    max_links: int,
    max_objects: int,
) -> None:
    max_links = _positive_integer(max_links, "max_links")
    max_objects = _positive_integer(max_objects, "max_objects")
    with h5py.File(file_object, 'r') as h5:
        link_count = 0
        link_problem = None

        def check_link(name, link):
            nonlocal link_count, link_problem
            link_count += 1
            if link_count > max_links:
                link_problem = "HDF5 link population exceeds policy"
                return True
            if isinstance(link, (h5py.ExternalLink, h5py.SoftLink)):
                link_problem = f"soft or external link denied: {name}"
                return True
            return None

        h5.visititems_links(check_link)
        if link_problem:
            raise PermissionError(link_problem)

        object_count = 0
        object_problem = None

        def check_object(name, obj):
            nonlocal object_count, object_problem
            object_count += 1
            if object_count > max_objects:
                object_problem = "HDF5 object population exceeds policy"
                return True
            if isinstance(obj, h5py.Dataset):
                if obj.external:
                    object_problem = f"external raw storage denied: {name}"
                    return True
                if obj.is_virtual:
                    object_problem = f"virtual dataset denied: {name}"
                    return True
            return None

        h5.visititems(check_object)
        if object_problem:
            raise PermissionError(object_problem)

Step 3: Deny reference semantics by data shape

For the ReferenceFileSystem JSON form, reject template or generator fields and every non-string reference value; do not search only for URL syntax. Strings are the inline form, while list-like values represent an external target even when relative, scheme-less, Windows-shaped, or null-defaulted. Parse bounded JSON without instantiating fsspec or Jinja.

AID-H-002.009-G002

Required properties

Bind the run to an exact object version/generation and expected SHA-256. Deny overwrite/delete/delete-version to the producer after handoff, or fail closed if the exact version disappears. The consumer fetches that exact version, re-hashes it, and opens one local descriptor with no symlink following. Hash and parse that same descriptor; compare fstat before and after. Give the worker no cloud credentials, service-account token, cross-tenant data, internal service access, or IMDS. Apply the isolation and resource profile selected by policy and verify the receipts owned by AID-I-001.*.

import contextlib, hashlib, os, stat

@contextlib.contextmanager
def open_verified_input(path: str, expected_sha256: str, max_bytes: int):
    if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes < 1:
        raise ValueError('max_bytes must be a positive integer')
    if not hasattr(os, 'O_NOFOLLOW'):
        raise OSError('O_NOFOLLOW unavailable; use an equivalent no-rebind staging primitive')
    flags = os.O_RDONLY | os.O_NOFOLLOW | getattr(os, 'O_CLOEXEC', 0)
    fd = os.open(path, flags)
    try:
        before = os.fstat(fd)
        if not stat.S_ISREG(before.st_mode) or before.st_size > max_bytes:
            raise PermissionError('input is not an allowed regular file')
        with os.fdopen(os.dup(fd), 'rb', closefd=True) as hashing_stream:
            digest_state = hashlib.sha256()
            for chunk in iter(lambda: hashing_stream.read(1024 * 1024), b''):
                digest_state.update(chunk)
            digest = digest_state.hexdigest()
        if digest != expected_sha256:
            raise PermissionError('input digest mismatch')
        os.lseek(fd, 0, os.SEEK_SET)
        with os.fdopen(os.dup(fd), 'rb', closefd=True) as parsing_stream:
            yield parsing_stream
        after = os.fstat(fd)
        identity = lambda s: (s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns, s.st_ctime_ns)
        if identity(before) != identity(after):
            raise PermissionError('input changed during parsing')
    finally:
        os.close(fd)

The example fails closed where O_NOFOLLOW is unavailable. On Windows or another such platform, stage through a platform primitive that supplies equivalent no-rebind identity and verify it independently; never silently omit the protection.

AID-H-002.009-G003

Default path

Read a bounded version-1 reference descriptor as raw JSON, reject duplicate keys, non-finite numbers, template/generator fields, and external-reference value shapes, and never construct ReferenceFileSystem from untrusted input. Keep the structural rejection even when using fsspec 2026.6.0 or later and pin Jinja2 3.1.6 or later wherever Jinja exists elsewhere in the image. SandboxedEnvironment is defense-in-depth, not permission to evaluate untrusted artifact templates.

import json

_TEMPLATE_MARKERS = ('{{', '}}', '{%', '%}')

def _pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise PermissionError(f'duplicate key: {key}')
        result[key] = value
    return result

def _bad_constant(value):
    raise PermissionError(f'non-finite number: {value}')

def load_inline_refs(raw: bytes, max_bytes: int, max_members: int) -> dict[str, str]:
    if len(raw) > max_bytes:
        raise PermissionError('descriptor byte limit exceeded')
    document = json.loads(raw.decode('utf-8', 'strict'), object_pairs_hook=_pairs, parse_constant=_bad_constant)
    if not isinstance(document, dict) or set(document) != {'version', 'refs'}:
        raise PermissionError('only explicit version and refs fields are allowed')
    if document['version'] != 1 or not isinstance(document['refs'], dict):
        raise PermissionError('unsupported reference descriptor version')
    refs = document['refs']
    if len(refs) > max_members:
        raise PermissionError('reference member limit exceeded')
    for key, value in refs.items():
        if not isinstance(key, str) or not key or not isinstance(value, str):
            raise PermissionError('external or malformed reference denied')
        if any(marker in key or marker in value for marker in _TEMPLATE_MARKERS):
            raise PermissionError('template marker denied')
    return refs

Routing

The caller consumes the returned strings as inline data with its own bounded decoder; it does not pass the document to fsspec. Dataset artifacts that contain pickle/opcode/model-loader semantics are denied on this route and sent to AID-H-003.009. An installation that truly needs external dataset references requires a separately reviewed path-confinement/fetch branch; it must not enable a generic allow_external switch on this default path.

AID-H-002.009-G004

Evidence contract

Bind the versioned case manifest and externally observed results to run ID, artifact version and digest, deployed parser-image digest, parser and native-library versions, policy digest, and exact parser-result digest. Use signatures only where the deployment's assurance profile requires independently authenticated receipts. Keep case_execution_state, parser_outcome (ADMIT, DENY, CRASH, OOM, or TIMEOUT), and parse_decision separate. Never accept worker-supplied isolation_ok, egress_blocked, or torn_down booleans as proof; validate policy-required dependency receipts at their owning controls.

Evaluate the complete case population

Compare every observed case with the manifest's exact expected parser decision and required dependency receipts. Any wrong decision, heap-canary disclosure, forbidden-feature admission, missing case, stale result, or unverifiable evidence prevents a successful control result. Emit one outcome under AID-H-002.009; this verification method supplies evidence and never creates a separately scoreable guidance result.

A structurally forbidden payload that crashes or exhausts resources instead of being denied is FAIL. A heap-over-read case is a disclosure regression, not an expected crash. Synthetic fault injection proves only fail-closed orchestration when explicitly manifested; it cannot replace benign or exploit-regression cases. An instrumented image supplements but does not substitute for the production-equivalent image.

AID-H-002.009-G005

Applicability and score boundary

Use this method only when the parser feeds a bounded row-preview, first-rows, or sampling endpoint. Bulk-only conversion or indexing can omit it without changing this control's applicability or parser-admission result. These checks are supporting evidence for AID-H-002.009; G005 is not independently selectable and produces no second PASS. In a mixed preview response, any model-generated summary remains owned by AID-H-006.002; this method owns only the bounded row-preview bytes.

Baseline handoff

The parser emits untrusted candidate JSON through a bounded, unidirectional, orchestrator-owned pipe or spool and holds no publication credential. If persisted, the orchestrator creates a new exact object version under a separate identity and records its SHA-256. The small broker fetches that exact version, recomputes its digest, verifies the exact versioned policy schema, applies a byte cap before decoding, and contains no ML or native artifact parser, template engine, URL fetcher, or broad credential. It publishes the exact returned buffer without another transformation. Thresholds come from versioned organizational policy.

from __future__ import annotations

import hashlib
import hmac
import json
import math


_POLICY_FIELDS = {
    "policy_schema_version",
    "candidate_schema_version",
    "columns",
    "max_columns",
    "max_rows",
    "max_candidate_bytes",
    "max_cell_bytes",
    "max_decoded_string_bytes",
    "max_publication_bytes",
    "max_abs_integer",
}


def _positive_integer(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{field} must be a positive integer")
    return value


def _pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise PermissionError(f"duplicate key: {key}")
        result[key] = value
    return result


def _bad_constant(value):
    raise PermissionError(f"non-finite number: {value}")


def _validated_policy(policy: dict) -> tuple[str, tuple[str, ...], dict[str, int]]:
    if not isinstance(policy, dict) or set(policy) != _POLICY_FIELDS:
        raise ValueError("publication policy schema differs")
    if policy["policy_schema_version"] != "aidefend.dataset-preview-policy.v1":
        raise ValueError("publication policy version is unsupported")
    candidate_schema = policy["candidate_schema_version"]
    if not isinstance(candidate_schema, str) or not candidate_schema:
        raise ValueError("candidate_schema_version must be a non-empty string")

    columns_value = policy["columns"]
    if (
        not isinstance(columns_value, list)
        or not columns_value
        or any(not isinstance(item, str) or not item for item in columns_value)
        or len(columns_value) != len(set(columns_value))
    ):
        raise ValueError("columns must be unique non-empty strings")
    columns = tuple(columns_value)

    limits = {
        field: _positive_integer(policy[field], field)
        for field in _POLICY_FIELDS
        if field.startswith("max_")
    }
    if len(columns) > limits["max_columns"]:
        raise ValueError("column population exceeds max_columns")
    return candidate_schema, columns, limits


def validate_and_rebuild(
    raw: bytes,
    expected_candidate_sha256: str,
    policy: dict,
) -> tuple[bytes, str]:
    if not isinstance(raw, bytes):
        raise TypeError("candidate must be bytes")
    candidate_schema, columns, limits = _validated_policy(policy)
    if len(raw) > limits["max_candidate_bytes"]:
        raise PermissionError("candidate byte limit exceeded")
    actual_candidate_sha256 = hashlib.sha256(raw).hexdigest()
    if (
        not isinstance(expected_candidate_sha256, str)
        or len(expected_candidate_sha256) != 64
        or not hmac.compare_digest(actual_candidate_sha256, expected_candidate_sha256)
    ):
        raise PermissionError("candidate digest mismatch")

    try:
        document = json.loads(
            raw.decode("utf-8", "strict"),
            object_pairs_hook=_pairs,
            parse_constant=_bad_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise PermissionError("candidate is not strict UTF-8 JSON") from exc
    if not isinstance(document, dict) or set(document) != {"schema_version", "rows"}:
        raise PermissionError("unexpected top-level shape")
    if document["schema_version"] != candidate_schema:
        raise PermissionError("candidate schema version differs")
    rows = document["rows"]
    if not isinstance(rows, list) or len(rows) > limits["max_rows"]:
        raise PermissionError("row population exceeds policy")

    decoded_string_bytes = 0
    rebuilt = []
    expected_columns = set(columns)
    for row in rows:
        if not isinstance(row, dict) or set(row) != expected_columns:
            raise PermissionError("column set mismatch")
        clean = {}
        for name in columns:
            value = row[name]
            if isinstance(value, (list, dict)):
                raise PermissionError("nested value denied")
            if value is not None and not isinstance(value, (str, int, float, bool)):
                raise PermissionError("non-primitive value denied")
            if isinstance(value, float) and not math.isfinite(value):
                raise PermissionError("non-finite float denied")
            if (
                isinstance(value, int)
                and not isinstance(value, bool)
                and abs(value) > limits["max_abs_integer"]
            ):
                raise PermissionError("integer outside policy domain")
            encoded_cell = json.dumps(
                value,
                ensure_ascii=False,
                allow_nan=False,
                separators=(",", ":"),
            ).encode("utf-8")
            if len(encoded_cell) > limits["max_cell_bytes"]:
                raise PermissionError("cell byte limit exceeded")
            if isinstance(value, str):
                decoded_string_bytes += len(value.encode("utf-8"))
                if decoded_string_bytes > limits["max_decoded_string_bytes"]:
                    raise PermissionError("decoded string limit exceeded")
            clean[name] = value
        rebuilt.append(clean)

    published = json.dumps(
        {"schema_version": candidate_schema, "rows": rebuilt},
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    if len(published) > limits["max_publication_bytes"]:
        raise PermissionError("publication byte limit exceeded")
    return published, hashlib.sha256(published).hexdigest()


def verify_observed_publication(
    approved_bytes: bytes,
    approved_sha256: str,
    observed_bytes: bytes,
) -> None:
    if not isinstance(approved_bytes, bytes) or not isinstance(observed_bytes, bytes):
        raise TypeError("publication observations must be bytes")
    approved_actual = hashlib.sha256(approved_bytes).hexdigest()
    observed_actual = hashlib.sha256(observed_bytes).hexdigest()
    if (
        not hmac.compare_digest(approved_actual, approved_sha256)
        or not hmac.compare_digest(observed_actual, approved_sha256)
        or not hmac.compare_digest(approved_bytes, observed_bytes)
    ):
        raise PermissionError("observed publication differs from approved bytes")

Executable negative tests

Run these tests in CI against the extracted broker and repeat equivalent cases through the deployed endpoint. They cover exact reconstruction and observation, digest mismatch, malformed policy, duplicate keys, extra or nested fields, non-finite values, and row, column, cell, decoded, integer, candidate, and publication limits. The final test deliberately documents the residual risk: a short schema-valid secret can still pass.

from __future__ import annotations

import hashlib
import json
import unittest

from preview_broker import validate_and_rebuild, verify_observed_publication


def encoded(value: dict) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, allow_nan=False, separators=(",", ":")
    ).encode("utf-8")


class PreviewBrokerTests(unittest.TestCase):
    def setUp(self):
        self.policy = {
            "policy_schema_version": "aidefend.dataset-preview-policy.v1",
            "candidate_schema_version": "preview.v1",
            "columns": ["id", "value"],
            "max_columns": 2,
            "max_rows": 2,
            "max_candidate_bytes": 512,
            "max_cell_bytes": 24,
            "max_decoded_string_bytes": 30,
            "max_publication_bytes": 512,
            "max_abs_integer": 1000,
        }

    def check(self, raw: bytes):
        return validate_and_rebuild(
            raw, hashlib.sha256(raw).hexdigest(), self.policy
        )

    def assert_denied(self, raw: bytes):
        with self.assertRaises((PermissionError, ValueError, TypeError)):
            self.check(raw)

    def test_canonical_exact_bytes_digest_and_observation(self):
        raw = encoded({
            "rows": [{"value": "ok", "id": 1}],
            "schema_version": "preview.v1",
        })
        published, digest = self.check(raw)
        expected = (
            b'{"rows":[{"id":1,"value":"ok"}],'
            b'"schema_version":"preview.v1"}'
        )
        self.assertEqual(published, expected)
        self.assertEqual(hashlib.sha256(published).hexdigest(), digest)
        verify_observed_publication(published, digest, expected)

    def test_digest_and_observation_mismatch_are_denied(self):
        raw = encoded({"schema_version": "preview.v1", "rows": []})
        with self.assertRaises(PermissionError):
            validate_and_rebuild(raw, "0" * 64, self.policy)
        published, digest = self.check(raw)
        with self.assertRaises(PermissionError):
            verify_observed_publication(published, digest, published + b" ")

    def test_duplicate_extra_nested_and_nonfinite_values_are_denied(self):
        self.assert_denied(
            b'{"schema_version":"preview.v1","rows":[],"rows":[]}'
        )
        self.assert_denied(encoded({
            "schema_version": "preview.v1",
            "rows": [{"id": 1, "value": "ok", "metadata": "x"}],
        }))
        self.assert_denied(encoded({
            "schema_version": "preview.v1",
            "rows": [{"id": 1, "value": {"secret": "x"}}],
        }))
        self.assert_denied(
            b'{"schema_version":"preview.v1",'
            b'"rows":[{"id":1,"value":NaN}]}'
        )

    def test_all_volume_and_numeric_limits_are_enforced(self):
        cases = [
            {"schema_version": "preview.v1", "rows": [
                {"id": 1, "value": "a"},
                {"id": 2, "value": "b"},
                {"id": 3, "value": "c"},
            ]},
            {"schema_version": "preview.v1", "rows": [
                {"id": 1, "value": "x" * 24},
            ]},
            {"schema_version": "preview.v1", "rows": [
                {"id": 1001, "value": "x"},
            ]},
        ]
        for case in cases:
            with self.subTest(case=case):
                self.assert_denied(encoded(case))

        raw = encoded({
            "schema_version": "preview.v1",
            "rows": [{"id": 1, "value": "ab"}, {"id": 2, "value": "cd"}],
        })
        for field in (
            "max_candidate_bytes",
            "max_decoded_string_bytes",
            "max_publication_bytes",
        ):
            policy = dict(self.policy)
            policy[field] = 3
            with self.subTest(field=field), self.assertRaises(PermissionError):
                validate_and_rebuild(
                    raw, hashlib.sha256(raw).hexdigest(), policy
                )

    def test_malformed_policy_is_denied(self):
        raw = encoded({"schema_version": "preview.v1", "rows": []})
        policies = [
            dict(self.policy, columns=["id", "id"]),
            dict(self.policy, max_rows=True),
            {key: value for key, value in self.policy.items() if key != "max_rows"},
            dict(self.policy, unexpected=True),
        ]
        for policy in policies:
            with self.subTest(policy=policy), self.assertRaises(ValueError):
                validate_and_rebuild(
                    raw, hashlib.sha256(raw).hexdigest(), policy
                )

    def test_short_schema_valid_secret_remains_possible(self):
        raw = encoded({
            "schema_version": "preview.v1",
            "rows": [{"id": 1, "value": "KEY-7F2A"}],
        })
        published, _ = self.check(raw)
        self.assertIn(b"KEY-7F2A", published)


if __name__ == "__main__":
    unittest.main()

Operational verification: independently capture the actual response bytes and pass them to verify_observed_publication. Measure latency, throughput, error rate, and cost against the deployment's versioned SLO. A missing observation, digest mismatch, unavailable broker, malformed policy, or verification error blocks the bounded response; it never authorizes fallback to raw parser output.

Open-source tools

• h5py (BSD-3-Clause; confined HDF5 feature inspection, pinned with the HDF5 runtime)
• python-magic (MIT; content-type routing backed by a separately maintained libmagic installation)
• Magika (Apache-2.0; bounded content-type routing signal only; structural parser admission remains authoritative)
• Atheris (Apache-2.0; coverage-guided fuzzing of Python admission code and instrumented native parser extensions)
• ClusterFuzzLite (Apache-2.0; CI fuzzing and sanitizer-backed parser regression testing)

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise
• AML.T0010.002 AI Supply Chain Compromise: Data
• AML.T0011 User Execution
• AML.T0011.000 User Execution: Unsafe AI Artifacts
• AML.T0025 Exfiltration via Cyber Means (the conditional bounded-preview method limits the parser-result response channel)
• AML.T0037 Data from Local System (hostile dataset features can cause a parser to disclose local files)
• AML.T0112 Machine Compromise
• AML.T0112.001 Machine Compromise: AI Artifacts
• AML.T0119 Exploit Automated Artifact Processing Pipeline

MAESTRO

• Compromised RAG Pipelines (L2) (when the parser is part of a RAG ingestion or indexing workflow)
• Input Validation Attacks (L3)
• Data Leakage (Cross-Layer) (the conditional bounded-preview method limits disclosure through the parser-result channel)

OWASP LLM Top 10 2026

• LLM02:2026 Sensitive Information Disclosure (when a bounded dataset response is exposed through an LLM-integrated pipeline)
• LLM04:2026 Supply Chain

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities (when an agentic workflow consumes the hostile dataset artifact)
• ASI05:2026 Unexpected Code Execution (RCE)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-13.1.2 Memory Flooding
• AISubtech-13.1.4 Application Denial of Service
• AITech-13.1 Disruption of Availability
• AITech-14.1 Unauthorized Access
• AITech-8.2 Data Exfiltration / Exposure
• AITech-8.3 Information Disclosure

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (an automatically invoked dataset parser is an integrated component exposed to hostile artifacts)

Databricks AI Security Framework 3.0

• Platform 12.7: Initial Access (a hostile dataset artifact can exploit an automatically invoked platform parser)