Scope boundary

Owns integrity and auditability of feature transformation logic, feature stores, transformed outputs, lineage, and training-serving consistency after raw data sanitization.

AID-H-002.001: AID-H-002.004 verifies how admitted raw records become model features; AID-H-002.001 validates and sanitizes the raw records themselves. A clean source record can still become a poisoned or skewed feature through compromised transformation logic.

Implementation guidance

• AID-H-002.004-G001: Keep all feature transformation code in version control with signed commits, protected branches, and mandatory review gates before deployment to training or serving pipelines.
• AID-H-002.004-G002: Continuously verify training-serving feature consistency so the features served at inference match the features the model was trained to expect.
• AID-H-002.004-G003: Run statistical validation on transformed feature outputs after each major pipeline step and block training if high-impact features violate baseline expectations.
• AID-H-002.004-G004: Record end-to-end feature lineage from raw data source through transformation steps to the final model input so incidents can be traced back quickly.

AID-H-002.004-G001

Production implementation: verify effective repository rules

A workflow grep is not branch protection. Query effective rules for each deployment branch and make this readback a required check. GitHub's signature rule proves a recognized signature; separately verify signer authorization against the organizational trust policy.

# File: ci/verify_feature_repository_rules.py
from __future__ import annotations
import json
import sys
from pathlib import Path

def verify(policy_path: Path, rules_path: Path) -> None:
    policy = json.loads(policy_path.read_text(encoding="utf-8"))
    rules = json.loads(rules_path.read_text(encoding="utf-8"))
    by_type = {rule["type"]: rule for rule in rules}
    required = {"deletion", "non_fast_forward", "required_signatures", "pull_request", "required_status_checks"}
    if not required.issubset(by_type):
        raise SystemExit("effective branch rules are incomplete")
    pull = by_type["pull_request"]["parameters"]
    if pull["required_approving_review_count"] < policy["minimum_approving_reviews"] or (policy["require_code_owner_review"] and not pull["require_code_owner_review"]) or pull.get("dismiss_stale_reviews_on_push") is not True:
        raise SystemExit("effective pull-request rule is weaker than policy")
    observed = {check["context"] for check in by_type["required_status_checks"]["parameters"]["required_status_checks"]}
    if not set(policy["required_status_checks"]).issubset(observed):
        raise SystemExit("required status-check population differs")

if __name__ == "__main__":
    verify(Path(sys.argv[1]), Path(sys.argv[2]))

Pinned readback job

permissions:
  contents: read
steps:
  - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
    with:
      fetch-depth: 0
  - name: Read and verify effective rules
    run: |
      set -euo pipefail
      DEFAULT_BRANCH=$(gh api "repos/$GITHUB_REPOSITORY" --jq .default_branch)
      gh api -H "Accept: application/vnd.github+json" -H "X-GitHub-Api-Version: 2022-11-28" \
        "repos/$GITHUB_REPOSITORY/rules/branches/$DEFAULT_BRANCH" > effective-rules.json
      python ci/verify_feature_repository_rules.py policy/feature-repository-policy.json effective-rules.json
      git verify-commit "$GITHUB_SHA"

Inject GH_TOKEN at job scope from the least-privilege workflow token. Retain exact rules, policy/commit/tree digests, signer fingerprint, review/owner records, status-check conclusions, and repository/ruleset IDs; re-read after merge and before deployment.

Keep the action safely bounded

block merge, training, or serving promotion on unreviewed code, incomplete lineage, schema violation, or training-serving mismatch.

AID-H-002.004-G002

Production implementation: replay both transforms on one golden population

Do not compare caller-supplied version strings. Run exact training and serving transformer artifacts against the same signed inputs, then compare complete outputs under a signed per-feature tolerance policy.

# File: feature_integrity/compare_outputs.py
from __future__ import annotations
import hashlib
import json
import math
from pathlib import Path

def load_jsonl(path: Path) -> dict[str, dict]:
    rows = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        row = json.loads(line)
        row_id = row.pop("case_id")
        if not isinstance(row_id, str) or not row_id or row_id in rows:
            raise ValueError("case IDs must be unique and non-empty")
        rows[row_id] = row
    if not rows:
        raise ValueError("output population is empty")
    return rows

