AID-D-005.003: Proactive AI Threat Hunting
This technique covers the proactive, hypothesis-driven search through AI system logs and telemetry for subtle, unknown, or 'low-and-slow' attacks that do not trigger predefined alerts. Threat hunting assumes an attacker may already be present and evading standard detections. It focuses on identifying novel attack patterns, reconnaissance activities, and anomalous behaviors by using exploratory data analysis, complex queries, and machine learning on historical data. The shipped methods cover systematic model-extraction query campaigns, retrieval-to-completion exfiltration signals and instruction-bearing assistant-link hunts. Unsupervised session clustering is triage support, not evidence of training poisoning, model backdoors, supply-chain compromise or membership inference; those require their dedicated detectors and hunt hypotheses.
Pillars: infra, model, app
Lifecycle: operation
Scope boundary
Owns human-led, hypothesis-driven investigation of historical AI telemetry for novel, subtle, low-and-slow, or evasive activity that predefined alerts may miss.
AID-D-005.008: AID-D-005.003 is analyst-led exploratory hunting for unknown or weakly signaled threats; AID-D-005.008 is a bounded read-only agent that prepares evidence for already queued alerts. Threat hunting starts from a hypothesis, while alert triage starts from a specific finding and cannot expand into autonomous response.
AID-D-008: AID-D-005.003 uses human-driven exploratory analysis over telemetry; AID-D-008 trains and serves a versioned fusion model over normalized detector findings. A learned meta-detector is not a substitute for open-ended analyst hypotheses, and hunting results do not become fusion scores without controlled feature and model governance.
Implementation guidance
- AID-D-005.003-G001: Formulate hypotheses based on AI threat models (ATLAS, MAESTRO) and hunt for corresponding TTPs.
- AID-D-005.003-G002: Use clustering to find anomalous user or agent sessions.
- AID-D-005.003-G003: Hunt for data exfiltration patterns in RAG systems.
- AID-D-005.003-G004: Hunt for crafted assistant links carrying persistent-memory or recommendation instructions and hand confirmed findings to the appropriate channel control.
AID-D-005.003-G001
Formulate hypotheses based on AI threat models (ATLAS, MAESTRO) and hunt for corresponding TTPs.
Run a policy-bound Splunk hunt
Use a fixed query template and take only bounded numeric values and an allowlisted index name from the signed hunt policy. The runner snapshots the policy, bundle, and public key before verification, pins the URL Toolbox app version, smoke-tests ut_levenshtein, dispatches a closed-window population count and hunt, paginates every result, and writes a digest-bound result. Do not paste unvalidated policy strings into SPL.
from __future__ import annotations
# File: hunting/run_model_extraction_hunt.py
import argparse
import hashlib
import hmac
import json
import math
import ssl
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import splunklib.client as splunk_client
from splunklib.results import JSONResultsReader
INDEX_NAME = re.compile(r"^[A-Za-z0-9_.-]{1,128}$")
COSIGN = "/usr/local/bin/cosign"
POLICY_KEY = Path("/etc/aidefend/d005003/trust/hunt-policy.pub")
RESULT_KEY = Path("/etc/aidefend/d005003/trust/hunt-result.pub")
SPLUNK_CA = Path("/etc/aidefend/d005003/trust/splunk-enterprise-ca.pem")
# Non-tunable process-safety ceilings (input sizes are bytes); signed policy and callers cannot raise them.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
HARD_INPUT_CEILING = 64 * 1024 * 1024
EXIT_CODES = {
"PASS": 0,
"FAIL": 1,
"INSUFFICIENT_DATA": 2,
"ERROR": 3,
}
def reject_constant(value: str) -> None:
raise ValueError(f"non-finite JSON number is forbidden: {value}")
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
def strict_json(raw: bytes, label: str) -> dict[str, Any]:
value = json.loads(
raw.decode("utf-8"),
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
if not isinstance(value, dict):
raise ValueError(f"{label} must be a JSON object")
return value
def canonical(value: Any) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def required_text(value: Any, label: str) -> str:
if (
not isinstance(value, str)
or not value.strip()
or len(value.strip()) > 2048
):
raise ValueError(f"{label} must be a bounded non-empty string")
return value.strip()
def positive_integer(value: Any, label: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{label} must be a positive integer")
return value
def finite_number(value: Any, label: str, *, allow_zero: bool) -> float:
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or float(value) < 0
or (not allow_zero and float(value) == 0)
):
raise ValueError(f"{label} is outside its allowed range")
return float(value)
def read_stable(
path: Path, *, trust: bool = False, allow_empty: bool = False,
) -> bytes:
requested = Path(os.path.abspath(path))
resolved = path.resolve(strict=True)
if requested != resolved or path.is_symlink():
raise ValueError(f"symlinked or non-canonical input: {path}")
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
try:
before = os.fstat(descriptor)
if (
not stat.S_ISREG(before.st_mode)
or (before.st_size < 1 and not allow_empty)
or before.st_size > HARD_INPUT_CEILING
):
raise ValueError(f"input is not a bounded regular file: {path}")
if trust and (before.st_uid != 0 or before.st_mode & 0o022):
raise PermissionError(f"trust input is not root-owned and protected: {path}")
chunks: list[bytes] = []
while True:
chunk = os.read(descriptor, 1024 * 1024)
if not chunk:
break
chunks.append(chunk)
after = os.fstat(descriptor)
finally:
os.close(descriptor)
identity = lambda item: (
item.st_dev, item.st_ino, item.st_size,
item.st_mtime_ns, item.st_ctime_ns,
)
raw = b"".join(chunks)
if identity(before) != identity(after) or len(raw) != before.st_size:
raise ValueError(f"input changed during capture: {path}")
return raw
def write_private(path: Path, raw: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
try:
view = memoryview(raw)
while view:
written = os.write(descriptor, view)
if written <= 0:
raise OSError("private snapshot write made no progress")
view = view[written:]
os.fsync(descriptor)
finally:
os.close(descriptor)
def capture_policy(
policy_path: Path, bundle_path: Path,
) -> tuple[bytes, str, dict[str, str]]:
policy_raw = read_stable(policy_path)
bundle_raw = read_stable(bundle_path)
key_raw = read_stable(POLICY_KEY, trust=True)
with tempfile.TemporaryDirectory(prefix="aidefend-d005003-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
policy_copy = root / "policy.json"
bundle_copy = root / "policy.bundle.json"
key_copy = root / "policy.pub"
for path, raw in (
(policy_copy, policy_raw),
(bundle_copy, bundle_raw),
(key_copy, key_raw),
):
write_private(path, raw)
subprocess.run(
[
COSIGN, "verify-blob", "--key", str(key_copy),
"--bundle", str(bundle_copy), str(policy_copy),
],
check=True,
capture_output=True,
text=True,
timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
for path, raw in (
(policy_copy, policy_raw),
(bundle_copy, bundle_raw),
(key_copy, key_raw),
):
if not hmac.compare_digest(path.read_bytes(), raw):
raise RuntimeError("private policy verification snapshot changed")
return policy_raw, hashlib.sha256(policy_raw).hexdigest(), {
"bundle_sha256": hashlib.sha256(bundle_raw).hexdigest(),
"verification_key_sha256": hashlib.sha256(key_raw).hexdigest(),
}
def validate_policy(value: dict[str, Any], evaluated_at: datetime) -> dict[str, Any]:
fields = {
"schema_version", "policy_version", "subject", "splunk_index",
"earliest_epoch", "latest_epoch", "index_settle_delay_seconds",
"previous_prompt_window", "minimum_query_count",
"minimum_eligible_event_count", "maximum_prompt_stdev",
"maximum_average_edit_distance", "result_page_size",
"url_toolbox_app_name", "url_toolbox_version",
}
if set(value) != fields:
raise ValueError("hunt-policy field set differs")
if value["schema_version"] != "aidefend.splunk-model-extraction-hunt-policy.v2":
raise ValueError("unsupported hunt-policy schema")
required_text(value["policy_version"], "policy.policy_version")
if not isinstance(value["subject"], dict) or set(value["subject"]) != {
"application_id", "deployment_revision"
}:
raise ValueError("hunt-policy subject field set differs")
for field in ("application_id", "deployment_revision"):
required_text(value["subject"][field], f"policy.subject.{field}")
index_name = required_text(value["splunk_index"], "policy.splunk_index")
if INDEX_NAME.fullmatch(index_name) is None:
raise ValueError("splunk_index is not an allowlisted identifier")
earliest = positive_integer(value["earliest_epoch"], "policy.earliest_epoch")
latest = positive_integer(value["latest_epoch"], "policy.latest_epoch")
if earliest >= latest:
raise ValueError("hunt window is empty or reversed")
settle_delay = positive_integer(
value["index_settle_delay_seconds"],
"policy.index_settle_delay_seconds",
)
if evaluated_at.timestamp() - latest < settle_delay:
raise ValueError("hunt window has not reached its signed index-settle delay")
positive_integer(value["previous_prompt_window"], "policy.previous_prompt_window")
positive_integer(value["minimum_query_count"], "policy.minimum_query_count")
positive_integer(
value["minimum_eligible_event_count"],
"policy.minimum_eligible_event_count",
)
finite_number(
value["maximum_prompt_stdev"],
"policy.maximum_prompt_stdev",
allow_zero=True,
)
finite_number(
value["maximum_average_edit_distance"],
"policy.maximum_average_edit_distance",
allow_zero=False,
)
page_size = positive_integer(value["result_page_size"], "policy.result_page_size")
if page_size > 50000:
raise ValueError("result_page_size exceeds Splunk's configured pagination bound")
required_text(value["url_toolbox_app_name"], "policy.url_toolbox_app_name")
required_text(value["url_toolbox_version"], "policy.url_toolbox_version")
return value
def connect_splunk():
host = required_text(os.environ.get("SPLUNK_HOST"), "SPLUNK_HOST")
token = required_text(
os.environ.get("SPLUNK_SESSION_TOKEN"), "SPLUNK_SESSION_TOKEN"
)
port = positive_integer(int(os.environ.get("SPLUNK_PORT", "8089")), "SPLUNK_PORT")
ca_raw = read_stable(SPLUNK_CA, trust=True)
try:
context = ssl.create_default_context(
cadata=ca_raw.decode("ascii", errors="strict")
)
except (UnicodeDecodeError, ssl.SSLError) as error:
raise ValueError("fixed enterprise CA bundle is invalid") from error
return splunk_client.connect(
host=host,
port=port,
scheme="https",
token=token,
verify=True,
context=context,
)
def collect_job_rows(job, page_size: int) -> list[dict[str, Any]]:
job.refresh()
if str(job.content.get("isDone", "")).lower() not in {"1", "true"}:
raise RuntimeError("Splunk blocking job is not complete")
if str(job.content.get("dispatchState", "")).upper() != "DONE":
raise RuntimeError("Splunk blocking job did not finish successfully")
result_count = int(job.content.get("resultCount", "-1"))
if result_count < 0:
raise ValueError("Splunk job omitted resultCount")
rows: list[dict[str, Any]] = []
for offset in range(0, result_count, page_size):
reader = JSONResultsReader(
job.results(
output_mode="json",
count=min(page_size, result_count - offset),
offset=offset,
)
)
page = [dict(item) for item in reader if isinstance(item, dict)]
if reader.is_preview:
raise RuntimeError("Splunk returned preview rather than final results")
if len(page) != min(page_size, result_count - offset):
raise RuntimeError("Splunk result page is incomplete or contains messages")
rows.extend(page)
if len(rows) != result_count:
raise RuntimeError("Splunk result population differs from resultCount")
return rows
def blocking_job(service, query: str, policy: dict[str, Any]):
return service.jobs.create(
query,
exec_mode="blocking",
earliest_time=str(policy["earliest_epoch"]),
latest_time=str(policy["latest_epoch"]),
)
def run_hunt(service, policy: dict[str, Any]) -> dict[str, Any]:
app = service.apps[policy["url_toolbox_app_name"]]
app.refresh()
if str(app.content.get("disabled", "0")) == "1":
raise RuntimeError("URL Toolbox is disabled")
if str(app.content.get("version", "")) != policy["url_toolbox_version"]:
raise RuntimeError("URL Toolbox version differs from signed policy")
levenshtein_macro = (
chr(96) + "ut_levenshtein(current_prompt, prev_prompt)" + chr(96)
)
smoke_query = (
'| makeresults | eval current_prompt="abc",prev_prompt="abd" '
f'| {levenshtein_macro} '
'| eval distance=tonumber(ut_levenshtein) | table distance'
)
smoke = service.jobs.create(smoke_query, exec_mode="blocking")
smoke_rows = collect_job_rows(smoke, policy["result_page_size"])
if len(smoke_rows) != 1 or float(smoke_rows[0].get("distance", "nan")) <= 0:
raise RuntimeError("ut_levenshtein smoke test failed")
query = (
f'search index={policy["splunk_index"]} event_type="api_inference" '
"| eval current_prompt=tostring('request.prompt') "
"| where isnotnull(user_id) AND isnotnull(current_prompt) "
"AND len(current_prompt)>0 "
"| eval prompt_length=len(current_prompt) "
"| eventstats count AS query_count stdev(prompt_length) AS prompt_stdev BY user_id "
"| eval prompt_stdev=coalesce(prompt_stdev,0) "
"| sort 0 user_id _time _indextime _cd "
f'| streamstats current=f window={policy["previous_prompt_window"]} '
"global=f last(current_prompt) AS prev_prompt BY user_id "
f"| {levenshtein_macro} "
"| eval prompt_distance=tonumber(ut_levenshtein) "
"| stats count AS eligible_event_count max(query_count) AS query_count "
"max(prompt_stdev) AS prompt_stdev avg(prompt_distance) AS avg_edit_dist BY user_id "
"| eventstats sum(eligible_event_count) AS total_eligible_event_count "
f'| eval is_finding=if(query_count>={policy["minimum_query_count"]} '
f'AND prompt_stdev<={policy["maximum_prompt_stdev"]} '
"AND isnotnull(avg_edit_dist) "
f'AND avg_edit_dist<={policy["maximum_average_edit_distance"]},1,0) '
"| appendpipe [ stats max(total_eligible_event_count) AS eligible_event_count "
'| eval eligible_event_count=coalesce(eligible_event_count,0),row_type="summary" ] '
'| where row_type="summary" OR is_finding=1 '
'| eval row_type=coalesce(row_type,"finding") '
'| eval eligible_event_count=if(row_type="summary",eligible_event_count,null()) '
"| table row_type user_id query_count prompt_stdev avg_edit_dist eligible_event_count "
"| sort 0 row_type - query_count user_id"
)
job = blocking_job(service, query, policy)
rows = collect_job_rows(job, policy["result_page_size"])
summaries = [row for row in rows if row.get("row_type") == "summary"]
unexpected = [
row for row in rows if row.get("row_type") not in {"summary", "finding"}
]
if len(summaries) != 1 or unexpected:
raise RuntimeError("single-job output lacks one exact summary row")
eligible_count = int(summaries[0].get("eligible_event_count", "-1"))
if eligible_count < 0:
raise RuntimeError("eligible-event count is invalid")
findings: list[dict[str, Any]] = []
seen_users: set[str] = set()
for number, row in enumerate(
(item for item in rows if item.get("row_type") == "finding"), 1
):
user_id = required_text(row.get("user_id"), f"result:{number}.user_id")
if user_id in seen_users:
raise RuntimeError("Splunk returned a duplicate hunt identity")
seen_users.add(user_id)
query_count = positive_integer(int(row.get("query_count", "0")), "query_count")
prompt_stdev = finite_number(
float(row.get("prompt_stdev", "nan")), "prompt_stdev", allow_zero=True
)
average_distance = finite_number(
float(row.get("avg_edit_dist", "nan")),
"avg_edit_dist",
allow_zero=True,
)
findings.append({
"user_id": user_id,
"query_count": query_count,
"prompt_stdev": prompt_stdev,
"average_edit_distance": average_distance,
})
return {
"eligible_event_count": eligible_count,
"findings": findings,
"hunt_job_sid": job.sid,
"hunt_query_sha256": hashlib.sha256(query.encode()).hexdigest(),
"result_row_count": len(rows),
}
def atomic_write(path: Path, value: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
raw = canonical(value) + b"\n"
descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
try:
with os.fdopen(descriptor, "wb") as handle:
handle.write(raw)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
if read_stable(path) != raw:
raise OSError("result readback differs")
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--policy", type=Path, required=True)
parser.add_argument("--policy-bundle", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
arguments = parser.parse_args()
# Live settle-delay enforcement is anchored only to this host's trusted UTC clock.
# A replay harness tests validate_policy() directly with an injected datetime;
# production callers cannot move the clock through a CLI argument.
evaluated_at = datetime.now(timezone.utc)
try:
policy_raw, policy_digest, policy_evidence = capture_policy(
arguments.policy,
arguments.policy_bundle,
)
policy = validate_policy(strict_json(policy_raw, "hunt policy"), evaluated_at)
hunt = run_hunt(connect_splunk(), policy)
if hunt["findings"]:
status, reason = "FAIL", "model-extraction hunt findings were returned"
elif hunt["eligible_event_count"] < policy["minimum_eligible_event_count"]:
status, reason = "INSUFFICIENT_DATA", "eligible event population is below policy minimum"
else:
status, reason = "PASS", "complete single-job hunt returned no matching identity"
result = {
"schema_version": "aidefend.model_extraction_hunt_result.v2",
"control": "AID-D-005.003",
"status": status,
"status_reason": reason,
"subject": policy["subject"],
"policy_version": policy["policy_version"],
"policy_sha256": policy_digest,
"policy_input_evidence": policy_evidence,
"window_start_epoch": policy["earliest_epoch"],
"window_end_epoch": policy["latest_epoch"],
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
**hunt,
}
except Exception as error:
result = {
"schema_version": "aidefend.model_extraction_hunt_result.v2",
"control": "AID-D-005.003",
"status": "ERROR",
"status_reason": "policy verification, Splunk execution, or result collection failed",
"error_class": type(error).__name__,
"evaluated_at": evaluated_at.isoformat().replace("+00:00", "Z"),
}
atomic_write(arguments.output, result)
return EXIT_CODES[result["status"]]
if __name__ == "__main__":
raise SystemExit(main())
Sign and verify the complete result
#!/usr/bin/env bash
set -euo pipefail
snapshot="$(mktemp -d)"
stage=""
cleanup() {
rm -rf -- "$snapshot"
if test -n "$stage" && test -e "$stage"; then rm -rf -- "$stage"; fi
}
trap cleanup EXIT
chmod 0700 "$snapshot"
set +e
python hunting/run_model_extraction_hunt.py \
--policy policy/model-extraction-hunt.json \
--policy-bundle policy/model-extraction-hunt.sigstore.json \
--output "$snapshot/result.json"
status=$?
set -e
result_key=/etc/aidefend/d005003/trust/hunt-result.pub
test -f "$result_key" && test ! -L "$result_key"
test "$(stat -c %u -- "$result_key")" -eq 0
key_mode="$(stat -c %a -- "$result_key")"
test "$((8#$key_mode & 8#022))" -eq 0
install -m 0400 -- "$result_key" "$snapshot/result.pub"
cosign sign-blob --yes --key env://D005003_RESULT_SIGNING_KEY \
--bundle "$snapshot/result.sigstore.json" "$snapshot/result.json"
chmod 0400 "$snapshot/result.sigstore.json"
cosign verify-blob --key "$snapshot/result.pub" \
--bundle "$snapshot/result.sigstore.json" "$snapshot/result.json"
result_sha="$(sha256sum -- "$snapshot/result.json" | awk '{print $1}')"
bundle_sha="$(sha256sum -- "$snapshot/result.sigstore.json" | awk '{print $1}')"
key_sha="$(sha256sum -- "$snapshot/result.pub" | awk '{print $1}')"
test -d evidence && test ! -L evidence
base=evidence/model-extraction-hunt-results
test ! -L "$base"
if test -e "$base"; then test -d "$base"; else mkdir -m 0750 -- "$base"; fi
stage="$(mktemp -d "$base/.stage.XXXXXX")"
chmod 0700 "$stage"
install -m 0440 -- "$snapshot/result.json" "$stage/result.json"
install -m 0440 -- "$snapshot/result.sigstore.json" "$stage/result.sigstore.json"
install -m 0440 -- "$snapshot/result.pub" "$stage/result.pub"
python - "$stage" "$result_sha" "$bundle_sha" "$key_sha" <<'PY'
import json
import os
import sys
from pathlib import Path
root = Path(sys.argv[1])
value = {
"schema_version": "aidefend.content-addressed-result.v1",
"result_sha256": sys.argv[2],
"bundle_sha256": sys.argv[3],
"verification_key_sha256": sys.argv[4],
}
raw = (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
descriptor = os.open(root / "completion.json", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o440)
with os.fdopen(descriptor, "wb") as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
directory = os.open(root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(directory)
finally:
os.close(directory)
PY
target="$base/$result_sha"
mv -T --no-clobber -- "$stage" "$target"
test ! -e "$stage"
stage=""
test "$result_sha" = "$(sha256sum -- "$target/result.json" | awk '{print $1}')"
test "$bundle_sha" = "$(sha256sum -- "$target/result.sigstore.json" | awk '{print $1}')"
test "$key_sha" = "$(sha256sum -- "$target/result.pub" | awk '{print $1}')"
cosign verify-blob --key "$target/result.pub" \
--bundle "$target/result.sigstore.json" "$target/result.json"
printf '%s\n' "$target"
exit "$status"Install Splunkbase URL Toolbox app 2734 on every search head and pin its internal app name and version in the signed policy. The SDK authenticates with a Splunk session token, builds its TLS context from the captured fixed enterprise CA, and does not auto-login. One blocking search job derives the eligible denominator, summary row, and every finding from the same dispatch; retain its SID, exact query digest, closed time bounds, final result count, policy bundle, and app inventory. A matching identity is FAIL; too few eligible events are INSUFFICIENT_DATA.
Signature, macro, app-version, pagination, TLS, query, or parsing faults are ERROR. Emit PASS only when the single summary row proves a complete policy-sized population and no finding row exists. The local content-addressed directory is create-once staging, not WORM; a separate retention identity reads back and verifies its completion record, result, bundle, and fixed verification key before copying them to append-only storage and replaying the recorded evaluation time.
Action: Calibrate the signed thresholds on deployment traffic, schedule the closed-window job, and send findings to a human threat hunter. This detector does not block users or declare a user malicious.
AID-D-005.003-G002
Use clustering to find anomalous user or agent sessions.
Reusable session-outlier hunt
Build privacy-minimized session aggregates from authenticated events, apply a scaler fitted on a signed benign baseline rather than the live hunt population, and preserve incomplete sessions instead of dividing by zero or dropping them.
from __future__ import annotations
# File: threat_hunting/session_outliers.py
import numpy as np
from sklearn.cluster import DBSCAN
FEATURES = (
"request_count", "mean_input_bytes", "error_rate",
"distinct_model_count", "tool_call_count",
)
def hunt_session_outliers(
sessions: list[dict],
scaler: dict,
policy: dict,
) -> dict:
if not sessions:
return {
"control_status": "INSUFFICIENT_DATA",
"finding_status": "UNKNOWN",
"reason": "no sessions in signed hunt scope",
}
session_ids = [str(item.get("session_ref", "")) for item in sessions]
if any(not value for value in session_ids) or len(session_ids) != len(set(session_ids)):
raise ValueError("session identities missing or duplicated")
incomplete = []
rows = []
row_ids = []
for session in sessions:
missing = [feature for feature in FEATURES if session.get(feature) is None]
if missing:
incomplete.append({"session_ref": session["session_ref"], "missing": missing})
continue
values = [float(session[feature]) for feature in FEATURES]
if not np.isfinite(values).all():
incomplete.append({"session_ref": session["session_ref"], "missing": ["non_finite_feature"]})
continue
rows.append(values)
row_ids.append(session["session_ref"])
if incomplete or len(rows) < int(policy["minimum_complete_sessions"]):
return {
"control_status": "INSUFFICIENT_DATA",
"finding_status": "UNKNOWN",
"session_count": len(sessions),
"complete_session_count": len(rows),
"incomplete_sessions": incomplete,
"policy_version": policy["policy_version"],
}
mean = np.asarray(scaler["mean"], dtype=float)
scale = np.asarray(scaler["scale"], dtype=float)
matrix = np.asarray(rows, dtype=float)
if mean.shape != (len(FEATURES),) or scale.shape != mean.shape or np.any(scale <= 0):
raise ValueError("signed scaler shape or scale invalid")
standardized = (matrix - mean) / scale
labels = DBSCAN(
eps=float(policy["eps"]),
min_samples=int(policy["min_samples"]),
metric="euclidean",
n_jobs=1,
).fit_predict(standardized)
outlier_ids = [
row_ids[index] for index, label in enumerate(labels) if int(label) == -1
]
return {
"schema_version": "aidefend.session_outlier_hunt.v1",
"control_status": "PASS",
"finding_status": "FINDING" if outlier_ids else "NO_FINDING",
"policy_version": policy["policy_version"],
"scaler_version": scaler["scaler_version"],
"session_count": len(sessions),
"complete_session_count": len(rows),
"outlier_session_refs": outlier_ids,
"cluster_labels": {
row_ids[index]: int(label) for index, label in enumerate(labels)
},
}Verification: Replay a signed hunt set with normal clusters, true outliers, one-event sessions, zero requests, missing features, non-finite values, duplicate identities, reordered sessions, and an undersized population. Pin scikit-learn and compare canonical labels.
Action: Hunt on the complete signed session scope using the independently approved scaler and parameters, and route outlier session references for analyst review without exposing raw prompts.
AID-D-005.003-G003
Hunt for data exfiltration patterns in RAG systems.
Seed the hunt from one globally consistent signed identity population
A deployment job first verifies the organization-signed tenant/principal inventory and publishes rag_expected_principal_inventory atomically. The whole lookup must contain one inventory version, one 64-hex manifest digest, one validity start, one validity end, and one row per authenticated tenant/principal pair. The saved search starts from that population and appends observations, so an expected identity with no events cannot disappear. Never derive the denominator from the telemetry being tested.
| inputlookup rag_expected_principal_inventory
| eval row_kind="expected", expected_member=1,
raw_tenant=tenant_id, raw_principal=authenticated_principal_id,
observed_event_id=null(), observed_event_type=null()
| append [ search index=ai_events (event_type="rag_retrieval" OR event_type="agent_goal_complete")
| eval row_kind="observed", expected_member=0,
raw_tenant=tenant_id, raw_principal=authenticated_principal_id,
observed_event_id=event_id, observed_event_type=event_type ]
| addinfo
| eval binding_error=if(isnull(raw_tenant) OR len(trim(tostring(raw_tenant)))=0 OR isnull(raw_principal) OR len(trim(tostring(raw_principal)))=0 OR (row_kind="observed" AND (isnull(observed_event_id) OR len(trim(tostring(observed_event_id)))=0)),1,0)
| eval tenant_key=if(isnull(raw_tenant) OR len(trim(tostring(raw_tenant)))=0,"__MISSING_TENANT__",tostring(raw_tenant)),
principal_key=if(isnull(raw_principal) OR len(trim(tostring(raw_principal)))=0,"__MISSING_PRINCIPAL__",tostring(raw_principal))
| eval expected_identity_key=if(expected_member=1,tenant_key."\u001f".principal_key,null())
| eventstats count(eval(if(expected_member=1,expected_identity_key,null()))) AS inventory_row_count
dc(expected_identity_key) AS expected_population_count
dc(eval(if(expected_member=1,inventory_version,null()))) AS inventory_version_count
dc(eval(if(expected_member=1,inventory_manifest_sha256,null()))) AS inventory_digest_count
dc(eval(if(expected_member=1,inventory_valid_from,null()))) AS inventory_valid_from_count
dc(eval(if(expected_member=1,inventory_valid_until,null()))) AS inventory_valid_until_count
values(eval(if(expected_member=1,inventory_version,null()))) AS global_inventory_version
values(eval(if(expected_member=1,inventory_manifest_sha256,null()))) AS global_inventory_manifest_sha256
min(eval(if(expected_member=1,tonumber(inventory_valid_from),null()))) AS global_inventory_valid_from
max(eval(if(expected_member=1,tonumber(inventory_valid_until),null()))) AS global_inventory_valid_until
max(binding_error) AS global_binding_error
min(info_min_time) AS global_dispatch_start
max(info_max_time) AS global_dispatch_end
| eval global_inventory_error=if(global_binding_error=1 OR expected_population_count<1 OR inventory_row_count!=expected_population_count OR inventory_version_count!=1 OR inventory_digest_count!=1 OR inventory_valid_from_count!=1 OR inventory_valid_until_count!=1 OR NOT match(mvindex(global_inventory_manifest_sha256,0),"^[0-9a-f]{64}$") OR isnull(global_inventory_valid_from) OR isnull(global_inventory_valid_until) OR global_inventory_valid_from>global_dispatch_start OR global_inventory_valid_until<global_dispatch_end,1,0)
| stats max(expected_member) AS expected_member
sum(expected_member) AS expected_row_count
values(inventory_version) AS inventory_version
values(inventory_manifest_sha256) AS inventory_manifest_sha256
values(inventory_valid_from) AS inventory_valid_from
values(inventory_valid_until) AS inventory_valid_until
dc(eval(if(observed_event_type="rag_retrieval",observed_event_id,null()))) AS retrievals
dc(eval(if(observed_event_type="agent_goal_complete",observed_event_id,null()))) AS completed_tasks
dc(eval(if(row_kind="observed",observed_event_id,null()))) AS observed_events
max(binding_error) AS binding_error
max(global_inventory_error) AS global_inventory_error
max(inventory_row_count) AS inventory_row_count
max(expected_population_count) AS expected_population_count
max(inventory_version_count) AS inventory_version_count
max(inventory_digest_count) AS inventory_digest_count
max(inventory_valid_from_count) AS inventory_valid_from_count
max(inventory_valid_until_count) AS inventory_valid_until_count
values(global_inventory_version) AS global_inventory_version
values(global_inventory_manifest_sha256) AS global_inventory_manifest_sha256
max(global_inventory_valid_from) AS global_inventory_valid_from
max(global_inventory_valid_until) AS global_inventory_valid_until
min(global_dispatch_start) AS dispatch_start
max(global_dispatch_end) AS dispatch_end
BY tenant_key principal_key
| lookup rag_abuse_policy tenant_id AS tenant_key OUTPUT policy_version retrieval_minimum maximum_retrieval_to_task_ratio
| lookup rag_event_source_health tenant_id AS tenant_key OUTPUT retrieval_source_healthy completion_source_healthy health_window_start health_window_end
| eval retrieval_minimum=tonumber(retrieval_minimum), maximum_retrieval_to_task_ratio=tonumber(maximum_retrieval_to_task_ratio)
| eval inventory_error=if(global_inventory_error=1 OR expected_member!=1 OR expected_row_count!=1 OR mvcount(inventory_version)!=1 OR mvcount(inventory_manifest_sha256)!=1 OR mvcount(inventory_valid_from)!=1 OR mvcount(inventory_valid_until)!=1 OR inventory_version!=mvindex(global_inventory_version,0) OR inventory_manifest_sha256!=mvindex(global_inventory_manifest_sha256,0) OR tonumber(inventory_valid_from)!=global_inventory_valid_from OR tonumber(inventory_valid_until)!=global_inventory_valid_until,1,0)
| eval policy_error=if(isnull(policy_version) OR isnull(retrieval_minimum) OR retrieval_minimum<=0 OR isnull(maximum_retrieval_to_task_ratio) OR maximum_retrieval_to_task_ratio<=0,1,0)
| eval source_gap=if(coalesce(retrieval_source_healthy,0)!=1 OR coalesce(completion_source_healthy,0)!=1 OR isnull(health_window_start) OR isnull(health_window_end) OR health_window_start>dispatch_start OR health_window_end<dispatch_end,1,0)
| eval retrieval_to_task_ratio=if(completed_tasks=0,null(),retrievals/completed_tasks)
| eval control_status=case(binding_error=1 OR inventory_error=1 OR policy_error=1,"ERROR",source_gap=1 OR observed_events=0,"INSUFFICIENT_DATA",true(),"PASS")
| eval finding_status=case(control_status!="PASS","UNKNOWN",retrievals>=retrieval_minimum AND completed_tasks=0,"FINDING",retrievals>=retrieval_minimum AND retrieval_to_task_ratio>=maximum_retrieval_to_task_ratio,"FINDING",true(),"NO_FINDING")
| eval reason=case(binding_error=1,"identity_or_event_binding_error",inventory_error=1,"expected_inventory_error",policy_error=1,"policy_error",source_gap=1,"missing_source_coverage",observed_events=0,"zero_observed_telemetry",finding_status="FINDING" AND completed_tasks=0,"retrievals_without_completions",finding_status="FINDING","high_retrieval_to_completion_ratio",true(),"complete_no_finding")
| eventstats sum(eval(if(expected_member=1,1,0))) AS expected_principal_row_count
sum(eval(if(expected_member=1 AND observed_events>0,1,0))) AS observed_principal_count
sum(eval(if(control_status="ERROR",1,0))) AS error_count
sum(eval(if(control_status="INSUFFICIENT_DATA",1,0))) AS insufficient_count
sum(eval(if(finding_status="FINDING",1,0))) AS finding_count
| eval row_type="principal"
| appendpipe [ stats max(expected_population_count) AS expected_population_count
max(expected_principal_row_count) AS expected_principal_row_count
max(observed_principal_count) AS observed_principal_count
max(error_count) AS error_count
max(insufficient_count) AS insufficient_count
max(finding_count) AS finding_count
max(global_inventory_error) AS global_inventory_error
max(inventory_row_count) AS inventory_row_count
max(inventory_version_count) AS inventory_version_count
max(inventory_digest_count) AS inventory_digest_count
max(inventory_valid_from_count) AS inventory_valid_from_count
max(inventory_valid_until_count) AS inventory_valid_until_count
values(global_inventory_version) AS inventory_version
values(global_inventory_manifest_sha256) AS inventory_manifest_sha256
max(global_inventory_valid_from) AS inventory_valid_from
max(global_inventory_valid_until) AS inventory_valid_until
| eval row_type="summary" ]
| table row_type tenant_key principal_key expected_member expected_row_count control_status finding_status reason retrievals completed_tasks observed_events retrieval_to_task_ratio policy_version expected_population_count expected_principal_row_count observed_principal_count error_count insufficient_count finding_count global_inventory_error inventory_row_count inventory_version_count inventory_digest_count inventory_valid_from_count inventory_valid_until_count inventory_version inventory_manifest_sha256 inventory_valid_from inventory_valid_until dispatch_start dispatch_end health_window_start health_window_end
| sort 0 row_type tenant_key principal_keyReconcile every expected principal in a zero-result watchdog
from __future__ import annotations
# File: hunting/watch_rag_hunt.py
import hashlib
import json
import math
import os
import re
import tempfile
from pathlib import Path
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SENTINELS = {"__MISSING_TENANT__", "__MISSING_PRINCIPAL__"}
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def one_text(value: object, field: str) -> str:
candidates = list(value) if isinstance(value, (list, tuple)) else str(value or "").splitlines()
candidates = [item.strip() for item in candidates if isinstance(item, str) and item.strip()]
if len(candidates) != 1:
raise ValueError(field + " must contain exactly one value")
return candidates[0]
def integer(row: dict, field: str) -> int:
value = row.get(field)
if isinstance(value, bool):
raise ValueError(field + " is not an integer")
parsed = int(str(value))
if parsed < 0:
raise ValueError(field + " is negative")
return parsed
def inventory_identity(row: dict) -> tuple[str, str, float, float]:
version = one_text(row.get("inventory_version"), "inventory_version")
digest = one_text(row.get("inventory_manifest_sha256"), "inventory_manifest_sha256")
valid_from = float(one_text(row.get("inventory_valid_from"), "inventory_valid_from"))
valid_until = float(one_text(row.get("inventory_valid_until"), "inventory_valid_until"))
if HEX64.fullmatch(digest) is None or not math.isfinite(valid_from) or not math.isfinite(valid_until) or valid_from >= valid_until:
raise ValueError("inventory digest or validity interval is malformed")
return version, digest, valid_from, valid_until
def evaluate_rows(rows: list[dict]) -> dict:
summaries = [row for row in rows if row.get("row_type") == "summary"]
principals = [row for row in rows if row.get("row_type") == "principal"]
unexpected = [row for row in rows if row.get("row_type") not in {"summary", "principal"}]
if not rows or not summaries:
return {"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN", "reason": "zero_rows_or_missing_summary", "summary": {}}
if len(summaries) != 1 or unexpected:
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "duplicate_summary_or_unexpected_row_type", "summary": {}}
summary = summaries[0]
try:
expected = integer(summary, "expected_population_count")
expected_rows = integer(summary, "expected_principal_row_count")
observed = integer(summary, "observed_principal_count")
inventory_rows = integer(summary, "inventory_row_count")
errors = integer(summary, "error_count")
insufficient = integer(summary, "insufficient_count")
findings = integer(summary, "finding_count")
inventory_counts = tuple(integer(summary, field) for field in (
"inventory_version_count", "inventory_digest_count",
"inventory_valid_from_count", "inventory_valid_until_count",
))
global_inventory_error = integer(summary, "global_inventory_error")
summary_inventory = inventory_identity(summary)
except (TypeError, ValueError, OverflowError):
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "malformed_summary_or_inventory", "summary": summary}
if (
global_inventory_error != 0 or inventory_counts != (1, 1, 1, 1)
or inventory_rows != expected or expected_rows != expected
):
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "global_inventory_integrity_failed", "summary": summary}
seen_keys = set()
row_error_count = 0
row_insufficient_count = 0
row_finding_count = 0
row_observed_count = 0
coverage_partial = False
try:
for row in principals:
tenant = one_text(row.get("tenant_key"), "tenant_key")
principal = one_text(row.get("principal_key"), "principal_key")
key = (tenant, principal)
if key in seen_keys or tenant in SENTINELS or principal in SENTINELS:
raise ValueError("principal identity is duplicate or sentinel-bound")
seen_keys.add(key)
if integer(row, "expected_member") != 1 or integer(row, "expected_row_count") != 1:
raise ValueError("principal row is not one expected inventory member")
if inventory_identity(row) != summary_inventory:
raise ValueError("principal inventory differs from summary inventory")
control_status = row.get("control_status")
finding_status = row.get("finding_status")
if control_status not in {"PASS", "INSUFFICIENT_DATA", "ERROR"} or finding_status not in {"FINDING", "NO_FINDING", "UNKNOWN"}:
raise ValueError("principal status is malformed")
if control_status == "PASS" and finding_status not in {"FINDING", "NO_FINDING"}:
raise ValueError("PASS principal has no evaluated finding state")
if control_status != "PASS" and finding_status != "UNKNOWN":
raise ValueError("non-PASS principal claims an evaluated finding state")
observed_events = integer(row, "observed_events")
row_error_count += int(control_status == "ERROR")
row_insufficient_count += int(control_status == "INSUFFICIENT_DATA")
row_finding_count += int(finding_status == "FINDING")
row_observed_count += int(observed_events > 0)
coverage_partial = coverage_partial or observed_events < 1
except (TypeError, ValueError, OverflowError):
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "principal_rows_are_malformed_or_inconsistent", "summary": summary}
if (errors, insufficient, findings, observed) != (row_error_count, row_insufficient_count, row_finding_count, row_observed_count):
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "summary_counts_do_not_match_principal_rows", "summary": summary}
if row_error_count:
return {"control_status": "ERROR", "finding_status": "UNKNOWN", "reason": "principal_or_policy_validation_failed", "summary": summary}
if expected < 1 or observed != expected or len(principals) != expected or row_observed_count != expected:
return {"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN", "reason": "expected_and_observed_principal_counts_do_not_reconcile", "summary": summary}
if coverage_partial or row_insufficient_count:
return {"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN", "reason": "principal_population_or_source_coverage_is_partial", "summary": summary}
return {
"control_status": "PASS",
"finding_status": "FINDING" if row_finding_count else "NO_FINDING",
"reason": "complete_expected_population_evaluated",
"summary": summary,
}
def collect_result(job, *, search_sha256: str, output: Path) -> dict:
from splunklib import results
rows = [dict(row) for row in results.ResultsReader(job.results(count=0)) if isinstance(row, dict)]
state = evaluate_rows(rows)
evidence = {
"schema_version": "aidefend.rag_exfiltration_hunt.v3",
"control": "AID-D-005.003",
"control_status": state["control_status"],
"finding_status": state["finding_status"],
"reason": state["reason"],
"splunk_sid": str(job.sid),
"splunk_result_count": len(rows),
"search_sha256": search_sha256,
"summary": state["summary"],
"rows_sha256": hashlib.sha256(canonical(rows)).hexdigest(),
}
output.parent.mkdir(parents=True, exist_ok=True)
raw = canonical(evidence) + b"\n"
descriptor, temporary = tempfile.mkstemp(prefix=".rag-hunt-", dir=output.parent)
try:
with os.fdopen(descriptor, "wb") as stream:
stream.write(raw); stream.flush(); os.fsync(stream.fileno())
os.replace(temporary, output)
if output.read_bytes() != raw:
raise OSError("evidence readback differs from emitted bytes")
finally:
if os.path.exists(temporary):
os.unlink(temporary)
return evidence
After atomic write, sign the exact result bytes with the detection-workload key, verify the signature with a separately provisioned fixed key, and read the retained object back under an independent identity. The watchdog independently requires exact expected/observed equality, one principal result per expected identity, unique non-sentinel keys, one inventory row per identity, one global inventory version/digest/validity tuple, and exact summary-to-row inventory equality. Partial coverage is INSUFFICIENT_DATA; duplicate keys, mixed inventory, malformed identity/status, signature, or result is ERROR. Neither may become PASS or NO_FINDING.
Regression fixtures for false-green states
# File: tests/test_watch_rag_hunt.py
from hunting.watch_rag_hunt import evaluate_rows
INVENTORY = {
"inventory_version": "inventory-7",
"inventory_manifest_sha256": "a" * 64,
"inventory_valid_from": "100",
"inventory_valid_until": "300",
}
def summary(expected: int, observed: int) -> dict:
return {
"row_type": "summary", "expected_population_count": str(expected),
"expected_principal_row_count": str(expected),
"observed_principal_count": str(observed), "inventory_row_count": str(expected),
"error_count": "0", "insufficient_count": "0", "finding_count": "0",
"inventory_version_count": "1", "inventory_digest_count": "1",
"inventory_valid_from_count": "1", "inventory_valid_until_count": "1",
"global_inventory_error": "0", **INVENTORY,
}
def principal(tenant: str, principal_id: str, *, version: str = "inventory-7") -> dict:
return {
"row_type": "principal", "tenant_key": tenant,
"principal_key": principal_id, "expected_member": "1",
"expected_row_count": "1", "observed_events": "2",
"control_status": "PASS", "finding_status": "NO_FINDING",
**{**INVENTORY, "inventory_version": version},
}
complete = [summary(2, 2), principal("tenant-a", "p1"), principal("tenant-a", "p2")]
partial = [summary(2, 1), principal("tenant-a", "p1")]
null_key = [summary(1, 1), principal("__MISSING_TENANT__", "p1")]
mixed_version = [summary(1, 1), principal("tenant-a", "p1", version="inventory-8")]
malformed = summary(1, 1)
malformed["finding_count"] = "1"
malformed_summary = [malformed, principal("tenant-a", "p1")]
pass_unknown_row = principal("tenant-a", "p1")
pass_unknown_row["finding_status"] = "UNKNOWN"
pass_unknown = [summary(1, 1), pass_unknown_row]
zero = []
assert evaluate_rows(complete)["control_status"] == "PASS"
assert evaluate_rows(partial)["control_status"] == "INSUFFICIENT_DATA"
assert evaluate_rows(null_key)["control_status"] == "ERROR"
assert evaluate_rows(mixed_version)["control_status"] == "ERROR"
assert evaluate_rows(malformed_summary)["control_status"] == "ERROR"
assert evaluate_rows(pass_unknown)["control_status"] == "ERROR"
assert evaluate_rows(zero)["control_status"] == "INSUFFICIENT_DATA"
Also replay confirmed retrieval abuse, no legitimate completions with healthy collectors, late completions, stale or missing collector checkpoints, unexpected identities, duplicate event IDs, policy lookup failure, and a zero-row saved-search failure. Independently query immutable events for every expected tenant/principal pair and reproduce the status. This is a hunt signal: investigate authorization and retrieved-chunk lineage before invoking a response control.
AID-D-005.003-G004
Hunt for crafted assistant links carrying persistent-memory or recommendation instructions and hand confirmed findings to the appropriate channel control.
When this method applies
Use this hunt where the organization has email, collaboration, web gateway, or click telemetry for users of assistant products, including vendor-managed assistants whose UI cannot be changed locally. Adapt the host/parameter catalog to currently deployed assistant versions. Ordinary prefilled prompts are legitimate; a link click or keyword match is not proof of a memory write or successful poisoning. Microsoft's recommendation-poisoning research supplies concrete search hypotheses.
Step 1: Produce bounded candidates
The following KQL is a seed hunt, not a complete injection classifier. It requires both listed Defender tables; missing telemetry must produce an explicit coverage gap rather than an empty clean result. Parse URLs, compare exact normalized hostnames, and inspect the product's known prompt fields. Use a separately reviewed adapter for Safe Links/other wrappers, fragments, repeated parameters, Unicode, and bounded percent-decoding; never fetch an attacker-supplied URL just to decode it. Extend the same method to the supported collaboration telemetry schema.
// Microsoft Defender XDR Advanced Hunting: requires the named tables/licensing.
// Set the lookback from the approved hunting policy before running this query.
let Lookback = 7d;
union isfuzzy=false
(EmailUrlInfo | where Timestamp > ago(Lookback)
| project Timestamp, Url, EvidenceId=tostring(NetworkMessageId), Surface="email"),
(UrlClickEvents | where Timestamp > ago(Lookback)
| project Timestamp, Url, EvidenceId=tostring(NetworkMessageId), Surface="click")
| extend Parsed=parse_url(Url)
| extend Host=tolower(tostring(Parsed.Host)), Query=Parsed["Query Parameters"]
| where Host in ("chatgpt.com", "chat.openai.com", "copilot.microsoft.com")
| extend Candidate=strcat(tostring(Query.q), " ", tostring(Query.prompt))
| where Candidate has_any ("remember", "trusted", "recommend", "prefer")
| project Timestamp, Surface, Host, EvidenceId, PromptSha256=hash_sha256(Candidate)Step 2: Correlate and respond
Review candidates in access-controlled evidence storage, correlate message IDs, click events, user/task context, and available assistant memory/change logs. Assess whether the link asks for durable trust or recommendation changes, private-data reads, or exports; include multilingual and obfuscated cases in validation. Emit the original artifact digest, parsed host, timing, product, and telemetry-coverage limits. Keep raw prompts and user data out of general logs. A confirmed malicious link can be handed to the email/collaboration/web gateway owner for message quarantine or blocking the specific URL/campaign indicator, and to the memory containment owner for suspected persistent writes. Verify the chosen response with a safe replay. Do not block the entire legitimate assistant domain or claim that hunting itself enforced the block.
For a confirmed malicious assistant-link campaign, the deployment-owned mail/SWG policy can block the exact observed URL or quarantined message and revoke its pending delivery. Require verified domain/URL scope and an incident authorization before that response; hunting owns evidence and routing to Isolate, not an automatic organization-wide block based solely on a keyword match.
Open-source tools
- JupyterLab
- pandas
- scikit-learn
- Matplotlib
- OpenSearch Dashboards
- NetworkX
- MISP
- Apache Spark
Commercial / hosted tools
- Splunk Enterprise Security
- Elastic Security
- SentinelOne Purple AI
- Databricks Data Intelligence Platform
- Hex
- Google Threat Intelligence
- Recorded Future Intelligence Cloud
Threat mappings
MITRE ATLAS
- AML.T0024.002 Exfiltration via AI Inference API: Extract AI Model (finding probing patterns)
- AML.T0040 AI Model Inference API Access (finding subtle scanning)
- AML.T0057 LLM Data Leakage (finding low-and-slow exfiltration)
- AML.T0131 Crafted AI Assistant Links (URL and click hunts identify externally supplied assistant prompts for investigation)
- AML.T0130 AI Agent Response Biasing (assistant-link hunts correlate persistent source-trust and recommendation instructions)
MAESTRO
- Model Stealing (L1)
- Data Exfiltration (L2)
- Agent Goal Manipulation (L7) (hunts externally supplied assistant instructions)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure (hypothesis-led hunts identify low-and-slow leakage campaigns)
- LLM05:2026 Data and Model Poisoning (G004 hunts link-delivered attempts to corrupt persistent memory; it does not detect training-data or model-weight poisoning)
- LLM01:2026 Prompt Injection (hunts assistant links carrying externally supplied instructions)
OWASP ML Top 10 2023
- ML05:2023 Model Theft
OWASP Top 10 for Agentic Applications 2026
- ASI06:2026 Memory & Context Poisoning (G004 hunts assistant links requesting persistent memory changes; it does not detect poisoned model weights)
- ASI01:2026 Agent Goal Hijack (G004 hunts externally supplied assistant instructions; a click is triage evidence, not proof of successful hijack)
NIST Adversarial Machine Learning 2025
- NISTAML.031 Model Extraction
- NISTAML.036 Leaking information from user interactions
Cisco Integrated AI Security and Safety Framework
- AISubtech-10.1.1 API Query Stealing (systematic high-similarity query campaigns)
- AITech-10.1 Model Extraction (systematic high-similarity query campaigns)
- AITech-8.2 Data Exfiltration / Exposure (bulk RAG extraction inferred from retrieval-to-completion ratios)
- AITech-1.2 Indirect Prompt Injection (hunts instruction-bearing assistant links distributed through external content)
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (threat hunting discovers novel injection techniques)
- MRE: Model Reverse Engineering (threat hunting discovers query-based model extraction campaigns)
Databricks AI Security Framework 3.0
- Model Serving - Inference response 10.1: Lack of audit and monitoring inference quality
- Model Serving - Inference requests 9.11: Model Inference API Access
- Agents - Core 13.1: Memory Poisoning (hunts links requesting persistent memory changes)