[pitboss] phase 12: Track B — Python harness emitter shapes

tests/common/fixture_harness.rs

@@ -16,8 +16,8 @@
 use nyx_scanner::commands::scan::Diag;
 use nyx_scanner::dynamic::verify::{verify_finding, VerifyOptions};
 use nyx_scanner::evidence::{
-    Confidence, Evidence, FlowStep, FlowStepKind, InconclusiveReason, UnsupportedReason,
-    VerifyResult, VerifyStatus,
+    Confidence, EntryKind, Evidence, FlowStep, FlowStepKind, InconclusiveReason,
+    UnsupportedReason, VerifyResult, VerifyStatus,
 };
 use nyx_scanner::labels::Cap;
 use nyx_scanner::patterns::{FindingCategory, Severity};
@@ -192,6 +192,238 @@ fn stage_fixture(src: &Path, tmp: &TempDir, copy: CopyStrategy) -> PathBuf {
     }
 }
 
+/// Phase 12 — per-shape acceptance helper.
+///
+/// Stages `fixture_root/<shape>/<file>` into a tempdir, builds a
+/// [`HarnessSpec`] with the caller's `entry_kind` / `payload_slot`,
+/// then executes it through [`nyx_scanner::dynamic::runner::run_spec`]
+/// directly.  Returns a [`VerifyResult`]-shaped summary so callers can
+/// reuse the same `assert_confirmed` / `assert_not_confirmed` helpers
+/// the older golden-based suite uses.
+///
+/// Bypasses [`verify_finding`] because the public verifier derives the
+/// payload slot from the synthetic Diag's flow steps and always lands
+/// on [`nyx_scanner::dynamic::spec::PayloadSlot::Param`], which the
+/// HTTP / pytest / CLI shapes cannot honour.  Going through the runner
+/// directly lets the test pin the slot the spec under test actually
+/// expects (e.g. [`nyx_scanner::dynamic::spec::PayloadSlot::QueryParam`]
+/// for HTTP routes).
+#[allow(clippy::too_many_arguments)]
+pub fn run_shape_fixture(
+    shape_dir: &str,
+    file: &str,
+    func: &str,
+    cap: Cap,
+    sink_line: u32,
+    entry_kind: EntryKind,
+    payload_slot: nyx_scanner::dynamic::spec::PayloadSlot,
+) -> VerifyResult {
+    use nyx_scanner::dynamic::runner::{run_spec, RunError};
+    use nyx_scanner::dynamic::sandbox::SandboxOptions;
+    use nyx_scanner::dynamic::spec::{HarnessSpec, SpecDerivationStrategy};
+    use nyx_scanner::symbol::Lang;
+
+    let _guard = FIXTURE_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+
+    let fixture_root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests/dynamic_fixtures/python")
+        .join(shape_dir);
+    let fixture_src = fixture_root.join(file);
+
+    let tmp = TempDir::new().expect("create tempdir");
+    let dst = tmp.path().join(file);
+    std::fs::copy(&fixture_src, &dst).expect("copy fixture into tempdir");
+
+    // SAFETY: env mutation is serialised by FIXTURE_LOCK and cleared at end.
+    unsafe {
+        std::env::set_var("NYX_REPRO_BASE", tmp.path().join("repro").to_str().unwrap());
+        std::env::set_var(
+            "NYX_TELEMETRY_PATH",
+            tmp.path().join("events.jsonl").to_str().unwrap(),
+        );
+    }
+
+    let entry_file = dst.to_string_lossy().into_owned();
+    // Per-fixture stable hash so workdir layout / cache key stays
+    // distinct between shapes and between vuln / benign fixtures.
+    let mut digest = blake3::Hasher::new();
+    digest.update(shape_dir.as_bytes());
+    digest.update(b"|");
+    digest.update(file.as_bytes());
+    let spec_hash = format!("{:016x}", {
+        let bytes = digest.finalize();
+        u64::from_le_bytes(bytes.as_bytes()[..8].try_into().unwrap())
+    });
+
+    let spec = HarnessSpec {
+        finding_id: spec_hash.clone(),
+        entry_file: entry_file.clone(),
+        entry_name: func.to_owned(),
+        entry_kind,
+        lang: Lang::Python,
+        toolchain_id: "python-3".into(),
+        payload_slot,
+        expected_cap: cap,
+        constraint_hints: vec![],
+        sink_file: entry_file,
+        sink_line,
+        spec_hash,
+        derivation: SpecDerivationStrategy::FromFlowSteps,
+        stubs_required: vec![],
+    };
+
+    let opts = SandboxOptions::default();
+    let outcome = run_spec(&spec, &opts);
+
+    unsafe {
+        std::env::remove_var("NYX_REPRO_BASE");
+        std::env::remove_var("NYX_TELEMETRY_PATH");
+    }
+
+    // Project the [`RunOutcome`] / [`RunError`] back onto a
+    // [`VerifyResult`] shape so callers can assert against
+    // [`VerifyStatus`] directly without learning the runner's API.
+    match outcome {
+        Ok(run) => {
+            let status = if run.triggered_by.is_some() {
+                VerifyStatus::Confirmed
+            } else if run.oracle_collision {
+                VerifyStatus::Inconclusive
+            } else {
+                VerifyStatus::NotConfirmed
+            };
+            VerifyResult {
+                finding_id: spec.finding_id.clone(),
+                status,
+                triggered_payload: run
+                    .triggered_by
+                    .and_then(|i| run.attempts.get(i))
+                    .map(|a| a.payload_label.to_owned()),
+                reason: None,
+                inconclusive_reason: None,
+                detail: None,
+                attempts: vec![],
+                toolchain_match: None,
+                differential: None,
+            }
+        }
+        Err(RunError::NoPayloadsForCap) => VerifyResult {
+            finding_id: spec.finding_id.clone(),
+            status: VerifyStatus::Unsupported,
+            triggered_payload: None,
+            reason: Some(UnsupportedReason::NoPayloadsForCap),
+            inconclusive_reason: None,
+            detail: None,
+            attempts: vec![],
+            toolchain_match: None,
+            differential: None,
+        },
+        Err(e) => VerifyResult {
+            finding_id: spec.finding_id.clone(),
+            status: VerifyStatus::Inconclusive,
+            triggered_payload: None,
+            reason: None,
+            inconclusive_reason: None,
+            detail: Some(format!("{e:?}")),
+            attempts: vec![],
+            toolchain_match: None,
+            differential: None,
+        },
+    }
+}
+
+/// Phase 12 — golden harness snapshot.
+///
+/// Stages `<shape>/<file>` into a tempdir, builds a [`HarnessSpec`] for
+/// the supplied entry kind / payload slot, emits the per-shape harness
+/// via [`nyx_scanner::dynamic::lang::emit`], and either writes the
+/// resulting source to `<shape>/<file>.golden_harness.py` (under
+/// `NYX_UPDATE_GOLDENS=1`) or diffs against the existing snapshot.  The
+/// emitter is deterministic, so the snapshot doubles as documentation
+/// of the per-shape harness shape.
+#[allow(clippy::too_many_arguments)]
+pub fn run_harness_snapshot(
+    shape_dir: &str,
+    file: &str,
+    func: &str,
+    cap: Cap,
+    sink_line: u32,
+    entry_kind: EntryKind,
+    payload_slot: nyx_scanner::dynamic::spec::PayloadSlot,
+) {
+    use nyx_scanner::dynamic::lang;
+    use nyx_scanner::dynamic::spec::{HarnessSpec, SpecDerivationStrategy};
+    use nyx_scanner::symbol::Lang;
+
+    let _guard = FIXTURE_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+
+    let fixture_root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests/dynamic_fixtures/python")
+        .join(shape_dir);
+    let fixture_src = fixture_root.join(file);
+    let snapshot_path = fixture_root.join(format!("{file}.golden_harness.py"));
+
+    // Stage into tempdir so the spec.entry_file path matches what the
+    // verifier sees at runtime.
+    let tmp = TempDir::new().expect("create tempdir");
+    let dst = tmp.path().join(file);
+    std::fs::copy(&fixture_src, &dst).expect("copy fixture into tempdir");
+    let entry_file = dst.to_string_lossy().into_owned();
+
+    let spec = HarnessSpec {
+        finding_id: "0000000000000001".into(),
+        entry_file: entry_file.clone(),
+        entry_name: func.to_owned(),
+        entry_kind,
+        lang: Lang::Python,
+        toolchain_id: "python-3".into(),
+        payload_slot,
+        expected_cap: cap,
+        constraint_hints: vec![],
+        sink_file: entry_file,
+        sink_line,
+        // Snapshot uses a fixed spec_hash so the emitted source stays
+        // stable; the runner regenerates the real hash at verify time.
+        spec_hash: "snapshotsnapshot".into(),
+        derivation: SpecDerivationStrategy::FromFlowSteps,
+        stubs_required: vec![],
+    };
+
+    let harness = lang::emit(&spec).expect("python emitter must produce a harness");
+
+    // Strip the tempdir prefix so the snapshot is stable across runs.
+    let tmp_prefix = tmp.path().to_string_lossy().into_owned();
+    let normalised = harness
+        .source
+        .replace(&tmp_prefix, "<TMPDIR>")
+        .replace(file, "<ENTRY_FILE>");
+
+    if std::env::var("NYX_UPDATE_GOLDENS").is_ok_and(|v| v == "1") {
+        std::fs::write(&snapshot_path, &normalised).unwrap_or_else(|e| {
+            panic!("write harness snapshot {}: {e}", snapshot_path.display())
+        });
+        return;
+    }
+
+    let expected = std::fs::read_to_string(&snapshot_path).unwrap_or_else(|e| {
+        panic!(
+            "missing harness snapshot {}: {e}\n\
+             current harness source:\n{normalised}\n\
+             rerun with NYX_UPDATE_GOLDENS=1 to seed it.",
+            snapshot_path.display()
+        )
+    });
+
+    if expected != normalised {
+        panic!(
+            "harness snapshot drift for {shape_dir}/{file}:\n\
+             ---- expected ----\n{expected}\n\
+             ---- actual ----\n{normalised}\n\
+             rerun with NYX_UPDATE_GOLDENS=1 if intended."
+        );
+    }
+}
+
 fn make_diag(path: &Path, func: &str, cap: Cap, sink_line: u32) -> Diag {
     let path_str = path.to_string_lossy().into_owned();
     let evidence = Evidence {

tests/dynamic_fixtures/python/async/benign.py

@@ -0,0 +1,22 @@
+"""Phase 12 — async coroutine, benign."""
+import asyncio
+import re
+import subprocess
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+async def run_ping(host):
+    await asyncio.sleep(0)
+    if not _VALID_HOST.fullmatch(host or ""):
+        print("invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/async/vuln.py

@@ -0,0 +1,21 @@
+"""Phase 12 — async coroutine, vulnerable.
+
+`async def` coroutine that shells out with concatenated user input.
+Nyx harness wraps the call in `asyncio.run`.
+"""
+import asyncio
+import subprocess
+
+
+async def run_ping(host):
+    """Vulnerable async coroutine."""
+    await asyncio.sleep(0)
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/async/vuln.py.golden_harness.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 13
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: async coroutine — wrap in asyncio.run.
+import asyncio
+
+try:
+    _coro = _entry_mod.run_ping(payload)
+    _result = asyncio.run(_coro)
+    if _result is not None:
+        try:
+            print(str(_result), flush=True)
+        except Exception:
+            pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/celery/benign.py

@@ -0,0 +1,25 @@
+"""Phase 12 — Celery task, benign."""
+import re
+import subprocess
+
+from celery import Celery
+
+app = Celery("nyx_fixture")
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+@app.task
+def run_job(host):
+    if not _VALID_HOST.fullmatch(host or ""):
+        print("invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/celery/vuln.py

@@ -0,0 +1,25 @@
+"""Phase 12 — Celery task, vulnerable.
+
+Celery's `@app.task` decorator wraps the underlying function on a Task
+object.  Nyx harness reaches the inner callable via `.run` /
+`.__wrapped__` so no broker is required.
+"""
+import subprocess
+
+from celery import Celery
+
+app = Celery("nyx_fixture")
+
+
+@app.task
+def run_job(host):
+    """Vulnerable Celery task body."""
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/celery/vuln.py.golden_harness.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 17
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: Celery task — call underlying function directly (eager).
+
+try:
+    _task = _entry_mod.run_job
+    # Celery tasks expose the underlying function via `.run` (always) and
+    # `.__wrapped__` (when the decorator preserves it).  Prefer the
+    # underlying callable so we don't go through Celery's broker.
+    _fn = getattr(_task, "run", None) or getattr(_task, "__wrapped__", None) or _task
+    _result = _fn(payload)
+    if _result is not None:
+        try:
+            print(str(_result), flush=True)
+        except Exception:
+            pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/cli/benign.py

@@ -0,0 +1,26 @@
+"""Phase 12 — CLI shape, benign."""
+import re
+import subprocess
+import sys
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+def main():
+    host = sys.argv[1] if len(sys.argv) > 1 else ""
+    if not _VALID_HOST.fullmatch(host):
+        print("invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")
+
+
+if __name__ == "__main__":
+    main()

tests/dynamic_fixtures/python/cli/vuln.py

@@ -0,0 +1,26 @@
+"""Phase 12 — CLI shape, vulnerable.
+
+Driven via `if __name__ == "__main__":` — Nyx harness sets
+`sys.argv[1]` to the payload and either calls `main()` or
+`runpy.run_module(..., run_name="__main__")` to fire the guard block.
+"""
+import subprocess
+import sys
+
+
+def main():
+    """Vulnerable: read host from argv[1] and shell out."""
+    host = sys.argv[1] if len(sys.argv) > 1 else ""
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")
+
+
+if __name__ == "__main__":
+    main()

tests/dynamic_fixtures/python/cli/vuln.py.golden_harness.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 14
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: CLI entry — drives `if __name__ == "__main__":` semantics.
+_argv_payload_slot = 0
+_new_argv = ["vuln"]
+for _i in range(_argv_payload_slot):
+    _new_argv.append("")
+_new_argv.append(payload)
+sys.argv = _new_argv
+try:
+    # If module exposes an explicit `main` callable, prefer that.
+    _entry_callable = getattr(_entry_mod, "main", None)
+    if callable(_entry_callable):
+        _result = _entry_callable()
+        if _result is not None:
+            print(str(_result), flush=True)
+    else:
+        # Fall back to re-importing under `__main__` to fire the
+        # `if __name__ == "__main__":` block.
+        import runpy
+        runpy.run_module("vuln", run_name="__main__")
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/django/benign.py

@@ -0,0 +1,21 @@
+"""Phase 12 — Django view, benign."""
+import re
+import subprocess
+
+from django.http import HttpResponse
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+def ping(request):
+    host = request.GET.get("host", "")
+    if not _VALID_HOST.fullmatch(host):
+        return HttpResponse("invalid host")
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return HttpResponse(result.stdout + result.stderr)

tests/dynamic_fixtures/python/django/vuln.py

@@ -0,0 +1,22 @@
+"""Phase 12 — Django view, vulnerable.
+
+Function-based view driven via `django.test.RequestFactory`.  The
+harness configures a minimal Django settings module at runtime so the
+view can be called without a project layout.
+"""
+import subprocess
+
+from django.http import HttpResponse
+
+
+def ping(request):
+    """Vulnerable: query parameter flows to subprocess(shell=True)."""
+    host = request.GET.get("host", "")
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return HttpResponse(result.stdout + result.stderr)

tests/dynamic_fixtures/python/django/vuln.py.golden_harness.py

@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 15
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: Django view — drive via RequestFactory.
+def _nyx_django_setup():
+    import django
+    from django.conf import settings
+    if not settings.configured:
+        settings.configure(
+            DEBUG=False,
+            DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
+            INSTALLED_APPS=["django.contrib.contenttypes", "django.contrib.auth"],
+            ROOT_URLCONF=None,
+            ALLOWED_HOSTS=["*"],
+            SECRET_KEY="nyx-test-key",
+            USE_TZ=True,
+        )
+    django.setup()
+
+_nyx_django_setup()
+from django.test import RequestFactory
+
+_view = getattr(_entry_mod, "ping", None)
+if _view is None:
+    # Try class-based view dispatch: find a class whose lowercased name
+    # matches "ping", instantiate it, and call as_view().
+    for attr in dir(_entry_mod):
+        val = getattr(_entry_mod, attr, None)
+        if isinstance(val, type):
+            try:
+                _view = val.as_view()
+                break
+            except Exception:
+                pass
+if _view is None:
+    print("NYX_DJANGO_VIEW_NOT_FOUND", file=sys.stderr, flush=True)
+    sys.exit(78)
+
+_factory = RequestFactory()
+_path = "/"
+_method = "GET"
+_query = {}
+_data = None
+if "query" == "query":
+    _query["host"] = payload
+elif "query" == "body":
+    _data = payload
+elif "query" == "env":
+    os.environ["host"] = payload
+_factory_method = getattr(_factory, _method.lower(), _factory.get)
+_request = _factory_method(_path, data=_query or _data, content_type="text/plain" if _data else None)
+try:
+    _resp = _view(_request)
+    try:
+        if hasattr(_resp, "render") and not getattr(_resp, "is_rendered", True):
+            _resp.render()
+        _content = getattr(_resp, "content", b"")
+        if isinstance(_content, (bytes, bytearray)):
+            _content = _content.decode("utf-8", "replace")
+        print(_content, flush=True)
+    except Exception:
+        pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/fastapi/benign.py

@@ -0,0 +1,23 @@
+"""Phase 12 — FastAPI route, benign."""
+import re
+import subprocess
+
+from fastapi import FastAPI
+
+app = FastAPI()
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+@app.get("/ping")
+def ping(host: str = ""):
+    if not _VALID_HOST.fullmatch(host):
+        return "invalid host"
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return result.stdout + result.stderr

tests/dynamic_fixtures/python/fastapi/vuln.py

@@ -0,0 +1,23 @@
+"""Phase 12 — FastAPI route, vulnerable.
+
+Nyx harness drives the route through `starlette.testclient.TestClient`
+so the framework's normal request pipeline fires without a real socket.
+"""
+import subprocess
+
+from fastapi import FastAPI
+
+app = FastAPI()
+
+
+@app.get("/ping")
+def ping(host: str = ""):
+    """Vulnerable: query parameter flows to subprocess(shell=True)."""
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return result.stdout + result.stderr

tests/dynamic_fixtures/python/fastapi/vuln.py.golden_harness.py

@@ -0,0 +1,234 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 16
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: FastAPI route — dispatch via starlette.testclient.TestClient.
+def _nyx_resolve_fastapi_app(mod):
+    try:
+        from fastapi import FastAPI
+    except ImportError:
+        return None
+    for n in ("app", "application"):
+        v = getattr(mod, n, None)
+        if isinstance(v, FastAPI):
+            return v
+    for attr in dir(mod):
+        val = getattr(mod, attr, None)
+        if isinstance(val, FastAPI):
+            return val
+    return None
+
+_app = _nyx_resolve_fastapi_app(_entry_mod)
+if _app is None:
+    print("NYX_FASTAPI_APP_NOT_FOUND", file=sys.stderr, flush=True)
+    sys.exit(78)
+
+try:
+    from starlette.testclient import TestClient
+except ImportError:
+    print("NYX_FASTAPI_TESTCLIENT_MISSING", file=sys.stderr, flush=True)
+    sys.exit(79)
+
+_path = None
+for _r in _app.routes:
+    _name = getattr(_r, "name", None)
+    _endpoint = getattr(_r, "endpoint", None)
+    _endpoint_name = getattr(_endpoint, "__name__", None)
+    if _name == "ping" or _endpoint_name == "ping":
+        _path = getattr(_r, "path", None)
+        break
+if _path is None and _app.routes:
+    _path = getattr(_app.routes[0], "path", None)
+if _path is None:
+    print("NYX_FASTAPI_ROUTE_NOT_FOUND", file=sys.stderr, flush=True)
+    sys.exit(80)
+
+# Strip path parameters; replace `{param}` with the payload when used
+# as the path slot, otherwise with "x".
+import re
+if "query" == "path":
+    _path = re.sub(r"\{[^}]+\}", payload, _path, count=1)
+else:
+    _path = re.sub(r"\{[^}]+\}", "x", _path)
+
+_client = TestClient(_app, raise_server_exceptions=False)
+_method = "GET"
+_query = {}
+_body = None
+if "query" == "query":
+    _query["host"] = payload
+elif "query" == "body":
+    _body = payload
+elif "query" == "env":
+    os.environ["host"] = payload
+try:
+    _resp = _client.request(_method, _path, params=_query, content=_body)
+    try:
+        print(_resp.text, flush=True)
+    except Exception:
+        pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/flask/benign.py

@@ -0,0 +1,24 @@
+"""Phase 12 — Flask route, benign."""
+import re
+import subprocess
+
+from flask import Flask, request
+
+app = Flask(__name__)
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+@app.route("/ping", methods=["GET"])
+def ping():
+    host = request.args.get("host", "")
+    if not _VALID_HOST.fullmatch(host):
+        return "invalid host"
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return result.stdout + result.stderr

tests/dynamic_fixtures/python/flask/vuln.py

@@ -0,0 +1,25 @@
+"""Phase 12 — Flask route, vulnerable.
+
+Vulnerable route reads the `host` query parameter and concatenates it
+into a shell command.  Nyx harness reaches the route via
+`app.test_client()` so no real network listener is bound.
+"""
+import subprocess
+
+from flask import Flask, request
+
+app = Flask(__name__)
+
+
+@app.route("/ping", methods=["GET"])
+def ping():
+    """Vulnerable: untrusted query param flows to subprocess(shell=True)."""
+    host = request.args.get("host", "")
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    return result.stdout + result.stderr

tests/dynamic_fixtures/python/flask/vuln.py.golden_harness.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 18
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: Flask route — dispatch via app.test_client().
+def _nyx_resolve_flask_app(mod):
+    from flask import Flask
+    candidates = [getattr(mod, n, None) for n in ("app", "application", "create_app")]
+    for c in candidates:
+        if callable(c) and not isinstance(c, Flask):
+            try:
+                got = c()
+                if isinstance(got, Flask):
+                    return got
+            except TypeError:
+                pass
+        if isinstance(c, Flask):
+            return c
+    for attr in dir(mod):
+        val = getattr(mod, attr, None)
+        if isinstance(val, Flask):
+            return val
+    return None
+
+_app = _nyx_resolve_flask_app(_entry_mod)
+if _app is None:
+    print("NYX_FLASK_APP_NOT_FOUND", file=sys.stderr, flush=True)
+    sys.exit(78)
+
+_route = None
+for _r in _app.url_map.iter_rules():
+    if _r.endpoint == "ping" or _r.endpoint.endswith("." + "ping"):
+        _route = _r
+        break
+if _route is None:
+    # Fall back: any rule will do, but pick the first POST/GET.
+    _rules = list(_app.url_map.iter_rules())
+    _route = _rules[0] if _rules else None
+if _route is None:
+    print("NYX_FLASK_ROUTE_NOT_FOUND", file=sys.stderr, flush=True)
+    sys.exit(79)
+
+_path = _route.rule
+# Strip route parameters; replace `<param>` with payload when used as
+# the path slot, otherwise with "x".
+import re
+if "query" == "path":
+    _path = re.sub(r"<[^>]+>", payload, _path, count=1)
+else:
+    _path = re.sub(r"<[^>]+>", "x", _path)
+
+_client = _app.test_client()
+_method = "GET"
+_query = {}
+_data = None
+if "query" == "query":
+    _query["host"] = payload
+elif "query" == "body":
+    _data = payload
+elif "query" == "env":
+    os.environ["host"] = payload
+try:
+    _resp = _client.open(_path, method=_method, query_string=_query, data=_data)
+    try:
+        print(_resp.get_data(as_text=True), flush=True)
+    except Exception:
+        pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/generic/benign.py

@@ -0,0 +1,28 @@
+"""Phase 12 — generic shape, benign.
+
+Validates the input against a strict allow-list (alphanumerics + dots
+only — RFC-1035 hostname character set) and refuses to shell out when
+the input contains anything outside the allow-list.  The CMDI marker
+substring (`NYX_PWN_CMDI`) never reaches stdout because the function
+returns before any subprocess call when the validation fails.
+"""
+import re
+import subprocess
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+def run_ping(host):
+    """Safe: allow-list validation; refuse and return on mismatch."""
+    if not _VALID_HOST.fullmatch(host or ""):
+        print("invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/generic/vuln.py

@@ -0,0 +1,20 @@
+"""Phase 12 — generic shape, vulnerable.
+
+Module-level function that shells out with user input directly
+concatenated.  Mirrors `cmdi_positive.py` but lives under the per-shape
+fixture tree so the shape detector hits the `Generic` path.
+"""
+import subprocess
+
+
+def run_ping(host):
+    """Vulnerable: user input concatenated into shell command."""
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/generic/vuln.py.golden_harness.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 12
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: generic module-level function.
+
+try:
+    _result = _entry_mod.run_ping(payload)
+    if _result is not None:
+        try:
+            print(str(_result), flush=True)
+        except Exception:
+            pass
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/python/pytest/benign.py

@@ -0,0 +1,22 @@
+"""Phase 12 — pytest shape, benign."""
+import os
+import re
+import subprocess
+
+_VALID_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
+
+
+def test_run_ping():
+    host = os.environ.get("NYX_PAYLOAD", "")
+    if not _VALID_HOST.fullmatch(host):
+        print("invalid host")
+        return
+    result = subprocess.run(
+        ["ping", "-c", "1", host],
+        shell=False,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/pytest/vuln.py

@@ -0,0 +1,22 @@
+"""Phase 12 — pytest shape, vulnerable.
+
+Pytest convention: function name starts with `test_`.  Nyx harness
+injects the payload via the `NYX_PAYLOAD` env var (the same channel
+pytest fixtures typically read from).
+"""
+import os
+import subprocess
+
+
+def test_run_ping():
+    """Vulnerable test: reads host from env, concatenates into shell."""
+    host = os.environ.get("NYX_PAYLOAD", "")
+    result = subprocess.run(
+        "ping -c 1 " + host,
+        shell=True,
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    print(result.stdout)
+    print(result.stderr, end="")

tests/dynamic_fixtures/python/pytest/vuln.py.golden_harness.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Nyx dynamic harness — auto-generated, do not edit."""
+import os
+import sys
+import traceback
+
+# ── Sink-reachability probe (sys.settrace) ────────────────────────────────────
+
+# ── __nyx_probe shim (Phase 06 — Track C.1, Phase 08 — Track C.4 + C.5) ──────
+# Deny-substring list mirrors crate::dynamic::policy::DENY_KEY_SUBSTRINGS; keep
+# in sync when the host-side policy gains new entries.
+_NYX_DENY_SUBSTRINGS = (
+    "TOKEN", "SECRET", "PASSWORD", "PASSWD", "API_KEY", "APIKEY",
+    "PRIVATE_KEY", "CREDENTIAL", "SESSION", "COOKIE", "AUTH", "BEARER",
+    "AWS_ACCESS", "AWS_SESSION", "GH_TOKEN", "GITHUB_TOKEN", "NPM_TOKEN",
+    "PYPI_TOKEN", "DOCKER_PASS",
+)
+_NYX_PAYLOAD_LIMIT = 16 * 1024
+_NYX_REDACTED = "<redacted-by-nyx-policy>"
+
+def __nyx_scrub_env():
+    import os
+    out = {}
+    for k, v in os.environ.items():
+        ku = str(k).upper()
+        if any(n in ku for n in _NYX_DENY_SUBSTRINGS):
+            out[k] = _NYX_REDACTED
+        else:
+            out[k] = v
+    return out
+
+def __nyx_witness(sink_callee, args):
+    import os
+    payload = os.environ.get("NYX_PAYLOAD", "")
+    payload_bytes = payload.encode("utf-8", "replace") if isinstance(payload, str) else bytes(payload)
+    if len(payload_bytes) > _NYX_PAYLOAD_LIMIT:
+        payload_bytes = payload_bytes[:_NYX_PAYLOAD_LIMIT]
+    args_repr = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            args_repr.append("<bytes:%d>" % len(a))
+        else:
+            args_repr.append(str(a))
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        cwd = ""
+    return {
+        "env_snapshot": __nyx_scrub_env(),
+        "cwd": cwd,
+        "payload_bytes": list(payload_bytes),
+        "callee": str(sink_callee),
+        "args_repr": args_repr,
+    }
+
+def __nyx_emit(rec):
+    import os, json
+    p = os.environ.get("NYX_PROBE_PATH")
+    if not p:
+        return
+    try:
+        with open(p, "a") as _f:
+            _f.write(json.dumps(rec) + "\n")
+    except OSError:
+        pass
+
+def __nyx_probe(sink_callee, *args):
+    import os, time
+    serialised = []
+    for a in args:
+        if isinstance(a, (bytes, bytearray)):
+            serialised.append({"kind": "Bytes", "value": list(a)})
+        elif isinstance(a, bool):
+            serialised.append({"kind": "Int", "value": 1 if a else 0})
+        elif isinstance(a, int):
+            serialised.append({"kind": "Int", "value": a})
+        else:
+            serialised.append({"kind": "String", "value": str(a)})
+    rec = {
+        "sink_callee": str(sink_callee),
+        "args": serialised,
+        "captured_at_ns": time.time_ns(),
+        "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+        "kind": {"kind": "Normal"},
+        "witness": __nyx_witness(sink_callee, args),
+    }
+    __nyx_emit(rec)
+
+# Phase 08: sink-site signal handler.  Call __nyx_install_crash_guard before
+# invoking the instrumented sink so a SIGSEGV / SIGABRT / etc. is captured as
+# a Crash probe (with witness) before the process aborts.  The shim re-raises
+# the signal on the default handler after writing so process-level outcome
+# observers (exit_code) still see the death.
+_NYX_SIGNAL_NAMES = {}
+
+def __nyx_install_crash_guard(sink_callee):
+    import signal, os, time
+    catchable = []
+    for nm in ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGFPE", "SIGILL"):
+        s = getattr(signal, nm, None)
+        if s is not None:
+            catchable.append((nm, s))
+            _NYX_SIGNAL_NAMES[s] = nm
+    def _handler(signum, frame):
+        nm = _NYX_SIGNAL_NAMES.get(signum, "SIG?")
+        rec = {
+            "sink_callee": str(sink_callee),
+            "args": [],
+            "captured_at_ns": time.time_ns(),
+            "payload_id": os.environ.get("NYX_PAYLOAD_ID", ""),
+            "kind": {"kind": "Crash", "signal": nm},
+            "witness": __nyx_witness(sink_callee, []),
+        }
+        __nyx_emit(rec)
+        # Reset to default and re-raise so the process actually dies.
+        signal.signal(signum, signal.SIG_DFL)
+        os.kill(os.getpid(), signum)
+    for _nm, s in catchable:
+        try:
+            signal.signal(s, _handler)
+        except (OSError, ValueError):
+            pass
+
+
+_NYX_SINK_FILE = "<TMPDIR>/<ENTRY_FILE>"
+_NYX_SINK_LINE = 14
+_NYX_SINK_HIT = False
+
+def _nyx_tracer(frame, event, arg):
+    global _NYX_SINK_HIT
+    if not _NYX_SINK_HIT and event == "line":
+        fname = frame.f_code.co_filename
+        if fname == _NYX_SINK_FILE or fname.endswith(_NYX_SINK_FILE) or (
+            os.path.basename(fname) == os.path.basename(_NYX_SINK_FILE)
+        ):
+            if _NYX_SINK_LINE <= frame.f_lineno <= _NYX_SINK_LINE + 5:
+                _NYX_SINK_HIT = True
+                print("__NYX_SINK_HIT__", flush=True)
+    return _nyx_tracer
+
+sys.settrace(_nyx_tracer)
+
+# ── Payload loading ────────────────────────────────────────────────────────────
+_payload_raw = os.environb.get(b"NYX_PAYLOAD", b"")
+if not _payload_raw:
+    import base64
+    _payload_b64 = os.environ.get("NYX_PAYLOAD_B64", "")
+    if _payload_b64:
+        _payload_raw = base64.b64decode(_payload_b64)
+try:
+    payload = _payload_raw.decode("utf-8")
+except UnicodeDecodeError:
+    payload = _payload_raw.decode("latin-1")
+
+# ── Entry module import ────────────────────────────────────────────────────────
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ".")
+try:
+    import vuln as _entry_mod
+except ImportError as _e:
+    print(f"NYX_IMPORT_ERROR: {_e}", file=sys.stderr, flush=True)
+    sys.exit(77)
+
+# Shape: pytest function — drive the single test directly.
+os.environ["NYX_PAYLOAD"] = payload
+try:
+    _result = _entry_mod.test_run_ping()
+    if _result is not None:
+        try:
+            print(str(_result), flush=True)
+        except Exception:
+            pass
+except AssertionError as _e:
+    # AssertionError is the typical pytest failure path; observable.
+    print(f"NYX_ASSERT: {_e}", file=sys.stderr, flush=True)
+except SystemExit as _e:
+    sys.exit(_e.code)
+except Exception as _e:
+    print(f"NYX_EXCEPTION: {type(_e).__name__}: {_e}", file=sys.stderr, flush=True)
+
+sys.settrace(None)

tests/dynamic_fixtures/spec_strategies/callgraph_entry_http.rs

@@ -0,0 +1,12 @@
+// Fixture: spec derived via FromCallgraphEntry (rule id matches `*.http.*`,
+// entry point classified as HttpRoute).
+//
+// Phase 12 — Track B added HttpRoute to the Python emitter's SUPPORTED list,
+// so to keep the entry-kind gate test honest the fixture targets Rust, whose
+// emitter still advertises `[EntryKind::Function]` only.
+
+use actix_web::{web, HttpResponse, Responder};
+
+pub async fn echo(query: web::Query<std::collections::HashMap<String, String>>) -> impl Responder {
+    HttpResponse::Ok().body(query.get("q").cloned().unwrap_or_default())
+}

tests/python_fixtures.rs

@@ -14,12 +14,15 @@ mod common;
 #[cfg(feature = "dynamic")]
 mod python_fixture_tests {
     use crate::common::fixture_harness::{
-        run_fixture_and_compare_to_golden, CopyStrategy, FixtureSpec,
+        run_fixture_and_compare_to_golden, run_harness_snapshot, run_shape_fixture,
+        CopyStrategy, FixtureSpec,
     };
     use nyx_scanner::commands::scan::Diag;
+    use nyx_scanner::dynamic::spec::PayloadSlot;
     use nyx_scanner::dynamic::verify::{verify_finding, VerifyOptions};
     use nyx_scanner::evidence::{
-        Confidence, Evidence, FlowStep, FlowStepKind, UnsupportedReason, VerifyStatus,
+        Confidence, EntryKind, Evidence, FlowStep, FlowStepKind, UnsupportedReason,
+        VerifyStatus,
     };
     use nyx_scanner::labels::Cap;
     use nyx_scanner::patterns::{FindingCategory, Severity};
@@ -275,6 +278,328 @@ mod python_fixture_tests {
         }
     }
 
+    // ── Phase 12 — per-shape acceptance ──────────────────────────────────────
+    //
+    // For each shape the suite asserts:
+    //   1. The vuln fixture confirms (oracle fires, sink hit).
+    //   2. The benign fixture does NOT confirm.
+    //   3. The emitted harness source matches the per-shape golden
+    //      snapshot under `tests/dynamic_fixtures/python/<shape>/`.
+    //
+    // Framework-bound shapes (Flask / FastAPI / Django / Celery) skip
+    // with an `eprintln!` when the framework is unimportable in the
+    // host's `python3` (and therefore unavailable to the harness's
+    // built venv without a successful pip install).
+
+    fn python_module_available(module: &'static str) -> bool {
+        std::process::Command::new("python3")
+            .arg("-c")
+            .arg(format!("import {module}"))
+            .output()
+            .map(|o| o.status.success())
+            .unwrap_or(false)
+    }
+
+    fn assert_confirmed(shape: &str, result: &nyx_scanner::evidence::VerifyResult) {
+        assert_eq!(
+            result.status,
+            VerifyStatus::Confirmed,
+            "{shape}/vuln.py: expected Confirmed, got {:?} ({:?})",
+            result.status,
+            result.detail,
+        );
+    }
+
+    fn assert_not_confirmed(shape: &str, result: &nyx_scanner::evidence::VerifyResult) {
+        assert!(
+            matches!(
+                result.status,
+                VerifyStatus::NotConfirmed | VerifyStatus::Inconclusive
+            ),
+            "{shape}/benign.py: expected NotConfirmed (or Inconclusive), got {:?} ({:?})",
+            result.status,
+            result.detail,
+        );
+        // Tighter check: a benign fixture must never light up `Confirmed`.
+        assert_ne!(
+            result.status,
+            VerifyStatus::Confirmed,
+            "{shape}/benign.py: must not confirm",
+        );
+    }
+
+    // ── generic ─────────────────────────────────────────────────────────────
+
+    #[test]
+    fn generic_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "generic", "vuln.py", "run_ping", Cap::CODE_EXEC, 12,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_confirmed("generic", &r);
+    }
+
+    #[test]
+    fn generic_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "generic", "benign.py", "run_ping", Cap::CODE_EXEC, 20,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_not_confirmed("generic", &r);
+    }
+
+    #[test]
+    fn generic_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "generic", "vuln.py", "run_ping", Cap::CODE_EXEC, 12,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+    }
+
+    // ── cli ─────────────────────────────────────────────────────────────────
+
+    #[test]
+    fn cli_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "cli", "vuln.py", "main", Cap::CODE_EXEC, 14,
+            EntryKind::CliSubcommand, PayloadSlot::Argv(0),
+        );
+        assert_confirmed("cli", &r);
+    }
+
+    #[test]
+    fn cli_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "cli", "benign.py", "main", Cap::CODE_EXEC, 11,
+            EntryKind::CliSubcommand, PayloadSlot::Argv(0),
+        );
+        assert_not_confirmed("cli", &r);
+    }
+
+    #[test]
+    fn cli_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "cli", "vuln.py", "main", Cap::CODE_EXEC, 14,
+            EntryKind::CliSubcommand, PayloadSlot::Argv(0),
+        );
+    }
+
+    // ── pytest ──────────────────────────────────────────────────────────────
+
+    #[test]
+    fn pytest_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "pytest", "vuln.py", "test_run_ping", Cap::CODE_EXEC, 14,
+            EntryKind::Function, PayloadSlot::EnvVar("NYX_PAYLOAD".into()),
+        );
+        assert_confirmed("pytest", &r);
+    }
+
+    #[test]
+    fn pytest_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "pytest", "benign.py", "test_run_ping", Cap::CODE_EXEC, 14,
+            EntryKind::Function, PayloadSlot::EnvVar("NYX_PAYLOAD".into()),
+        );
+        assert_not_confirmed("pytest", &r);
+    }
+
+    #[test]
+    fn pytest_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "pytest", "vuln.py", "test_run_ping", Cap::CODE_EXEC, 14,
+            EntryKind::Function, PayloadSlot::EnvVar("NYX_PAYLOAD".into()),
+        );
+    }
+
+    // ── async ───────────────────────────────────────────────────────────────
+
+    #[test]
+    fn async_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "async", "vuln.py", "run_ping", Cap::CODE_EXEC, 13,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_confirmed("async", &r);
+    }
+
+    #[test]
+    fn async_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        let r = run_shape_fixture(
+            "async", "benign.py", "run_ping", Cap::CODE_EXEC, 14,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_not_confirmed("async", &r);
+    }
+
+    #[test]
+    fn async_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "async", "vuln.py", "run_ping", Cap::CODE_EXEC, 13,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+    }
+
+    // ── celery ──────────────────────────────────────────────────────────────
+
+    #[test]
+    fn celery_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("celery") {
+            eprintln!("SKIP: celery not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "celery", "vuln.py", "run_job", Cap::CODE_EXEC, 17,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_confirmed("celery", &r);
+    }
+
+    #[test]
+    fn celery_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("celery") {
+            eprintln!("SKIP: celery not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "celery", "benign.py", "run_job", Cap::CODE_EXEC, 17,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+        assert_not_confirmed("celery", &r);
+    }
+
+    #[test]
+    fn celery_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "celery", "vuln.py", "run_job", Cap::CODE_EXEC, 17,
+            EntryKind::Function, PayloadSlot::Param(0),
+        );
+    }
+
+    // ── flask ───────────────────────────────────────────────────────────────
+
+    #[test]
+    fn flask_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("flask") {
+            eprintln!("SKIP: flask not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "flask", "vuln.py", "ping", Cap::CODE_EXEC, 18,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_confirmed("flask", &r);
+    }
+
+    #[test]
+    fn flask_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("flask") {
+            eprintln!("SKIP: flask not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "flask", "benign.py", "ping", Cap::CODE_EXEC, 17,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_not_confirmed("flask", &r);
+    }
+
+    #[test]
+    fn flask_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "flask", "vuln.py", "ping", Cap::CODE_EXEC, 18,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+    }
+
+    // ── fastapi ─────────────────────────────────────────────────────────────
+
+    #[test]
+    fn fastapi_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("fastapi") {
+            eprintln!("SKIP: fastapi not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "fastapi", "vuln.py", "ping", Cap::CODE_EXEC, 16,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_confirmed("fastapi", &r);
+    }
+
+    #[test]
+    fn fastapi_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("fastapi") {
+            eprintln!("SKIP: fastapi not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "fastapi", "benign.py", "ping", Cap::CODE_EXEC, 16,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_not_confirmed("fastapi", &r);
+    }
+
+    #[test]
+    fn fastapi_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "fastapi", "vuln.py", "ping", Cap::CODE_EXEC, 16,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+    }
+
+    // ── django ──────────────────────────────────────────────────────────────
+
+    #[test]
+    fn django_vuln_is_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("django") {
+            eprintln!("SKIP: django not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "django", "vuln.py", "ping", Cap::CODE_EXEC, 15,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_confirmed("django", &r);
+    }
+
+    #[test]
+    fn django_benign_not_confirmed() {
+        if !python3_available() { eprintln!("SKIP: python3 not available"); return; }
+        if !python_module_available("django") {
+            eprintln!("SKIP: django not importable");
+            return;
+        }
+        let r = run_shape_fixture(
+            "django", "benign.py", "ping", Cap::CODE_EXEC, 14,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+        assert_not_confirmed("django", &r);
+    }
+
+    #[test]
+    fn django_harness_snapshot_matches_golden() {
+        run_harness_snapshot(
+            "django", "vuln.py", "ping", Cap::CODE_EXEC, 15,
+            EntryKind::HttpRoute, PayloadSlot::QueryParam("host".into()),
+        );
+    }
+
     /// Sensitive-filename gate fires before any harness execution; no
     /// python3 needed.
     #[test]

tests/spec_derivation_strategies.rs

@@ -320,14 +320,16 @@ mod spec_strategies {
     /// `Inconclusive(EntryKindUnsupported { lang, attempted, supported, hint })`
     /// rather than `Unsupported`. End-to-end coverage:
     ///   - construct an HttpRoute spec via `derive_from_callgraph_entry`
-    ///     (Python emitter currently advertises `[Function]` only);
+    ///     against a language whose emitter still advertises `[Function]`
+    ///     only (Rust, post Phase 12 — the Python emitter now supports
+    ///     `HttpRoute` and would short-circuit the gate);
     ///   - drive it through `verify_finding`;
     ///   - assert the verdict shape matches the promise.
     #[test]
     fn entry_kind_gate_promotes_unsupported_to_inconclusive_with_hint() {
         let mut diag = make_diag(
-            "py.http.flask_route",
-            "tests/dynamic_fixtures/spec_strategies/callgraph_entry_http.py",
+            "rs.http.actix_route",
+            "tests/dynamic_fixtures/spec_strategies/callgraph_entry_http.rs",
             8,
         );
         let mut ev = Evidence::default();
@@ -357,7 +359,7 @@ mod spec_strategies {
                 supported,
                 hint,
             }) => {
-                assert_eq!(lang, nyx_scanner::symbol::Lang::Python);
+                assert_eq!(lang, nyx_scanner::symbol::Lang::Rust);
                 assert!(matches!(attempted, EntryKind::HttpRoute));
                 assert!(
                     !supported.is_empty(),
@@ -365,7 +367,7 @@ mod spec_strategies {
                 );
                 assert!(
                     supported.contains(&EntryKind::Function),
-                    "Python emitter must advertise Function support; got {supported:?}"
+                    "Rust emitter must advertise Function support; got {supported:?}"
                 );
                 assert!(
                     !hint.is_empty(),