Scope boundary

Owns safe, bounded, non-executing parsing and normalization of composite skill files before deeper admission stages. It specializes generic deserialization and code-analysis safeguards for skill loaders without deciding final admission.

AID-H-021: AID-H-030.004 hardens parsing of candidate skill artifacts in the admission pipeline; AID-H-021 protects active agent configuration artifacts and control-plane configuration. A safe skill parser does not establish runtime configuration integrity.

AID-H-025.001: AID-H-030.004 safely parses and bounds skill package structure; AID-H-025.001 scans executable text and code for prohibited constructs. Loader safety must precede deeper static analysis because parsing itself is attackable.

AID-H-018.009: AID-H-030.004 safely parses and normalizes candidate skill packages before deeper admission; AID-H-018.009 confines model-callable runtime and reusable filesystem operations. Package parsing and runtime filesystem confinement retain separate evidence and outcomes.

Implementation guidance

• AID-H-030.004-G001: Parse skill manifests and frontmatter with safe parser defaults, explicit schemas, and bounded decoding rules inside an isolated loader stage.

AID-H-030.004-G001

Concept:

Normalize a skill package as hostile input before any dynamic import, hook, dependency install, or semantic analysis. The reference path accepts an immutable ZIP snapshot and releases only digest-bound canonical data. A signature-verified loader profile supplies exact member allowlists, schemas, and resource bounds; unknown members are rejected, not silently ignored. Run this module inside the isolated loader container or microVM, with no credentials, network, host mounts, or package-manager access.

Bounded archive and manifest loader

# File: skill_admission/safe_loader.py
from __future__ import annotations

import hashlib
import io
import json
import math
import os
import stat
import unicodedata
import zipfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Protocol

import jsonschema
import yaml
from yaml.events import AliasEvent


@dataclass(frozen=True)
class LoaderPolicy:
    revision: str
    allowed_files: tuple[str, ...]
    required_files: tuple[str, ...]
    manifest_file: str
    skill_markdown_file: str
    manifest_schema: dict
    maximum_archive_bytes: int
    maximum_members: int
    maximum_path_depth: int
    maximum_file_bytes: int
    maximum_total_uncompressed_bytes: int
    maximum_compression_ratio: float
    maximum_object_depth: int
    maximum_object_nodes: int
    allowed_zip_compressions: tuple[int, ...]


@dataclass(frozen=True)
class MemberDigest:
    path: str
    size: int
    sha256: str


@dataclass(frozen=True)
class NormalizedSkill:
    artifact_sha256: str
    policy_revision: str
    members: tuple[MemberDigest, ...]
    manifest_canonical_json: bytes
    frontmatter_canonical_json: bytes


class VerifiedLoaderPolicyProvider(Protocol):
    def load_verified(self) -> LoaderPolicy:
        """Verify policy signature, schema, validity window, and revision."""
        raise RuntimeError("protocol-only method")


class _UniqueBoundedSafeLoader(yaml.SafeLoader):
    def __init__(self, stream: str, maximum_nodes: int) -> None:
        super().__init__(stream)
        self.maximum_nodes = maximum_nodes
        self.composed_nodes = 0

    def compose_node(self, parent, index):
        if self.check_event(AliasEvent):
            raise yaml.YAMLError("YAML aliases are disabled")
        self.composed_nodes += 1
        if self.composed_nodes > self.maximum_nodes:
            raise yaml.YAMLError("YAML node cap exceeded")
        return super().compose_node(parent, index)


def _construct_unique_mapping(loader, node, deep=False):
    loader.flatten_mapping(node)
    result = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        if not isinstance(key, str):
            raise yaml.YAMLError("mapping keys must be strings")
        if key in result:
            raise yaml.YAMLError(f"duplicate YAML key: {key}")
        result[key] = loader.construct_object(value_node, deep=deep)
    return result


_UniqueBoundedSafeLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    _construct_unique_mapping,
)


