Scope boundary

Owns external dataset supplier and provenance admission. It does not characterize the security and privacy properties of the dataset contents or sanitize individual records.

AID-M-002.003: AID-H-003.003 decides whether the dataset's supplier, provenance, rights, signature, and lineage are admissible; AID-M-002.003 characterizes sensitive content, distributions, malformed records, and poisoning indicators. Trusted provenance does not make content safe, and safe-looking content does not approve an untrusted source.

AID-H-002.001: AID-H-003.003 admits the external source and immutable dataset artifact; AID-H-002.001 sanitizes and quarantines unsafe records in the candidate snapshot. Supplier admission and record-level content admission must both pass before training use.

Implementation guidance

• AID-H-003.003-G001: Maintain per-file hashes in DVC/LakeFS; block pipeline if hash drift.
• AID-H-003.003-G002: Verify publisher identity and a signed immutable dataset manifest before admission.
• AID-H-003.003-G003: Require documented source and license approval before datasets enter shared storage or feature stores.
• AID-H-003.003-G004: Embed signed provenance metadata ('Datasheets for Datasets') for auditing.

AID-H-003.003-G001

Concept:

Data Version Control (DVC) and LakeFS are tools that bring Git-like principles to data. They work by storing small 'pointer' files in Git that contain the hash of the actual data, which is stored in a separate object storage. This allows you to version your data and ensure that a change in the data is as explicit as a change in code.

Step 1: Track a Dataset with DVC

Use the DVC command-line tool to start tracking your dataset. This creates a `.dvc` file that contains the hash and location of your data.

# --- Terminal Commands ---

# Initialize DVC in your Git repo
> dvc init

# Configure remote storage (e.g., an S3 bucket)
> dvc remote add -d my-storage s3://my-data-bucket/dvc-store

# Tell DVC to track your data file
> dvc add data/training_data.csv

# Now, commit the .dvc pointer file to Git
> git add data/training_data.csv.dvc .gitignore
> git commit -m "Track initial training data"

Step 2: Verify Data in CI/CD

In your training pipeline, instead of just using the data, you first run `dvc pull`. This command checks the hash in the `.dvc` file against the data in your remote storage. If there's a mismatch, or if the local file has been changed without being re-committed via DVC, the pipeline can be configured to fail.

# In your CI/CD script (e.g., GitHub Actions)
steps:
  - name: Pull and verify data with DVC
    run: |
      dvc pull data/training_data.csv
      # Capture machine-readable status; --quiet suppresses the evidence needed here.
      dvc status --json > dvc-status.json
      if ! jq -e 'type == "object" and length == 0' dvc-status.json >/dev/null; then
        echo "DVC reports drift, missing data, or an uncommitted change." >&2
        cat dvc-status.json >&2
        exit 1
      fi

Action: Use DVC or LakeFS to manage all of your training, validation, and RAG datasets. Make `dvc pull` and a status check a mandatory first step in any pipeline that consumes these datasets to ensure integrity.

Before you begin

Apply AID-H-003.003-G001 only when training, fine-tuning, evaluation, or feature data is acquired, mirrored, licensed, transformed, or shared across trust boundaries.

Keep the action safely bounded

Action: Maintain per-file hashes and signed source/rights provenance in DVC, LakeFS, or an equivalent immutable data registry. Block admission on drift, missing provenance, unapproved rights, incomplete population, or signature failure. Dataset-content privacy and sensitive-data characterization remain owned by AID-M-002.003.

AID-H-003.003-G002

Delivery level: reusable dataset-admission verifier

DVC is the acquisition dependency, not the trust decision. Admission verifies the publisher signature over one immutable manifest snapshot and independently reconciles every staged regular file against that manifest. A successful dvc status command alone is not evidence of publisher identity or dataset membership.

Step 1: Publish a closed signed manifest

