AID-E-003.001: Compromised Model Quarantine & Recovery Routing

Version 1.20261015 · Actionable sub-technique

Make the exact confirmed or policy-mandated suspect model digest ineligible for serving, preserve it as incident evidence, and route the impact to detachable-adapter eviction, signed known-good rollback, clean retraining, or continued quarantine.

Pillars: model

Lifecycle: response, improvement

Scope boundary

Owns quarantine and serving ineligibility of an exact suspect whole-model digest, preservation of that digest as incident evidence, and signed routing to the appropriate recovery path. Candidate backdoor scans and differential tests are inputs and do not prove quarantine or recovery completion.

AID-E-003.005: AID-E-003.001 quarantines an affected whole-model digest; AID-E-003.005 detaches, tombstones, and traces compromised adapters and their derivatives. Use adapter-level eviction when the compromised influence can be separated. Route merged or irreducible whole-model impact back to this quarantine and the applicable Restore control.

AID-R-001: AID-E-003.001 makes a suspect model ineligible and preserves it as evidence; AID-R-001 releases a trusted rollback or newly remediated model. Quarantine is an eviction result, not recovered-model readiness, and model recovery does not prove that the suspect digest is unavailable everywhere.

Warning

LLM Backdoor Removal Requires Conservative Assurance

Neural Cleanse, Activation Clustering, and Fine-Pruning were designed primarily for classifier-style trigger backdoors and remain useful for vision or structured classifiers. LLM and sleeper-agent backdoors may not localize cleanly to one class, trigger patch, or small neuron set, and can survive safety fine-tuning or superficial behavior repair. For LLM incidents, treat these methods as triage evidence. Prefer rollback to a signed golden model, removal of malicious adapters, data-layer cleanup followed by retraining, or quarantine when high-assurance removal cannot be proven.

Implementation guidance

AID-E-003.001-G001

Route a reproducibly compromised neural-network artifact to adapter detachment, signed rollback, clean retraining, or continued quarantine.
Before you begin

Use after reproducible evidence confirms a specific neural-network artifact is compromised or policy requires continued quarantine pending a final finding.

When model quarantine applies:

Use this decision path after a malicious behavior is reproducible and the affected artifact is bound by digest. Activation clustering, neuron pruning, and safety fine-tuning are experiments that may generate candidates; none proves that a backdoor was present or removed. Never overwrite the quarantined original or promote an experimental fine-pruned/fine-tuned derivative as recovered solely because one trigger stopped firing.

Exclude the suspect artifact from serving before selecting recovery

This executable path applies to a model gateway whose authoritative artifact and route registry is SQLite. Enroll artifacts and route aliases under a separate registry-owner process. Only that process has filesystem write access to SQLite; responders submit authenticated incident manifests to its narrow quarantine operation and never receive database-file write permissions. SQLite does not provide per-table credential privileges. The serving identity opens a fresh read-only autocommit connection for each request and calls admit_request, including for cached models; close that connection after admission and never reuse an open read cursor or transaction across requests. The authenticated incident manifest supplies the suspect digest, complete route population and evidence digest. The responder applies quarantine_artifact, then a separate verifier opens the authoritative database read-only and runs verify_quarantine. Fail closed on registry errors; do not cache an allow across quarantine changes. Preserve the original artifact and reproducer in immutable incident evidence. This path stops new serving admission; already-running inference must be cancelled by the applicable Isolate/active-execution eviction owner before incident closure.

# File: backdoor_removal/quarantine_gateway.py
"""Executable model-gateway integration path embedded in E-003.001-G001."""
from __future__ import annotations

import hashlib
import json
import sqlite3
from pathlib import Path

from supply_chain.verify_tombstones import check


def require_digest(value: str) -> str:
    if not isinstance(value, str) or len(value) != 64 or set(value) - set("0123456789abcdef") or value == "0" * 64:
        raise ValueError("nonzero lowercase artifact SHA-256 required")
    return value