def _read_regular_snapshot(path_text: str, maximum_bytes: int) -> bytes:
    path = Path(path_text)
    before = path.lstat()
    if (
        stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode)
        or before.st_size > maximum_bytes
    ):
        raise RuntimeError("skill package is not a bounded regular file")
    flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags)
    try:
        opened = os.fstat(descriptor)
        if (
            opened.st_dev != before.st_dev or opened.st_ino != before.st_ino
            or opened.st_size != before.st_size
            or opened.st_mtime_ns != before.st_mtime_ns
        ):
            raise RuntimeError("skill package changed before read")
        chunks = []
        total = 0
        while True:
            block = os.read(descriptor, min(1024 * 1024, maximum_bytes + 1 - total))
            if not block:
                break
            chunks.append(block)
            total += len(block)
            if total > maximum_bytes:
                raise RuntimeError("skill package exceeds archive cap")
        after = os.fstat(descriptor)
        if (
            after.st_size != opened.st_size
            or after.st_mtime_ns != opened.st_mtime_ns
        ):
            raise RuntimeError("skill package changed during read")
    finally:
        os.close(descriptor)
    return b"".join(chunks)


def _canonical_member_name(raw: str, maximum_depth: int) -> str:
    if (
        not raw or "\\" in raw or "\x00" in raw
        or unicodedata.normalize("NFC", raw) != raw
    ):
        raise ValueError("noncanonical ZIP member name")
    trimmed = raw[:-1] if raw.endswith("/") else raw
    path = PurePosixPath(trimmed)
    canonical = str(path)
    if (
        not trimmed or path.is_absolute() or canonical != trimmed
        or any(part in {"", ".", ".."} for part in path.parts)
        or len(path.parts) > maximum_depth
    ):
        raise ValueError("unsafe ZIP member path")
    return canonical