def compare(training_output: Path, serving_output: Path, tolerance_policy: dict) -> dict:
    training, serving = load_jsonl(training_output), load_jsonl(serving_output)
    if set(training) != set(serving):
        raise PermissionError("training and serving case populations differ")
    failures = []
    for case_id in sorted(training):
        if set(training[case_id]) != set(serving[case_id]):
            failures.append({"case_id": case_id, "reason": "feature_set"})
            continue
        for feature in sorted(training[case_id]):
            left, right = training[case_id][feature], serving[case_id][feature]
            tolerance = tolerance_policy["numeric_absolute_tolerance"].get(feature)
            equal = left == right if tolerance is None else (not isinstance(left, bool) and not isinstance(right, bool) and math.isfinite(float(left)) and math.isfinite(float(right)) and math.isclose(float(left), float(right), rel_tol=0.0, abs_tol=float(tolerance)))
            if not equal:
                failures.append({"case_id": case_id, "feature": feature})
    return {"schema_version": "aidefend.feature-consistency.v1", "population_count": len(training), "evaluated_count": len(training), "failure_count": len(failures), "failures": failures, "training_output_sha256": hashlib.sha256(training_output.read_bytes()).hexdigest(), "serving_output_sha256": hashlib.sha256(serving_output.read_bytes()).hexdigest()}

Execute immutable transformers

set -euo pipefail
case "$TRAIN_TRANSFORM_IMAGE" in *@sha256:*) ;; *) exit 2 ;; esac
case "$SERVE_TRANSFORM_IMAGE" in *@sha256:*) ;; *) exit 2 ;; esac
docker run --rm --network none --read-only -v "$PWD/golden-input.jsonl:/input.jsonl:ro" -v "$PWD/out:/out" "$TRAIN_TRANSFORM_IMAGE" --input /input.jsonl --output /out/training.jsonl
docker run --rm --network none --read-only -v "$PWD/golden-input.jsonl:/input.jsonl:ro" -v "$PWD/out:/out" "$SERVE_TRANSFORM_IMAGE" --input /input.jsonl --output /out/serving.jsonl

Action: Invoke the comparator with the signature-verified tolerance policy and bind input, image, output, schema, policy, and comparator digests in the signed decision. Missing cases/features, non-finite values, mismatch, or output mutation blocks promotion.

Before you begin

Apply AID-H-002.004-G002 only when feature values are transformed between raw-source ingestion, training, batch scoring, or online serving.

AID-H-002.004-G003

Delivery level: reusable pipeline gate

Run this module on the exact transformed-feature snapshot produced by each high-impact pipeline step. The signed profile owns feature bounds and distribution tolerances; the module owns complete-row validation, a deterministic report, and the blocking exit code.

Step 1: Define and sign the feature profile

{
  "schema_version": "aidefend.transformed-feature-profile.v1",
  "policy_version": "fraud-features-2026-07",
  "minimum_rows": 1000,
  "features": {
    "avg_txn_amount_30d": {
      "minimum": 0.0,
      "maximum": 1000000.0,
      "maximum_null_fraction": 0.0,
      "baseline_mean": 218.4,
      "baseline_standard_deviation": 91.2,
      "maximum_mean_z_shift": 0.35
    }
  }
}

Store the profile and its Cosign bundle in read-only policy storage. Thresholds are organization-specific and must not be copied from this illustrative shape.

Step 2: Validate every row and emit the gate report

# File: feature_integrity/validate_transformed_snapshot.py
from __future__ import annotations

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


def strict_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 reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number: {value}")


def load_json_bytes(payload: bytes) -> Any:
    return json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=strict_object,
        parse_constant=reject_constant,
    )


def sha256_bytes(payload: bytes) -> str:
    return hashlib.sha256(payload).hexdigest()