def initialize_registry(connection: sqlite3.Connection) -> None:
    # Provision once by the registry owner. Gateway identities have read only
    # access; responders invoke the authenticated registry-owner operation,
    # never receive a writable database connection or filesystem access.
    connection.executescript("""
        PRAGMA foreign_keys=ON;
        CREATE TABLE IF NOT EXISTS artifacts (
          sha256 TEXT PRIMARY KEY, source_uri TEXT NOT NULL,
          state TEXT NOT NULL CHECK(state IN ('ACTIVE','QUARANTINED')));
        CREATE TABLE IF NOT EXISTS routes (
          route_id TEXT PRIMARY KEY, artifact_sha256 TEXT NOT NULL
          REFERENCES artifacts(sha256));
        CREATE TABLE IF NOT EXISTS quarantine (
          artifact_sha256 TEXT PRIMARY KEY REFERENCES artifacts(sha256),
          incident_id TEXT NOT NULL, evidence_sha256 TEXT NOT NULL);
    """)


def quarantine_artifact(connection: sqlite3.Connection, *, suspect_sha256: str,
                        incident_id: str, evidence_sha256: str,
                        expected_routes: list[str]) -> dict:
    require_digest(suspect_sha256)
    require_digest(evidence_sha256)
    if not incident_id or not expected_routes or expected_routes != sorted(set(expected_routes)):
        raise ValueError("incident and complete sorted serving-route population required")
    connection.execute("BEGIN IMMEDIATE")
    try:
        artifact = connection.execute("SELECT source_uri FROM artifacts WHERE sha256=?", (suspect_sha256,)).fetchone()
        routes = [r[0] for r in connection.execute(
            "SELECT route_id FROM routes WHERE artifact_sha256=? ORDER BY route_id", (suspect_sha256,))]
        if artifact is None or routes != expected_routes:
            raise RuntimeError("authoritative artifact or route population differs")
        existing = connection.execute("SELECT incident_id,evidence_sha256 FROM quarantine WHERE artifact_sha256=?", (suspect_sha256,)).fetchone()
        if existing is not None and existing != (incident_id, evidence_sha256):
            raise RuntimeError("quarantine already has different incident evidence; preserve and reconcile")
        connection.execute("INSERT OR IGNORE INTO quarantine VALUES (?,?,?)", (suspect_sha256, incident_id, evidence_sha256))
        connection.execute("UPDATE artifacts SET state='QUARANTINED' WHERE sha256=?", (suspect_sha256,))
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    return {"artifact_sha256": suspect_sha256, "source_uri": artifact[0],
            "incident_id": incident_id, "evidence_sha256": evidence_sha256,
            "affected_routes": routes, "state": "QUARANTINED"}


def open_serving_registry(path: Path) -> sqlite3.Connection:
    # Create per request under the serving identity; close after the call.
    return sqlite3.connect(path.resolve(strict=True).as_uri() + "?mode=ro",
                           uri=True, isolation_level=None)


def admit_request(connection: sqlite3.Connection, route_id: str, *, policy_path: Path,
                  signature_path: Path, version_store) -> tuple[str, str]:
    # Run at the serving gateway on EVERY request, including warm-model/cache
    # paths. Neither a cached model handle nor a new alias bypasses this join.
    if connection.isolation_level is not None or connection.in_transaction:
        raise RuntimeError("serving registry requires fresh autocommit reads")
    row = connection.execute("""
        SELECT a.source_uri,a.sha256,a.state,q.artifact_sha256
        FROM routes r JOIN artifacts a ON a.sha256=r.artifact_sha256
        LEFT JOIN quarantine q ON q.artifact_sha256=a.sha256 WHERE r.route_id=?
    """, (route_id,)).fetchone()
    if row is None or row[2] != "ACTIVE" or row[3] is not None:
        raise PermissionError("model_quarantined_or_unregistered")
    require_digest(row[1])
    # Reuse H-003.006-G008 for signed, current load/promotion tombstone policy.
    check(row[0], row[1], policy_path, signature_path, version_store)
    return row[0], row[1]