def _strict_json(raw: bytes) -> object:
    def unique(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

    return json.loads(raw.decode("utf-8"), object_pairs_hook=unique)


def _safe_yaml(text: str, maximum_nodes: int) -> object:
    loader = _UniqueBoundedSafeLoader(text, maximum_nodes)
    try:
        return loader.get_single_data()
    finally:
        loader.dispose()


def _frontmatter(markdown: bytes, maximum_nodes: int) -> object:
    text = markdown.decode("utf-8")
    if text.startswith("\ufeff"):
        raise ValueError("UTF-8 BOM is not accepted")
    lines = text.splitlines()
    if not lines or lines[0] != "---":
        return {}
    try:
        closing = lines[1:].index("---") + 1
    except ValueError as error:
        raise ValueError("unterminated YAML frontmatter") from error
    return _safe_yaml("\n".join(lines[1:closing]), maximum_nodes) or {}


def _validate_shape(
    value: object, maximum_depth: int, maximum_nodes: int
) -> None:
    remaining = [maximum_nodes]

    def visit(item: object, depth: int) -> None:
        remaining[0] -= 1
        if remaining[0] < 0 or depth > maximum_depth:
            raise ValueError("decoded object exceeds policy shape bounds")
        if item is None or isinstance(item, (bool, str, int)):
            return
        if isinstance(item, float):
            if not math.isfinite(item):
                raise ValueError("non-finite numeric value")
            return
        if isinstance(item, list):
            for child in item:
                visit(child, depth + 1)
            return
        if isinstance(item, dict):
            for key, child in item.items():
                if not isinstance(key, str):
                    raise ValueError("object keys must be strings")
                visit(child, depth + 1)
            return
        raise ValueError(f"unsupported decoded type: {type(item).__name__}")

    visit(value, 0)


def _canonical_json(value: object) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")


def _reject_external_schema_references(value: object) -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            if key in {"$ref", "$dynamicRef"}:
                if not isinstance(child, str) or not child.startswith("#"):
                    raise ValueError("external JSON Schema references are disabled")
            _reject_external_schema_references(child)
    elif isinstance(value, list):
        for child in value:
            _reject_external_schema_references(child)


def normalize_skill_zip(
    package_path: str, policy_provider: VerifiedLoaderPolicyProvider
) -> NormalizedSkill:
    policy = policy_provider.load_verified()
    positive_bounds = (
        policy.maximum_archive_bytes, policy.maximum_members,
        policy.maximum_path_depth, policy.maximum_file_bytes,
        policy.maximum_total_uncompressed_bytes,
        policy.maximum_object_depth, policy.maximum_object_nodes,
    )
    if (
        not policy.revision or not policy.allowed_files
        or any(value <= 0 for value in positive_bounds)
        or not math.isfinite(policy.maximum_compression_ratio)
        or policy.maximum_compression_ratio < 1
        or not policy.allowed_zip_compressions
        or len(policy.allowed_files) != len(set(policy.allowed_files))
        or not set(policy.required_files).issubset(policy.allowed_files)
        or policy.manifest_file not in policy.required_files
        or policy.skill_markdown_file not in policy.required_files
    ):
        raise RuntimeError("invalid verified loader policy")

    artifact = _read_regular_snapshot(
        package_path, policy.maximum_archive_bytes
    )
    artifact_sha256 = hashlib.sha256(artifact).hexdigest()
    payloads: dict[str, bytes] = {}
    digests: list[MemberDigest] = []
    seen_casefold: set[str] = set()
    total_uncompressed = 0

    with zipfile.ZipFile(io.BytesIO(artifact), "r") as archive:
        infos = archive.infolist()
        if not infos or len(infos) > policy.maximum_members:
            raise RuntimeError("invalid ZIP member population")
        for info in infos:
            name = _canonical_member_name(
                info.filename, policy.maximum_path_depth
            )
            folded = name.casefold()
            if folded in seen_casefold:
                raise RuntimeError("duplicate or case-colliding ZIP member")
            seen_casefold.add(folded)
            if info.flag_bits & 0x1:
                raise RuntimeError("encrypted ZIP members are not accepted")
            unix_mode = (info.external_attr >> 16) & 0xFFFF
            file_type = stat.S_IFMT(unix_mode)
            if info.is_dir():
                if file_type not in {0, stat.S_IFDIR}:
                    raise RuntimeError("ambiguous ZIP directory type")
                continue
            if file_type not in {0, stat.S_IFREG}:
                raise RuntimeError("link, directory, or special ZIP member rejected")
            if name not in set(policy.allowed_files):
                raise RuntimeError(f"unapproved ZIP member: {name}")
            if info.compress_type not in set(policy.allowed_zip_compressions):
                raise RuntimeError("unapproved ZIP compression method")
            ratio = info.file_size / max(info.compress_size, 1)
            total_uncompressed += info.file_size
            if (
                info.file_size > policy.maximum_file_bytes
                or total_uncompressed > policy.maximum_total_uncompressed_bytes
                or ratio > policy.maximum_compression_ratio
            ):
                raise RuntimeError("ZIP resource bound exceeded")
            with archive.open(info, "r") as member:
                raw = member.read(policy.maximum_file_bytes + 1)
                if len(raw) != info.file_size:
                    raise RuntimeError("ZIP member size mismatch")
            payloads[name] = raw
            digests.append(MemberDigest(
                path=name,
                size=len(raw),
                sha256=hashlib.sha256(raw).hexdigest(),
            ))

    if set(policy.required_files) - set(payloads):
        raise RuntimeError("required skill member missing")
    manifest = _strict_json(payloads[policy.manifest_file])
    frontmatter = _frontmatter(
        payloads[policy.skill_markdown_file], policy.maximum_object_nodes
    )
    if not isinstance(manifest, dict) or not isinstance(frontmatter, dict):
        raise ValueError("manifest and frontmatter must be objects")
    _validate_shape(
        manifest, policy.maximum_object_depth, policy.maximum_object_nodes
    )
    _validate_shape(
        frontmatter, policy.maximum_object_depth, policy.maximum_object_nodes
    )
    _reject_external_schema_references(policy.manifest_schema)
    jsonschema.Draft202012Validator.check_schema(policy.manifest_schema)
    errors = sorted(
        jsonschema.Draft202012Validator(policy.manifest_schema).iter_errors(
            manifest
        ),
        key=lambda error: list(error.absolute_path),
    )
    if errors:
        raise ValueError(
            "manifest schema rejected: "
            + "; ".join(error.message for error in errors[:10])
        )

    return NormalizedSkill(
        artifact_sha256=artifact_sha256,
        policy_revision=policy.revision,
        members=tuple(sorted(digests, key=lambda item: item.path)),
        manifest_canonical_json=_canonical_json(manifest),
        frontmatter_canonical_json=_canonical_json(frontmatter),
    )

Production implementation

Provision the policy provider with a signed profile whose limits reflect the deployment and whose Draft 2020-12 schema is closed with additionalProperties: false where appropriate. The loader container receives one read-only staged archive and emits the serialized NormalizedSkill plus its execution attestation; downstream stages must bind to artifact_sha256, policy_revision, canonical manifest/frontmatter bytes, and the full member-digest list. This reference rejects YAML aliases and arbitrary include expansion; if includes are required, add a separately bounded, cycle-detecting resolver to the profile instead of enabling parser-native fetching.

Independent verification

Replay the isolated loader with traversal, absolute and backslash paths, Unicode/case collisions, duplicate ZIP names, symlink/device entries, encrypted or unsupported compression, declared-size and compression-ratio bombs, concurrent package replacement, invalid UTF-8, duplicate JSON/YAML keys, YAML tags/aliases, excessive nodes/depth, unknown members, schema drift, missing required files, network attempts, and loader timeout. Confirm no rejected package releases normalized objects and that the positive output rehashes to the exact staged archive.

Open-source tools

• python-jsonschema
• Pydantic
• ruamel.yaml
• gVisor
• Firecracker

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise
• AML.T0011 User Execution
• AML.T0115 Publish Poisoned AI Artifacts
• AML.T0010.005 AI Supply Chain Compromise: AI Agent Tool
• AML.T0011.001 User Execution: Malicious Package (safe loader path prevents malicious package content from abusing the admission process itself)
• AML.T0115.002 Publish Poisoned AI Artifacts: AI Agent Tools (skill parsing and loader hardening reduce poisoned artifact preparation risk)
• AML.T0119 Exploit Automated Artifact Processing Pipeline (hardened loaders keep the skill admission pipeline from being exploited by the artifact it parses)

MAESTRO

• Input Validation Attacks (L3) (strict parser and loader constraints prevent malformed skill files from abusing admission-time parsing)
• Supply Chain Attacks (Cross-Layer) (safe parsing hardens the first stage of handling third-party skill artifacts)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks (safe parsing and loader hardening defend the artifact-ingestion stage)

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities (skill loader hardening protects enterprise ingestion of third-party skills)
• ASI05:2026 Unexpected Code Execution (RCE) (safe parser defaults reduce deserialization and loader-triggered execution risk)

NIST Adversarial Machine Learning 2025

• N/A (safe skill parsing is primarily a software artifact ingestion control rather than a direct AML threat mapping)

Cisco Integrated AI Security and Safety Framework

• AISubtech-13.1.1 Compute Exhaustion
• AISubtech-13.1.2 Memory Flooding
• AISubtech-13.1.4 Application Denial of Service
• AISubtech-9.3.1 Malicious Package / Tool Injection
• AITech-13.1 Disruption of Availability
• AITech-9.3 Dependency / Plugin Compromise

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (safe parsing validates integrated skill components before deeper analysis)

Databricks AI Security Framework 3.0

• Agents - Core 13.11: Unexpected RCE and Code Attacks (safe loader behavior reduces deserialization-triggered execution risk)