def verified_policy(
    policy_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("verification timeout must be finite and positive")
    payload = policy_path.read_bytes()
    with tempfile.NamedTemporaryFile(prefix="feature-profile-", 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)
    profile = load_json_bytes(payload)
    if not isinstance(profile, dict) or set(profile) != {
        "schema_version", "policy_version", "minimum_rows", "features"
    }:
        raise ValueError("feature profile has an unsupported schema")
    if profile["schema_version"] != "aidefend.transformed-feature-profile.v1":
        raise ValueError("unsupported feature profile version")
    if not isinstance(profile["policy_version"], str) or not profile["policy_version"]:
        raise ValueError("policy_version is required")
    if type(profile["minimum_rows"]) is not int or profile["minimum_rows"] < 1:
        raise ValueError("minimum_rows must be a positive integer")
    if not isinstance(profile["features"], dict) or not profile["features"]:
        raise ValueError("at least one high-impact feature is required")
    required = {
        "minimum", "maximum", "maximum_null_fraction", "baseline_mean",
        "baseline_standard_deviation", "maximum_mean_z_shift",
    }
    for name, rule in profile["features"].items():
        if not isinstance(name, str) or not name or not isinstance(rule, dict) or set(rule) != required:
            raise ValueError("invalid feature rule")
        values = [float(rule[key]) for key in required]
        if not all(math.isfinite(value) for value in values):
            raise ValueError(f"non-finite rule for {name}")
        if (
            float(rule["minimum"]) > float(rule["maximum"])
            or not 0 <= float(rule["maximum_null_fraction"]) <= 1
            or float(rule["baseline_standard_deviation"]) <= 0
            or float(rule["maximum_mean_z_shift"]) < 0
        ):
            raise ValueError(f"invalid bounds for {name}")
    return profile, sha256_bytes(payload)


def validate_snapshot(snapshot_path: Path, profile: dict[str, Any]) -> dict[str, Any]:
    payload = snapshot_path.read_bytes()
    rows: list[dict[str, Any]] = []
    row_ids: set[str] = set()
    for line_number, line in enumerate(payload.splitlines(), start=1):
        if not line.strip():
            raise ValueError(f"blank JSONL row at line {line_number}")
        row = load_json_bytes(line)
        if not isinstance(row, dict):
            raise ValueError(f"row {line_number} is not an object")
        row_id = row.get("row_id")
        if not isinstance(row_id, str) or not row_id or row_id in row_ids:
            raise ValueError(f"row_id is absent or duplicate at line {line_number}")
        row_ids.add(row_id)
        rows.append(row)
    if len(rows) < profile["minimum_rows"]:
        raise RuntimeError("transformed population is smaller than signed policy minimum")

    violations: list[dict[str, Any]] = []
    feature_results: dict[str, Any] = {}
    for feature, rule in sorted(profile["features"].items()):
        values: list[float] = []
        null_count = 0
        for row in rows:
            raw = row.get(feature)
            if raw is None:
                null_count += 1
                continue
            if isinstance(raw, bool) or not isinstance(raw, (int, float)):
                violations.append({"row_id": row["row_id"], "feature": feature, "reason": "non_numeric"})
                continue
            value = float(raw)
            if not math.isfinite(value):
                violations.append({"row_id": row["row_id"], "feature": feature, "reason": "non_finite"})
                continue
            values.append(value)
            if value < float(rule["minimum"]) or value > float(rule["maximum"]):
                violations.append({"row_id": row["row_id"], "feature": feature, "reason": "outside_range"})
        null_fraction = null_count / len(rows)
        if null_fraction > float(rule["maximum_null_fraction"]):
            violations.append({"feature": feature, "reason": "null_fraction", "observed": null_fraction})
        if not values:
            violations.append({"feature": feature, "reason": "no_numeric_values"})
            observed_mean = None
            mean_z_shift = None
        else:
            observed_mean = sum(values) / len(values)
            mean_z_shift = abs(observed_mean - float(rule["baseline_mean"])) / float(rule["baseline_standard_deviation"])
            if mean_z_shift > float(rule["maximum_mean_z_shift"]):
                violations.append({"feature": feature, "reason": "mean_shift", "observed_z": mean_z_shift})
        feature_results[feature] = {
            "numeric_count": len(values), "null_count": null_count,
            "observed_mean": observed_mean, "mean_z_shift": mean_z_shift,
        }
    return {
        "schema_version": "aidefend.transformed-feature-validation.v1",
        "policy_version": profile["policy_version"],
        "input_sha256": sha256_bytes(payload),
        "population_count": len(rows),
        "evaluated_count": len(rows),
        "feature_results": feature_results,
        "violations": violations,
        "outcome": "FAIL" if violations else "PASS",
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--policy", type=Path, required=True)
    parser.add_argument("--policy-bundle", type=Path, required=True)
    parser.add_argument("--policy-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()
    profile, policy_sha256 = verified_policy(
        args.policy, args.policy_bundle, args.policy_key, args.verify_timeout_seconds
    )
    report = validate_snapshot(args.input, profile)
    report["policy_sha256"] = policy_sha256
    args.report.parent.mkdir(parents=True, exist_ok=True)
    temporary = args.report.with_suffix(args.report.suffix + ".tmp")
    temporary.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    os.replace(temporary, args.report)
    return 0 if report["outcome"] == "PASS" else 1


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

Step 3: Wire the blocking stage

python feature_integrity/validate_transformed_snapshot.py   --input artifacts/transformed-features.jsonl   --policy policy/transformed-feature-profile.json   --policy-bundle policy/transformed-feature-profile.sigstore.json   --policy-key /opt/aidefend/trust/feature-profile.pub   --report evidence/transformed-feature-validation.json   --verify-timeout-seconds "$FEATURE_POLICY_VERIFY_TIMEOUT_SECONDS"

The pipeline must promote only the exact input digest recorded in a PASS report. A malformed profile, signature failure, duplicate row, incomplete population, non-finite value, dependency error, or policy breach stops the stage; it is never converted to PASS.

Action: Run this gate against every transformed release partition before training or evaluation, and block promotion unless the complete signed population produces PASS for the exact inputs and policy.

AID-H-002.004-G004

Concept:

For every feature set used by a model, retain one lineage record that identifies immutable raw-source bytes, the exact transformation revision and runtime image, the authoritative pipeline run, the feature-store materialization, the feature schema, and the consuming model run. A mutable URI or short Git prefix is not enough to reconstruct an incident.

Record immutable identities

{
  "schema_version": "aidefend.feature-lineage.v1",
  "feature_set": "fraud_features_v12",
  "raw_source": {
    "uri": "s3://raw/payments/2026-03-17.parquet",
    "version_id": "3LgYJ7Qk2mF4pN8sV6xC1bR0tW9uH5eA",
    "sha256": "9d4f92e3ab6f4767d6c2f6c5fe1da3e46860887d2c05c0e24e4e19b73d4fb64c"
  },
  "transform": {
    "git_commit": "9f3e2a1c4d5b6e708192a3b4c5d6e7f8091a2b3c",
    "container_image_digest": "sha256:5d2dc1b58f8da5d323aa481e655607fb8f5bff4ffac54fc7538aa45b8cc9cb20"
  },
  "pipeline_run_id": "airflow_12345",
  "materialization_id": "feast_20260317_1200",
  "feature_schema_sha256": "7f145fd24a605bb4f0eb77c7f1e3f125c99ef9ad91d5a41aa659205012a32a67",
  "feature_manifest_sha256": "304cf3a7a46d709909c87b628b66286b81da80c3c44afc96b831037c6d0a369f",
  "model_training_run_id": "mlflow_run_abc"
}

Re-hash the version-pinned source

# File: feature_lineage/verify_s3_source.py
from __future__ import annotations

import hashlib
import hmac
from urllib.parse import urlparse

import boto3


def sha256_versioned_s3(uri: str, version_id: str) -> str:
    parsed = urlparse(uri)
    if parsed.scheme != "s3" or not parsed.netloc or not parsed.path.lstrip("/"):
        raise ValueError("raw source must be an s3:// bucket/key URI")
    if parsed.params or parsed.query or parsed.fragment or not version_id:
        raise ValueError("raw source URI or VersionId is invalid")
    response = boto3.client("s3").get_object(
        Bucket=parsed.netloc,
        Key=parsed.path.lstrip("/"),
        VersionId=version_id,
    )
    if response.get("VersionId") != version_id:
        raise PermissionError("S3 source VersionId differs")
    digest = hashlib.sha256()
    observed_bytes = 0
    body = response["Body"]
    while True:
        chunk = body.read(1024 * 1024)
        if not chunk:
            break
        observed_bytes += len(chunk)
        digest.update(chunk)
    if observed_bytes != response["ContentLength"]:
        raise IOError("versioned S3 source was not read completely")
    return digest.hexdigest()


def verify_source(lineage: dict) -> None:
    expected = lineage["raw_source"]["sha256"]
    observed = sha256_versioned_s3(
        lineage["raw_source"]["uri"],
        lineage["raw_source"]["version_id"],
    )
    if not hmac.compare_digest(observed, expected):
        raise PermissionError("raw-source digest differs from lineage")

Verify the complete lineage before promotion

1. Derive the expected feature-set population from the exact model input schema or promotion manifest and require set equality with the lineage records; do not silently ignore an unrecorded feature.
2. Fetch every raw object by immutable version and recompute its digest. For another object store, use its equivalent immutable generation identifier and authenticated byte readback.
3. Resolve the full Git commit from the protected repository and the image digest from the registry, then compare them with the transformation actually launched by the pipeline rather than a label supplied by the transform job.
4. Query the orchestrator and feature store under a read-only identity for the recorded run and materialization. Confirm the feature-set name, schema digest, canonical feature-manifest digest, completion state, and consuming model-run binding.
5. Store the canonical lineage record with the existing AID-M-002 lineage system and bind its digest to model promotion. Re-run the checks after a source-version, transform, image, schema, materialization, or model-run change.

Exercise a valid record plus changed source bytes, wrong object version, short or unknown commit, mutable image tag, missing feature set, stale materialization, schema mismatch, and model-run mismatch. Any mismatch or unavailable authoritative readback blocks promotion; it must not be treated as a complete lineage record.

Before you begin

Apply AID-H-002.004-G004 only when feature values are transformed between raw-source ingestion, training, batch scoring, or online serving.

Open-source tools

• Feast
• Hopsworks
• GX Core (Great Expectations)
• dbt
• DVC
• Pandera
• Git
• Apache Airflow
• Prefect
• OpenLineage (feature-pipeline lineage instrumentation)
• Marquez (OpenLineage metadata collection and inspection)

Commercial / hosted tools

• Tecton
• Hopsworks Enterprise
• Databricks Feature Engineering in Unity Catalog
• Amazon SageMaker Feature Store
• Arize AX
• Fiddler AI Observability Platform
• Evidently Cloud
• NannyML Cloud

Threat mappings

MITRE ATLAS

• AML.T0020 Training Data Poisoning (feature-level poisoning through transformation logic)
• AML.T0059 Erode Dataset Integrity (feature pipeline corruption erodes dataset integrity)

MAESTRO

• Data Poisoning (L2) (feature-level poisoning through transformation layer)
• Data Tampering (L2) (feature transformation logic tampering)

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• ML02:2023 Data Poisoning Attack (feature-level poisoning)
• ML08:2023 Model Skewing (training-serving feature inconsistency)

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.013 Data Poisoning (feature-level poisoning)
• NISTAML.024 Targeted Poisoning (targeted feature manipulation)

Cisco Integrated AI Security and Safety Framework

• AITech-6.1 Training Data Poisoning
• AITech-9.1 Model or Agentic System Manipulation (Protects serving-time feature transformations from runtime manipulation)

Google Secure AI Framework 2.0 - Risks

• DP: Data Poisoning (feature pipeline as poisoning vector)

Databricks AI Security Framework 3.0

• Data Prep 2.1: Preprocessing integrity
• Data Prep 2.2: Feature manipulation
• Data Prep 2.3: Raw data criteria
• Algorithms 5.1: Lack of tracking and reproducibility of experiments (feature pipeline versioning ensures reproducibility)
• Raw Data 1.6: Insufficient data lineage (feature lineage from raw source to model input)