{
  "schema_version": "aidefend.dataset-manifest.v1",
  "dataset_id": "fraud-training",
  "dataset_version": "2026-07-26",
  "files": [
    {
      "path": "splits/train.parquet",
      "size": 4829317,
      "sha256": "7f6d2a7cb76f723cd88d8f498e238faba673ac9b841c4a7e156f7bb8522e2406"
    }
  ]
}

The publisher signs the exact JSON bytes with the approved dataset-signing identity. The manifest must list the complete dataset root; do not include the manifest or signature bundle inside that root.

Step 2: Acquire into a private staging directory

set -euo pipefail
umask 077
test -n "$DVC_REMOTE"
test -n "$DVC_TARGET"
stage="$(mktemp -d)"
trap 'rm -rf -- "$stage"' EXIT
dvc pull --remote "$DVC_REMOTE" "$DVC_TARGET"
test -d "$DVC_TARGET"
install -d -m 0700 "$stage/dataset"
cp -a --reflink=auto -- "$DVC_TARGET"/. "$stage/dataset/"
python admission/verify_dataset_manifest.py   --dataset-root "$stage/dataset"   --manifest evidence/dataset-manifest.json   --bundle evidence/dataset-manifest.sigstore.json   --public-key /opt/aidefend/trust/dataset-publisher.pub   --report "$stage/admission-report.json"   --verify-timeout-seconds "$DATASET_SIGNATURE_TIMEOUT_SECONDS"
# Promote the staged directory by its reported manifest digest; never return to the mutable source path.

Step 3: Verify signature, membership, size, and digest

# File: admission/verify_dataset_manifest.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path
from typing import Any

SHA256 = re.compile(r"[0-9a-f]{64}")


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


def strict_json(payload: bytes) -> Any:
    return json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")),
    )


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def verify_manifest(
    manifest_path: Path, bundle_path: Path, public_key: Path, timeout_seconds: float
) -> tuple[dict[str, Any], str]:
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("signature timeout must be finite and positive")
    payload = manifest_path.read_bytes()
    with tempfile.NamedTemporaryFile(prefix="dataset-manifest-", delete=False) as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
        snapshot = Path(handle.name)
    try:
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(public_key),
                "--bundle", str(bundle_path), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
    finally:
        snapshot.unlink(missing_ok=True)
    manifest = strict_json(payload)
    if not isinstance(manifest, dict) or set(manifest) != {
        "schema_version", "dataset_id", "dataset_version", "files"
    }:
        raise ValueError("dataset manifest schema mismatch")
    if manifest["schema_version"] != "aidefend.dataset-manifest.v1":
        raise ValueError("unsupported dataset manifest")
    if any(not isinstance(manifest[key], str) or not manifest[key] for key in ("dataset_id", "dataset_version")):
        raise ValueError("dataset identity is incomplete")
    if not isinstance(manifest["files"], list) or not manifest["files"]:
        raise ValueError("dataset manifest has no files")
    expected: set[str] = set()
    for entry in manifest["files"]:
        if not isinstance(entry, dict) or set(entry) != {"path", "size", "sha256"}:
            raise ValueError("invalid manifest member")
        relative = entry["path"]
        candidate = Path(relative)
        if (
            not isinstance(relative, str) or not relative
            or candidate.is_absolute() or ".." in candidate.parts
            or candidate.as_posix() != relative or relative in expected
        ):
            raise ValueError("manifest path is unsafe or duplicate")
        if type(entry["size"]) is not int or entry["size"] < 0:
            raise ValueError("manifest size is invalid")
        if not isinstance(entry["sha256"], str) or not SHA256.fullmatch(entry["sha256"]):
            raise ValueError("manifest digest is invalid")
        expected.add(relative)
    return manifest, hashlib.sha256(payload).hexdigest()