def verify_quarantine(connection: sqlite3.Connection, receipt: dict, *, policy_path: Path,
                      signature_path: Path, version_store) -> dict:
    # Invoke using a separately opened read-only connection and verifier
    # identity after the writer commits. No receipt-only PASS is possible.
    digest = require_digest(receipt["artifact_sha256"])
    observed = connection.execute("SELECT state FROM artifacts WHERE sha256=?", (digest,)).fetchone()
    binding = connection.execute("SELECT incident_id,evidence_sha256 FROM quarantine WHERE artifact_sha256=?", (digest,)).fetchone()
    routes = [r[0] for r in connection.execute("SELECT route_id FROM routes WHERE artifact_sha256=? ORDER BY route_id", (digest,))]
    if observed != ("QUARANTINED",) or binding != (receipt["incident_id"], receipt["evidence_sha256"]) or routes != receipt["affected_routes"]:
        raise RuntimeError("quarantine independent readback differs")
    for route in routes:
        try:
            admit_request(connection, route, policy_path=policy_path, signature_path=signature_path, version_store=version_store)
        except PermissionError as error:
            if str(error) != "model_quarantined_or_unregistered":
                raise
        else:
            raise RuntimeError("quarantined route still admits model requests")
    return {"state": "SERVING_DENIED", "artifact_sha256": digest,
            "routes_sha256": hashlib.sha256(json.dumps(routes, separators=(",", ":")).encode()).hexdigest(),
            "incident_id": receipt["incident_id"], "evidence_sha256": receipt["evidence_sha256"]}

Also publish the suspect source URI/digest to the existing signed tombstone authority in AID-H-003.006-G008 so promotion, reload and restoration cannot reintroduce it. Its version_store must be the deployment's durable authoritative monotonic store outside artifact rollback. The registry owner prevents replacement of the authority or reset of quarantine state by loader credentials. For a different serving platform, replace this registry integration with its real authoritative API and repeat the same complete-population negative probes. Do not report serving denial merely because a recovery route string was selected.

Choose a production recovery owner
# File: backdoor_removal/route_production_recovery.py
from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class RecoveryEvidence:
    # Construct this record only after verifying the detached signature on the
    # canonical reproducer receipt under the pinned incident-verifier key.
    suspect_model_sha256: str
    verified_reproducer_receipt_sha256: str | None
    detachable_adapter_sha256: str | None
    signed_known_good_model_sha256: str | None
    clean_retraining_snapshot_sha256: str | None


def require_optional_sha256(value: str | None, field: str) -> str | None:
    if value is None:
        return None
    if (
        len(value) != 64
        or set(value) - set("0123456789abcdef")
        or value == "0" * 64
    ):
        raise ValueError(f"{field} must be a nonzero lowercase SHA-256 digest")
    return value


def select_recovery_route(evidence: RecoveryEvidence) -> str:
    suspect = require_optional_sha256(
        evidence.suspect_model_sha256, "suspect_model_sha256"
    )
    if suspect is None:
        raise ValueError("the quarantined suspect model digest is required")
    reproducer = require_optional_sha256(
        evidence.verified_reproducer_receipt_sha256,
        "verified_reproducer_receipt_sha256",
    )
    adapter = require_optional_sha256(
        evidence.detachable_adapter_sha256, "detachable_adapter_sha256"
    )
    rollback = require_optional_sha256(
        evidence.signed_known_good_model_sha256,
        "signed_known_good_model_sha256",
    )
    retraining = require_optional_sha256(
        evidence.clean_retraining_snapshot_sha256,
        "clean_retraining_snapshot_sha256",
    )
    if reproducer is None:
        return "quarantine_and_continue_triage"
    if adapter is not None:
        return "AID-E-003.005_detach_and_quarantine_adapter"
    if rollback is not None:
        return "AID-R-001.001_restore_signed_known_good_model"
    if retraining is not None:
        return "AID-R-001.002_retrain_from_verified_clean_data"
    return "quarantine_and_disable_serving"
Independent release evidence

The artifact owner must produce the quarantined-original digest, selected recovery route, new artifact digest, lineage, and full before/after evaluation inputs. A separate verifier must fetch those digest-pinned artifacts, replay the incident trigger suite plus clean-task regression suite, and sign its observations. Missing reproducer coverage, missing trusted rollback/retraining input, an unmeasured activation context, or a signature/digest mismatch is insufficient evidence and keeps the candidate quarantined.

Action: Route detachable adapters to AID-E-003.005, whole-model rollback to AID-R-001.001, and clean retraining to AID-R-001.002. Treat pruning or fine-tuning only as a separately evaluated candidate-generation method, never as proof of eradication.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-E-003.001: Compromised Model Quarantine & Recovery Routing | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service