def reconcile(root: Path, manifest: dict[str, Any]) -> list[dict[str, Any]]:
    root = root.resolve(strict=True)
    actual: dict[str, Path] = {}
    for path in root.rglob("*"):
        if path.is_symlink():
            raise ValueError(f"dataset contains a symlink: {path}")
        if path.is_file():
            relative = path.relative_to(root).as_posix()
            if relative in actual:
                raise ValueError(f"duplicate dataset member: {relative}")
            actual[relative] = path
        elif not path.is_dir():
            raise ValueError(f"dataset contains a non-regular member: {path}")
    expected = {entry["path"]: entry for entry in manifest["files"]}
    if set(actual) != set(expected):
        missing = sorted(set(expected) - set(actual))
        unexpected = sorted(set(actual) - set(expected))
        raise PermissionError(f"dataset population mismatch missing={missing} unexpected={unexpected}")
    verified = []
    for relative in sorted(expected):
        path = actual[relative]
        before = path.stat()
        digest = file_sha256(path)
        after = path.stat()
        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (
            after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns
        ):
            raise RuntimeError(f"dataset member changed during verification: {relative}")
        entry = expected[relative]
        if before.st_size != entry["size"] or digest != entry["sha256"]:
            raise PermissionError(f"dataset member does not match manifest: {relative}")
        verified.append({"path": relative, "size": before.st_size, "sha256": digest})
    return verified


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dataset-root", type=Path, required=True)
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--bundle", type=Path, required=True)
    parser.add_argument("--public-key", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--verify-timeout-seconds", type=float, required=True)
    args = parser.parse_args()
    manifest, manifest_sha256 = verify_manifest(
        args.manifest, args.bundle, args.public_key, args.verify_timeout_seconds
    )
    verified = reconcile(args.dataset_root, manifest)
    report = {
        "schema_version": "aidefend.dataset-admission-result.v1",
        "dataset_id": manifest["dataset_id"],
        "dataset_version": manifest["dataset_version"],
        "manifest_sha256": manifest_sha256,
        "file_count": len(verified),
        "files": verified,
        "outcome": "PASS",
    }
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Signature failure, a malformed or duplicate manifest entry, missing or unexpected file, symlink, size mismatch, digest mismatch, or mutation during hashing fails the admission job. Downstream training receives only the same staged directory identified by the PASS report.

Action: Acquire data only into the private staging directory and release it downstream only after the publisher-signed manifest and every staged member reconcile exactly with a durable PASS report.

AID-H-003.003-G003

Production implementation: signed source and rights manifest

A pull-request checklist is not a release gate. Require one canonical manifest signed by the data-governance identity and bound to the exact dataset tree digest.

# File: dataset_admission/verify_rights_manifest.py
from __future__ import annotations
import math
import json
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path

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

DATA_GOVERNANCE_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H003_DATA_GOVERNANCE_VERIFY_TIMEOUT_SECONDS")


FIELDS = {"schema_version", "dataset_id", "dataset_tree_sha256", "source_uri", "source_revision", "license_expression", "intended_uses", "prohibited_uses", "owner", "review_ticket", "approved_at", "not_after", "policy_version"}

def verify(path: Path, signature: Path, *, expected_tree_sha256: str, allowed_licenses: set[str], intended_use: str) -> dict:
    raw, sig = path.read_bytes(), signature.read_bytes()
    with tempfile.TemporaryDirectory(prefix="dataset-rights-") as directory:
        snapshot, sig_snapshot = Path(directory) / "manifest.json", Path(directory) / "manifest.sig"
        snapshot.write_bytes(raw)
        sig_snapshot.write_bytes(sig)
        subprocess.run(["cosign", "verify-blob", "--key", os.environ["DATA_GOVERNANCE_VERIFY_KEY"], "--bundle", str(sig_snapshot), str(snapshot)], check=True, capture_output=True, text=True, timeout=DATA_GOVERNANCE_VERIFY_TIMEOUT_SECONDS)
        verified = snapshot.read_bytes()
    manifest = json.loads(verified.decode("utf-8", errors="strict"))
    if set(manifest) != FIELDS or manifest["schema_version"] != "aidefend.dataset-rights-manifest.v1":
        raise ValueError("rights manifest schema differs")
    expiry = datetime.fromisoformat(manifest["not_after"].replace("Z", "+00:00"))
    if expiry.tzinfo is None or expiry <= datetime.now(timezone.utc):
        raise PermissionError("rights approval is expired")
    if manifest["dataset_tree_sha256"] != expected_tree_sha256:
        raise PermissionError("rights manifest is bound to different bytes")
    if manifest["license_expression"] not in allowed_licenses:
        raise PermissionError("license is outside signed policy")
    if intended_use not in manifest["intended_uses"] or intended_use in manifest["prohibited_uses"]:
        raise PermissionError("intended use is not approved")
    return manifest

Action: Verify source revision, SPDX license expression, owner, review ticket, intended/prohibited uses, policy version, exact bytes, signature, and freshness before mirror admission. Legal or governance review supplies the signed decision; this technical control enforces it without claiming to determine legal rights.

Keep the action safely bounded

block admission on drift, missing provenance, unapproved rights, unresolved sensitive data, incomplete population, or signature failure.

AID-H-003.003-G004

Concept:

A 'datasheet for datasets' is a standardized document that describes a dataset's motivation, composition, collection process, and recommended uses. Creating and cryptographically signing this datasheet provides a verifiable record of the data's provenance that can be used for auditing and building trust.

Step 1: Create a Datasheet Template

Define a standard Markdown or JSON template for your datasheets.

# File: templates/datasheet_template.md

## Datasheet for: [Dataset Name]

- **Version:** [e.g., 1.2]
- **SHA-256 Hash:** [Hash of the dataset artifact]
- **Date:** YYYY-MM-DD

### Motivation
- **For what purpose was the dataset created?**

### Composition
- **What do the instances that comprise the dataset represent?**
- **How many instances are there?**

### Collection Process
- **How was the data collected?**
- **Who was involved in the data collection process?**

### Preprocessing/Cleaning/Labeling
- **Was the data cleaned, processed, or annotated? If so, how?**

### Uses & Known Limitations
- **What are the recommended uses for this dataset?**
- **What are known limitations or biases in this data?**

Step 2: Generate and Sign the Datasheet

In your data processing pipeline, after the data is finalized, generate its hash, fill out the datasheet, sign the datasheet with Cosign v3.1.2, and verify the emitted bundle against the deployment-pinned certificate identity and OIDC issuer before distribution.

# File: ci/build_and_sign_datasheet.sh
#!/usr/bin/env bash
set -euo pipefail

readonly DATASET="dist/my_dataset.csv.gz"
readonly DATASHEET="dist/datasheet.md"
readonly BUNDLE="dist/datasheet.sigstore.json"

mkdir -p dist
gzip -c data/final/my_dataset.csv > "$DATASET"
DATA_HASH="$(sha256sum "$DATASET" | awk '{print $1}')"
readonly DATA_HASH
python scripts/generate_datasheet.py \
  --hash "$DATA_HASH" \
  --output "$DATASHEET"

cosign sign-blob --yes --bundle "$BUNDLE" "$DATASHEET"

Step 3: Verify in a Separate Admission Job

Run this verifier under a separately administered admission identity. The signing job must not control the read-only trust-policy mounts. Admission captures the transferred dataset, datasheet, and bundle into one private snapshot, verifies and hashes only that snapshot, and hands downstream systems only the verified snapshot bytes or its immutable content digest; the mutable source paths are never promoted.

# File: admission/verify_datasheet.sh
#!/usr/bin/env bash
set -euo pipefail

readonly SOURCE_DATASET="dist/my_dataset.csv.gz"
readonly SOURCE_DATASHEET="dist/datasheet.md"
readonly SOURCE_BUNDLE="dist/datasheet.sigstore.json"
SNAPSHOT_DIR="$(mktemp -d)"
readonly SNAPSHOT_DIR
trap 'rm -rf -- "$SNAPSHOT_DIR"' EXIT
chmod 0700 "$SNAPSHOT_DIR"
install -m 0400 "$SOURCE_DATASET" "$SNAPSHOT_DIR/my_dataset.csv.gz"
install -m 0400 "$SOURCE_DATASHEET" "$SNAPSHOT_DIR/datasheet.md"
install -m 0400 "$SOURCE_BUNDLE" "$SNAPSHOT_DIR/datasheet.sigstore.json"

readonly CERTIFICATE_IDENTITY="$(tr -d '\r\n' < /opt/aidefend/trust/datasheet-signer.identity)"
readonly OIDC_ISSUER="$(tr -d '\r\n' < /opt/aidefend/trust/datasheet-signer.issuer)"
test -n "$CERTIFICATE_IDENTITY"
test -n "$OIDC_ISSUER"

cosign verify-blob \
  --bundle "$SNAPSHOT_DIR/datasheet.sigstore.json" \
  --certificate-identity "$CERTIFICATE_IDENTITY" \
  --certificate-oidc-issuer "$OIDC_ISSUER" \
  "$SNAPSHOT_DIR/datasheet.md"

MATCH_COUNT="$(sed -nE 's/^- \*\*SHA-256 Hash:\*\* ([0-9a-f]{64})$/\1/p' "$SNAPSHOT_DIR/datasheet.md" | wc -l | tr -d ' ')"
test "$MATCH_COUNT" -eq 1
EXPECTED_DATA_HASH="$(sed -nE 's/^- \*\*SHA-256 Hash:\*\* ([0-9a-f]{64})$/\1/p' "$SNAPSHOT_DIR/datasheet.md")"
readonly EXPECTED_DATA_HASH
ACTUAL_DATA_HASH="$(sha256sum "$SNAPSHOT_DIR/my_dataset.csv.gz" | awk '{print $1}')"
readonly ACTUAL_DATA_HASH
test "$ACTUAL_DATA_HASH" = "$EXPECTED_DATA_HASH"

# Emit the immutable handoff identity; downstream retrieves or copies only the verified snapshot.
printf 'sha256:%s\n' "$ACTUAL_DATA_HASH"

Action: Implement a pipeline step that generates a datasheet for every versioned dataset. The datasheet should include the dataset's hash. The signing job should emit a Sigstore bundle, and an independent admission job should verify the exact datasheet bytes against the deployment-approved signer identity and OIDC issuer before distribution.

Before you begin

Apply AID-H-003.003-G004 only when training, fine-tuning, evaluation, or feature data is acquired, mirrored, licensed, transformed, or shared across trust boundaries.

Keep the action safely bounded

block admission on drift, missing provenance, unapproved rights, unresolved sensitive data, incomplete population, or signature failure.

Open-source tools

• DVC
• lakeFS
• Witness
• in-toto
• OpenMetadata (dataset provenance evidence source; separate admission gate required)

Commercial / hosted tools

• Databricks Unity Catalog
• Alation
• Collibra

Threat mappings

MITRE ATLAS

• AML.T0074 Masquerading
• AML.T0020 Training Data Poisoning
• AML.T0010 AI Supply Chain Compromise (dataset supply chain is part of broader AI supply chain)
• AML.T0010.002 AI Supply Chain Compromise: Data

MAESTRO

• Data Tampering (L2)
• Supply Chain Attacks (Cross-Layer)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain
• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-7.3.1 Corrupted Third-Party Data
• AITech-6.1 Training Data Poisoning (Blocks alteration of an approved external training-data snapshot)
• AITech-7.3 Data Source Abuse and Manipulation

Google Secure AI Framework 2.0 - Risks

• UTD: Unauthorized Training Data

Databricks AI Security Framework 3.0

• Raw Data 1.7: Lack of data trustworthiness
• Raw Data 1.8: Legality of data
• Raw Data 1.11: Compromised 3rd-party datasets
• Raw Data 1.6: Insufficient data lineage (provenance metadata provides lineage)