AID-D-005.009: AI-Service C2 & Abuse-Channel Detection
Detect suspicious use of AI service APIs, model-provider endpoints, SaaS AI apps, or agent/MCP service channels as command-and-control, staging, exfiltration, or abuse coordination channels. This sub-technique focuses on behavioral detection over AI-service traffic: beacon-like request cadence, encoded payload patterns, abnormal prompt/completion entropy, unusual API-key or principal usage, and correlations with endpoint, proxy, identity, and model-gateway telemetry.
Pillars: app, infra
Lifecycle: operation
Scope boundary
Owns behavioral detection that AI service APIs, provider endpoints, SaaS AI applications, or agent and MCP channels are being used for command-and-control, staging, exfiltration, or abuse coordination. It does not discover all AI services or enforce network containment.
AID-M-001.004: AID-D-005.009 determines whether observed AI-service traffic exhibits attacker communication or abuse-channel behavior; AID-M-001.004 discovers and reconciles which external AI services are being used. Service existence and ownership are inventory facts, while C2 behavior is a security finding over the traffic.
AID-D-005.007: AID-D-005.009 detects communication cadence, encoded payload shape, entropy, and identity correlation consistent with C2 or staging; AID-D-005.007 detects broader token, parameter, volume, latency, and cost anomalies. Resource abuse can support a C2 finding but does not by itself establish attacker communication.
AID-I-002.002: AID-D-005.009 emits AI-service C2 and abuse-channel findings; AID-I-002.002 restricts or isolates external AI service connectivity. Detection identifies suspicious use, while network and service admission enforcement blocks or limits the channel.
Implementation guidance
- AID-D-005.009-G001: Normalize AI-service traffic into a detector-ready schema with principal, provider, model, request cadence, byte counts, token counts, and payload-shape indicators.
- AID-D-005.009-G002: Detect beacon-like cadence, encoded payload shape, and abnormal prompt/completion entropy in AI-service traffic.
- AID-D-005.009-G003: Correlate AI-service C2 findings with identity, endpoint, proxy, and API-key telemetry to create an investigation-ready alert.
AID-D-005.009-G001
Normalize AI-service traffic into a detector-ready schema with principal, provider, model, request cadence, byte counts, token counts, and payload-shape indicators.
Concept:
AI-service C2 detection requires consistent fields across model gateways, proxy logs, CASB/SWG telemetry, and endpoint events. Do not store raw sensitive prompts by default; store hashes, sizes, entropy, token counts, provider identity, and stable trace IDs so detectors can reason about cadence and payload shape.
Step 1: Define detector-ready telemetry
from __future__ import annotations
# File: detection/ai_service_channel_event.py
import hashlib
import math
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
def sha256_hex(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def shannon_entropy(value: str) -> float:
if not value:
return 0.0
counts = {char: value.count(char) for char in set(value)}
return -sum((count / len(value)) * math.log2(count / len(value)) for count in counts.values())
@dataclass(frozen=True)
class AiServiceChannelEvent:
observed_at: str
provider: str
model_or_service: str
principal_id: str
api_key_id_hash: str | None
source_ip: str | None
user_agent: str | None
trace_id: str
request_bytes: int
response_bytes: int
prompt_sha256: str
prompt_entropy: float
completion_entropy: float
payload_shape_flags: list[str]
token_count: int | None
tool_call_count: int
@staticmethod
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def to_event(self) -> dict:
return {"event_type": "ai_service_channel", **asdict(self)}
Step 2: Convert gateway records into channel events
from __future__ import annotations
# File: detection/normalize_model_gateway.py
import json
import sys
from detection.ai_service_channel_event import AiServiceChannelEvent, sha256_hex, shannon_entropy
def normalize(record: dict) -> dict:
prompt = str(record.get("prompt_preview") or "")
completion = str(record.get("completion_preview") or "")
api_key_id = record.get("api_key_id")
payload_shape_flags = []
compact_prompt = "".join(prompt.split())
if len(compact_prompt) >= 80 and all(char.isalnum() or char in "+/=_-" for char in compact_prompt):
payload_shape_flags.append("base64_like_prompt")
if prompt.count("\n") >= 20:
payload_shape_flags.append("many_short_lines")
event = AiServiceChannelEvent(
observed_at=record.get("timestamp") or AiServiceChannelEvent.now_iso(),
provider=record["provider"],
model_or_service=record.get("model", "unknown"),
principal_id=record["principal_id"],
api_key_id_hash=sha256_hex(api_key_id) if api_key_id else None,
source_ip=record.get("source_ip"),
user_agent=record.get("user_agent"),
trace_id=record["trace_id"],
request_bytes=int(record.get("request_bytes") or len(prompt.encode("utf-8"))),
response_bytes=int(record.get("response_bytes") or len(completion.encode("utf-8"))),
prompt_sha256=sha256_hex(prompt),
prompt_entropy=round(shannon_entropy(prompt), 4),
completion_entropy=round(shannon_entropy(completion), 4),
payload_shape_flags=payload_shape_flags,
token_count=record.get("token_count"),
tool_call_count=int(record.get("tool_call_count") or 0),
)
return event.to_event()
for line in sys.stdin:
print(json.dumps(normalize(json.loads(line)), sort_keys=True))
Verify the detector and retain evidence
from __future__ import annotations
# File: detection/verify_ai_channel_schema.py
import json
from pathlib import Path
REQUIRED = {"provider", "principal_id", "trace_id", "request_bytes", "response_bytes", "prompt_sha256"}
events = [json.loads(line) for line in Path("out/ai_service_channel_events.jsonl").read_text(encoding="utf-8").splitlines()]
missing = [
{"trace_id": event.get("trace_id"), "missing": sorted(field for field in REQUIRED if event.get(field) in {None, ""})}
for event in events
if any(event.get(field) in {None, ""} for field in REQUIRED)
]
artifact = {
"schema_version": "aidefend.ai_service_channel.schema_check.v1",
"event_count": len(events),
"missing_required_field_count": len(missing),
"missing_required_fields": missing[:100],
"control_status": "PASS" if events and not missing else "FAIL",
"finding_status": "NOT_EVALUATED",
}
Path("artifacts").mkdir(exist_ok=True)
Path("artifacts/ai-service-channel-schema-evidence.json").write_text(
json.dumps(artifact, indent=2, sort_keys=True),
encoding="utf-8",
)
if artifact["control_status"] != "PASS":
raise SystemExit("AI service channel schema verification failed")
Action: Feed normalized AI-service channel events to your SIEM. The evidence artifact for this step is artifacts/ai-service-channel-schema-evidence.json, proving detector input completeness without storing raw prompt bodies.
Verify safely
An independent verifier must query network, gateway and provider ranges independently, join by workload, endpoint, timing and request IDs, and compare normalized principal, cadence, byte, token, payload-shape, error and missing-event fields, read evidence back, verify signatures, digests, immutability and effective privileges, and compare the canonical result digest.
AID-D-005.009-G002
Detect beacon-like cadence, encoded payload shape, and abnormal prompt/completion entropy in AI-service traffic.
Concept:
AI-service C2 often looks like normal API use at the destination layer. Detection must inspect behavior: periodic request intervals, repeated payload sizes, high-entropy prompt fragments, base64-like payloads, and prompt/response pairs that look like command envelopes rather than user tasks.
Step 1: Compute cadence and payload-shape signals
from __future__ import annotations
# File: detection/ai_service_c2_detector.py
import hashlib
import json
import math
import os
import subprocess
from collections import defaultdict
from datetime import datetime
from pathlib import Path
import hmac
import re
import tempfile
from statistics import mean, pstdev
def _aidefend_write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(
path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600,
)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _aidefend_strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def _aidefend_capture_signed_bytes(
payload_path: Path,
proof_path: Path,
key_reference: str,
timeout_seconds: float,
proof_flag: str,
) -> bytes:
if payload_path == proof_path or proof_flag != "--bundle":
raise RuntimeError("signed payload/proof inputs are invalid")
payload_raw = payload_path.read_bytes()
proof_raw = proof_path.read_bytes()
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
)) or key_reference.startswith("pkcs11:")
key_raw = None if remote_key else (
key_reference.encode("utf-8")
if key_reference.startswith("-----BEGIN ")
else Path(key_reference).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
proof_snapshot = root / "proof"
_aidefend_write_private(payload_snapshot, payload_raw)
_aidefend_write_private(proof_snapshot, proof_raw)
captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
key_argument = key_reference
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_aidefend_write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
[
"cosign", "verify-blob", "--key", key_argument,
"--bundle", str(proof_snapshot), str(payload_snapshot),
],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
if any(
not hmac.compare_digest(path.read_bytes(), expected)
for path, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for path, _ in captured:
os.chmod(path, 0o600)
return payload_raw
def load_signed_c2_policy() -> tuple[dict, str]:
policy_path = Path(os.environ["D005009_C2_POLICY_PATH"])
bundle_path = Path(os.environ["D005009_C2_POLICY_SIGSTORE_BUNDLE"])
verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(verify_timeout) or verify_timeout <= 0:
raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
raw = _aidefend_capture_signed_bytes(
policy_path, bundle_path, os.environ["D005009_C2_POLICY_VERIFY_KEY"],
verify_timeout, "--bundle",
)
policy = _aidefend_strict_json(raw)
required = {
"schema_version", "policy_version", "minimum_intervals",
"minimum_interval_seconds", "maximum_interval_seconds",
"maximum_interval_cv", "minimum_group_events",
"minimum_repeated_size_ratio", "minimum_high_entropy_events",
"prompt_entropy_threshold", "minimum_base64_events",
"maximum_size_cv", "periodic_weight", "repeated_size_weight",
"high_entropy_weight", "base64_weight", "low_variance_weight",
"finding_score_threshold", "maximum_trace_ids",
}
if not isinstance(policy, dict) or set(policy) != required:
raise RuntimeError("signed C2 detection policy fields differ")
if (
policy["schema_version"] != "aidefend.ai-service-c2-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
):
raise RuntimeError("signed C2 detection policy identity is invalid")
integer_fields = (
"minimum_intervals", "minimum_group_events",
"minimum_high_entropy_events", "minimum_base64_events",
"periodic_weight", "repeated_size_weight", "high_entropy_weight",
"base64_weight", "low_variance_weight", "finding_score_threshold",
"maximum_trace_ids",
)
for field in integer_fields:
value = policy[field]
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise RuntimeError("signed C2 integer is invalid: " + field)
for field in (
"minimum_interval_seconds", "maximum_interval_seconds",
"prompt_entropy_threshold",
):
value = policy[field]
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or float(value) <= 0
):
raise RuntimeError("signed C2 bound is invalid: " + field)
for field in (
"maximum_interval_cv", "minimum_repeated_size_ratio", "maximum_size_cv",
):
value = policy[field]
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or not 0 <= float(value) <= 1
):
raise RuntimeError("signed C2 ratio is invalid: " + field)
if policy["maximum_interval_seconds"] < policy["minimum_interval_seconds"]:
raise RuntimeError("signed C2 interval range is inverted")
maximum_score = sum(policy[field] for field in (
"periodic_weight", "repeated_size_weight", "high_entropy_weight",
"base64_weight", "low_variance_weight",
))
if policy["finding_score_threshold"] > maximum_score:
raise RuntimeError("signed C2 score threshold exceeds attainable score")
return policy, "sha256:" + hashlib.sha256(raw).hexdigest()
POLICY, POLICY_SHA256 = load_signed_c2_policy()
def coefficient_of_variation(values: list[float]) -> float | None:
if len(values) < 2:
return None
average = mean(values)
return None if average == 0 else pstdev(values) / average
def parse_observed_at(value: str) -> float:
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
def inter_arrival_seconds(ordered_events: list[dict]) -> list[float]:
timestamps = [parse_observed_at(event["observed_at"]) for event in ordered_events]
intervals = [
later - earlier
for earlier, later in zip(timestamps, timestamps[1:])
]
if any(interval < 0 for interval in intervals):
raise ValueError("C2 event population is not chronologically valid")
return intervals
def validate_event(event: dict) -> None:
required = {
"principal_id", "provider", "observed_at", "request_bytes",
"prompt_entropy", "payload_shape_flags", "trace_id",
}
if not isinstance(event, dict) or not required.issubset(event):
raise ValueError("C2 event schema is incomplete")
if any(
not isinstance(event[field], str) or not event[field]
for field in ("principal_id", "provider", "observed_at", "trace_id")
):
raise ValueError("C2 event identity is invalid")
request_bytes = event["request_bytes"]
prompt_entropy = event["prompt_entropy"]
if (
isinstance(request_bytes, bool)
or not isinstance(request_bytes, int)
or request_bytes < 0
or isinstance(prompt_entropy, bool)
or not isinstance(prompt_entropy, (int, float))
or not math.isfinite(float(prompt_entropy))
or not isinstance(event["payload_shape_flags"], list)
or any(not isinstance(flag, str) for flag in event["payload_shape_flags"])
):
raise ValueError("C2 event measurements are invalid")
parse_observed_at(event["observed_at"])
def detect_c2_patterns(events: list[dict]) -> dict:
if len({event.get("trace_id") for event in events}) != len(events):
raise ValueError("C2 event population contains duplicate trace IDs")
grouped: dict[tuple[str, str], list[dict]] = defaultdict(list)
for event in events:
validate_event(event)
grouped[(event["principal_id"], event["provider"])].append(event)
findings = []
unevaluated_groups = []
for (principal_id, provider), group in grouped.items():
ordered = sorted(group, key=lambda event: parse_observed_at(event["observed_at"]))
if len(ordered) < POLICY["minimum_group_events"]:
unevaluated_groups.append({
"principal_id": principal_id,
"provider": provider,
"event_count": len(ordered),
})
continue
request_sizes = [event["request_bytes"] for event in ordered]
high_entropy = [
event for event in ordered
if float(event["prompt_entropy"]) >= float(POLICY["prompt_entropy_threshold"])
]
base64ish = [
event for event in ordered
if "base64_like_prompt" in set(event["payload_shape_flags"])
]
size_cv = coefficient_of_variation(request_sizes)
repeated_size_ratio = 1 - len(set(request_sizes)) / len(request_sizes)
intervals = inter_arrival_seconds(ordered)
interval_cv = coefficient_of_variation(intervals)
mean_interval = mean(intervals) if intervals else None
score = 0
reasons = []
if (
len(intervals) >= POLICY["minimum_intervals"]
and mean_interval is not None
and interval_cv is not None
and float(POLICY["minimum_interval_seconds"])
<= mean_interval
<= float(POLICY["maximum_interval_seconds"])
and interval_cv <= float(POLICY["maximum_interval_cv"])
):
score += POLICY["periodic_weight"]
reasons.append("periodic_request_cadence")
if repeated_size_ratio >= float(POLICY["minimum_repeated_size_ratio"]):
score += POLICY["repeated_size_weight"]
reasons.append("repeated_request_size")
if len(high_entropy) >= POLICY["minimum_high_entropy_events"]:
score += POLICY["high_entropy_weight"]
reasons.append("high_entropy_prompts")
if len(base64ish) >= POLICY["minimum_base64_events"]:
score += POLICY["base64_weight"]
reasons.append("base64_like_payloads")
if size_cv is not None and size_cv <= float(POLICY["maximum_size_cv"]):
score += POLICY["low_variance_weight"]
reasons.append("low_size_variance")
if score >= POLICY["finding_score_threshold"]:
findings.append({
"finding_type": "possible_ai_service_c2_channel",
"principal_id": principal_id,
"provider": provider,
"event_count": len(ordered),
"score": score,
"reasons": reasons,
"mean_inter_arrival_seconds": mean_interval,
"inter_arrival_cv": interval_cv,
"sample_trace_ids": [
event["trace_id"]
for event in ordered[:POLICY["maximum_trace_ids"]]
],
})
return {
"unevaluated_groups": unevaluated_groups,
"findings": findings,
"control_status": "INSUFFICIENT_DATA" if not events or unevaluated_groups else "PASS",
"finding_status": (
"FINDING" if findings
else "UNKNOWN" if not events or unevaluated_groups
else "NO_FINDING"
),
}
from dataclasses import dataclass
from typing import Sequence
DETECTOR_RESULT_FIELDS = {
"schema_version", "control", "tenant_id", "source_id",
"window_start_unix", "window_end_unix", "source_artifact_sha256",
"upstream_control_status", "policy_version", "policy_sha256",
"expected_population_count", "observed_population_count",
"expected_population_root_sha256", "observed_population_root_sha256",
"population_reconciled", "unevaluated_groups", "findings",
"control_status", "finding_status", "detector_result_sha256",
}
HEX64 = re.compile(r"^[0-9a-f]{64}$")
@dataclass(frozen=True)
class VerifiedEventPopulation:
tenant_id: str
source_id: str
window_start_unix: float
window_end_unix: float
source_artifact_sha256: str
upstream_control_status: str
expected_trace_ids: tuple[str, ...]
expected_population_root_sha256: str
events: tuple[dict, ...]
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def population_root(trace_ids: Sequence[str]) -> str:
return hashlib.sha256(canonical(sorted(trace_ids))).hexdigest()
def finalize_detector_result(value: dict) -> dict:
if "detector_result_sha256" in value:
raise ValueError("result digest must be computed by the producer")
value["detector_result_sha256"] = hashlib.sha256(canonical(value)).hexdigest()
if set(value) != DETECTOR_RESULT_FIELDS:
raise ValueError("detector-result envelope schema mismatch")
return value
def build_detector_result(population: VerifiedEventPopulation) -> dict:
expected = tuple(population.expected_trace_ids)
events = list(population.events)
observed = [event.get("trace_id") for event in events]
if (
not population.tenant_id or not population.source_id
or not math.isfinite(population.window_start_unix)
or not math.isfinite(population.window_end_unix)
or population.window_start_unix >= population.window_end_unix
or HEX64.fullmatch(population.source_artifact_sha256) is None
or population.upstream_control_status not in {"PASS", "INSUFFICIENT_DATA", "ERROR"}
or not expected or len(expected) != len(set(expected))
or any(not isinstance(item, str) or not item for item in expected)
or HEX64.fullmatch(population.expected_population_root_sha256) is None
or population_root(expected) != population.expected_population_root_sha256
):
raise ValueError("verified event-population receipt is malformed or misbound")
for event in events:
validate_event(event)
observed_at = parse_observed_at(event["observed_at"])
if not population.window_start_unix <= observed_at < population.window_end_unix:
raise ValueError("event is outside the verified detector window")
observed_root = population_root(observed)
reconciled = (
bool(events) and len(observed) == len(set(observed))
and set(observed) == set(expected)
and observed_root == population.expected_population_root_sha256
)
base = {
"schema_version": "aidefend.ai-service-c2-detector-result.v2",
"control": "AID-D-005.009",
"tenant_id": population.tenant_id,
"source_id": population.source_id,
"window_start_unix": population.window_start_unix,
"window_end_unix": population.window_end_unix,
"source_artifact_sha256": population.source_artifact_sha256,
"upstream_control_status": population.upstream_control_status,
"policy_version": POLICY["policy_version"],
"policy_sha256": POLICY_SHA256,
"expected_population_count": len(expected),
"observed_population_count": len(observed),
"expected_population_root_sha256": population.expected_population_root_sha256,
"observed_population_root_sha256": observed_root,
"population_reconciled": reconciled,
}
if population.upstream_control_status == "ERROR":
return finalize_detector_result({
**base, "unevaluated_groups": [], "findings": [],
"control_status": "ERROR", "finding_status": "UNKNOWN",
})
if population.upstream_control_status != "PASS" or not reconciled:
return finalize_detector_result({
**base, "unevaluated_groups": [], "findings": [],
"control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN",
})
detector = detect_c2_patterns(events)
findings = []
for item in detector["findings"]:
identity = {
"tenant_id": population.tenant_id,
"principal_id": item["principal_id"],
"provider": item["provider"],
"score": float(item["score"]),
"observed_at_unix": population.window_end_unix,
"reasons": item["reasons"],
"sample_trace_ids": item["sample_trace_ids"],
"evidence_ref": "sha256:" + population.source_artifact_sha256,
}
findings.append({
"finding_id": hashlib.sha256(canonical(identity)).hexdigest(), **identity,
})
return finalize_detector_result({
**base,
"unevaluated_groups": detector["unevaluated_groups"],
"findings": findings,
"control_status": detector["control_status"],
"finding_status": detector["finding_status"],
})Publish one verified detector-result envelope
The source adapter must verify the event-population artifact signature before constructing VerifiedEventPopulation. Canonicalize the returned aidefend.ai-service-c2-detector-result.v2 object, write it atomically, sign the exact bytes, verify the signature with a separately provisioned key, and read it back before publication. The envelope binds source, tenant, closed window, policy, source artifact digest, upstream control state, expected and observed population counts and roots, reconciliation state, findings, and its own canonical digest. An empty event list, missing trace, extra trace, duplicate trace, upstream INSUFFICIENT_DATA, or unreconciled root returns INSUFFICIENT_DATA/UNKNOWN; an upstream or producer failure returns ERROR/UNKNOWN. None may become PASS/NO_FINDING.
Verify safely
Replay complete benign and beacon populations plus empty, one-missing, one-extra, duplicate, reordered, stale-window, wrong-source-digest, upstream-error, undersized-group, encoded-shape, jittered-beacon, and signature-failure fixtures. An independent runner verifies the result signature and recomputes the exact event population, roots, features, finding IDs, envelope digest, and control/finding states from immutable source bytes.
Regression fixture for a finding plus an unevaluated group
# File: tests/test_ai_service_c2_population.py
import detection.ai_service_c2_detector as detector
def event(trace_id: str, principal: str, second: int) -> dict:
return {
"principal_id": principal, "provider": "provider-a",
"observed_at": f"2026-01-01T00:00:{second:02d}Z",
"request_bytes": 128, "prompt_entropy": 7.5,
"payload_shape_flags": ["base64_like_prompt"], "trace_id": trace_id,
}
def test_mixed_population_preserves_findings(monkeypatch):
monkeypatch.setattr(detector, "POLICY", {
"policy_version": "fixture-1", "minimum_group_events": 3,
"minimum_intervals": 2, "minimum_interval_seconds": 1,
"maximum_interval_seconds": 20, "maximum_interval_cv": 0.1,
"minimum_repeated_size_ratio": 0.5, "minimum_high_entropy_events": 3,
"prompt_entropy_threshold": 7, "minimum_base64_events": 3,
"maximum_size_cv": 0.1, "periodic_weight": 1,
"repeated_size_weight": 1, "high_entropy_weight": 1,
"base64_weight": 1, "low_variance_weight": 1,
"finding_score_threshold": 3, "maximum_trace_ids": 5,
})
events = [
event("a1", "finding-principal", 0),
event("a2", "finding-principal", 10),
event("a3", "finding-principal", 20),
event("b1", "unevaluated-principal", 5),
]
result = detector.detect_c2_patterns(events)
assert result["findings"] and result["unevaluated_groups"]
assert result["control_status"] == "INSUFFICIENT_DATA"
assert result["finding_status"] == "FINDING"
population = detector.VerifiedEventPopulation(
tenant_id="tenant-a", source_id="gateway-a",
window_start_unix=1767225600.0, window_end_unix=1767225660.0,
source_artifact_sha256="a" * 64, upstream_control_status="PASS",
expected_trace_ids=tuple(item["trace_id"] for item in events),
expected_population_root_sha256=detector.population_root(
tuple(item["trace_id"] for item in events)
),
events=tuple(events),
)
envelope = detector.build_detector_result(population)
assert envelope["control_status"] == "INSUFFICIENT_DATA"
assert envelope["finding_status"] == "FINDING"Incomplete group coverage remains INSUFFICIENT_DATA, while verified findings from measured groups remain FINDING and are routed to AID-D-005.002 even if correlation sources are unavailable. Publish the coverage gap alongside the primary findings; never label a partially measured window NO_FINDING. ERROR in the primary detector cannot produce trusted findings.
AID-D-005.009-G003
Correlate AI-service C2 findings with identity, endpoint, proxy, and API-key telemetry to create an investigation-ready alert.
Consume the exact signed G002 detector-result envelope
The correlation adapter verifies the G002 result signature and immutable artifact digest before constructing VerifiedDetectorEnvelope. The consumer below repeats the exact producer field set and canonical-digest rule; it does not accept a bare findings array. Source, tenant, window, upstream state, expected and observed populations, roots, policy, and result digest therefore remain bound through the detector chain.
from __future__ import annotations
# File: detection/correlate_ai_service_c2.py
import hashlib
import json
import math
import re
from dataclasses import dataclass
from typing import Mapping
HEX64 = re.compile(r"^[0-9a-f]{64}$")
DETECTOR_RESULT_FIELDS = {
"schema_version", "control", "tenant_id", "source_id",
"window_start_unix", "window_end_unix", "source_artifact_sha256",
"upstream_control_status", "policy_version", "policy_sha256",
"expected_population_count", "observed_population_count",
"expected_population_root_sha256", "observed_population_root_sha256",
"population_reconciled", "unevaluated_groups", "findings",
"control_status", "finding_status", "detector_result_sha256",
}
FINDING_FIELDS = {
"finding_id", "tenant_id", "principal_id", "provider", "score",
"observed_at_unix", "reasons", "sample_trace_ids", "evidence_ref",
}
@dataclass(frozen=True)
class VerifiedDetectorEnvelope:
artifact_sha256: str
value: dict
@dataclass(frozen=True)
class VerifiedSnapshot:
source_id: str
tenant_id: str
coverage_start_unix: float
coverage_end_unix: float
artifact_sha256: str
upstream_control_status: str
events: tuple[dict, ...]
@dataclass(frozen=True)
class CorrelationPolicy:
policy_version: str
required_source_ids: tuple[str, ...]
correlation_window_seconds: int
critical_score_threshold: float
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def validate_detector_envelope(
envelope: VerifiedDetectorEnvelope,
*, tenant_id: str, window_start_unix: float, window_end_unix: float,
) -> dict:
value = dict(envelope.value)
if set(value) != DETECTOR_RESULT_FIELDS:
raise ValueError("G002 detector-result envelope schema mismatch")
claimed_digest = value.pop("detector_result_sha256")
observed_digest = hashlib.sha256(canonical(value)).hexdigest()
value["detector_result_sha256"] = claimed_digest
expected_count = value["expected_population_count"]
observed_count = value["observed_population_count"]
if (
HEX64.fullmatch(envelope.artifact_sha256) is None
or value["schema_version"] != "aidefend.ai-service-c2-detector-result.v2"
or value["control"] != "AID-D-005.009"
or value["tenant_id"] != tenant_id
or value["window_start_unix"] != window_start_unix
or value["window_end_unix"] != window_end_unix
or HEX64.fullmatch(value["source_artifact_sha256"]) is None
or value["upstream_control_status"] not in {"PASS", "INSUFFICIENT_DATA", "ERROR"}
or not isinstance(value["policy_version"], str) or not value["policy_version"]
or re.fullmatch(r"sha256:[0-9a-f]{64}", value["policy_sha256"]) is None
or isinstance(expected_count, bool) or not isinstance(expected_count, int)
or isinstance(observed_count, bool) or not isinstance(observed_count, int)
or expected_count < 1 or observed_count < 0
or HEX64.fullmatch(value["expected_population_root_sha256"]) is None
or HEX64.fullmatch(value["observed_population_root_sha256"]) is None
or not isinstance(value["population_reconciled"], bool)
or not isinstance(value["unevaluated_groups"], list)
or not isinstance(value["findings"], list)
or value["control_status"] not in {"PASS", "INSUFFICIENT_DATA", "ERROR"}
or value["finding_status"] not in {"FINDING", "NO_FINDING", "UNKNOWN"}
or HEX64.fullmatch(claimed_digest) is None
or claimed_digest != observed_digest
):
raise ValueError("G002 detector-result envelope is malformed or misbound")
complete = (
value["upstream_control_status"] == "PASS"
and expected_count == observed_count
and observed_count > 0
and value["population_reconciled"] is True
and value["expected_population_root_sha256"] == value["observed_population_root_sha256"]
and not value["unevaluated_groups"]
)
if value["control_status"] == "PASS" and not complete:
raise ValueError("G002 returned PASS over an incomplete population")
if value["control_status"] == "ERROR" and (value["finding_status"] != "UNKNOWN" or value["findings"]):
raise ValueError("failed producer cannot supply verified findings")
if value["control_status"] == "INSUFFICIENT_DATA" and value["finding_status"] != (
"FINDING" if value["findings"] else "UNKNOWN"
):
raise ValueError("partial coverage must preserve measured findings")
if value["control_status"] == "PASS" and value["finding_status"] != (
"FINDING" if value["findings"] else "NO_FINDING"
):
raise ValueError("detector finding state disagrees with finding population")
for finding in value["findings"]:
if set(finding) != FINDING_FIELDS:
raise ValueError("G002 finding schema mismatch")
return valuePropagate upstream state before correlation
# Continuing in detection/correlate_ai_service_c2.py
def correlate_detector_result(
*, tenant_id: str, detector: VerifiedDetectorEnvelope,
snapshots: Mapping[str, VerifiedSnapshot],
window_start_unix: float, window_end_unix: float,
policy: CorrelationPolicy,
) -> dict:
result = validate_detector_envelope(
detector, tenant_id=tenant_id,
window_start_unix=window_start_unix, window_end_unix=window_end_unix,
)
if result["control_status"] != "PASS":
return {
"schema_version": "aidefend.ai_service_c2_correlation.v3",
"control_status": (
"ERROR" if result["control_status"] == "ERROR"
or result["upstream_control_status"] == "ERROR"
else "INSUFFICIENT_DATA"
),
"finding_status": result["finding_status"],
"primary_findings": result["findings"],
"policy_version": policy.policy_version,
"detector_result_sha256": result["detector_result_sha256"],
"detector_artifact_sha256": detector.artifact_sha256,
"reason": "G002 detector population is not completely evaluated",
}
expected_sources = set(policy.required_source_ids)
if (
not tenant_id or not policy.policy_version or not expected_sources
or len(expected_sources) != len(policy.required_source_ids)
or set(snapshots) != expected_sources
or not math.isfinite(window_start_unix) or not math.isfinite(window_end_unix)
or window_start_unix >= window_end_unix
or isinstance(policy.correlation_window_seconds, bool)
or policy.correlation_window_seconds < 1
or not math.isfinite(float(policy.critical_score_threshold))
or float(policy.critical_score_threshold) < 0
):
raise ValueError("signed correlation source population or policy is invalid")
coverage_start = window_start_unix - policy.correlation_window_seconds
coverage_end = window_end_unix + policy.correlation_window_seconds
indexed: dict[tuple[str, str], list[dict]] = {}
snapshot_digests = {}
for source_id, snapshot in snapshots.items():
if snapshot.upstream_control_status == "ERROR":
return {
"schema_version": "aidefend.ai_service_c2_correlation.v3",
"control_status": "ERROR", "finding_status": result["finding_status"],
"primary_findings": result["findings"],
"policy_version": policy.policy_version, "source_id": source_id,
"reason": "correlation source reported ERROR",
}
if (
snapshot.upstream_control_status != "PASS"
or snapshot.source_id != source_id or snapshot.tenant_id != tenant_id
or not math.isfinite(snapshot.coverage_start_unix)
or not math.isfinite(snapshot.coverage_end_unix)
or snapshot.coverage_start_unix > coverage_start
or snapshot.coverage_end_unix < coverage_end
or HEX64.fullmatch(snapshot.artifact_sha256) is None
):
return {
"schema_version": "aidefend.ai_service_c2_correlation.v3",
"control_status": "INSUFFICIENT_DATA", "finding_status": result["finding_status"],
"primary_findings": result["findings"],
"policy_version": policy.policy_version, "source_id": source_id,
"reason": "correlation source is incomplete, stale, or misbound",
}
snapshot_digests[source_id] = snapshot.artifact_sha256
seen = set()
for event in snapshot.events:
required = {"evidence_id", "tenant_id", "principal_id", "observed_at_unix", "evidence_ref"}
if not required.issubset(event):
raise ValueError("correlation event schema is incomplete")
observed_at = float(event["observed_at_unix"])
if (
event["tenant_id"] != tenant_id or not event["principal_id"]
or not event["evidence_id"] or event["evidence_id"] in seen
or not math.isfinite(observed_at)
or not snapshot.coverage_start_unix <= observed_at <= snapshot.coverage_end_unix
):
raise ValueError("correlation event is duplicated, malformed, or misbound")
seen.add(event["evidence_id"])
indexed.setdefault((source_id, event["principal_id"]), []).append(event)
alerts, seen_findings = [], set()
for finding in result["findings"]:
finding_id = finding["finding_id"]
principal_id = finding["principal_id"]
score = float(finding["score"])
observed_at = float(finding["observed_at_unix"])
if (
finding["tenant_id"] != tenant_id or not finding_id
or finding_id in seen_findings or not principal_id
or not math.isfinite(score) or score < 0
or not window_start_unix <= observed_at <= window_end_unix
):
raise ValueError("G002 finding is duplicated, malformed, or misbound")
seen_findings.add(finding_id)
related = []
for source_id in policy.required_source_ids:
for event in indexed.get((source_id, principal_id), []):
if abs(float(event["observed_at_unix"]) - observed_at) <= policy.correlation_window_seconds:
related.append({
"source_id": source_id, "evidence_id": event["evidence_id"],
"evidence_ref": event["evidence_ref"],
})
alerts.append({
"finding_id": finding_id, "tenant_id": tenant_id,
"principal_id": principal_id, "provider": finding["provider"],
"score": score,
"severity": "critical" if score >= policy.critical_score_threshold and related else "high" if related else "medium",
"reasons": finding["reasons"],
"sample_trace_ids": finding["sample_trace_ids"],
"primary_evidence_ref": finding["evidence_ref"],
"related_evidence": related,
})
return {
"schema_version": "aidefend.ai_service_c2_correlation.v3",
"control_status": "PASS",
"finding_status": "FINDING" if alerts else "NO_FINDING",
"policy_version": policy.policy_version,
"window_start_unix": window_start_unix,
"window_end_unix": window_end_unix,
"detector_source_id": result["source_id"],
"detector_source_artifact_sha256": result["source_artifact_sha256"],
"detector_result_sha256": result["detector_result_sha256"],
"detector_artifact_sha256": detector.artifact_sha256,
"expected_population_count": result["expected_population_count"],
"observed_population_count": result["observed_population_count"],
"snapshot_sha256": snapshot_digests,
"evaluated_finding_count": len(result["findings"]),
"alerts": alerts,
}Publish and independently replay
Signature-verification adapters are deployment dependencies and must return verified bytes plus measured digests; they must not accept a request-provided trusted flag. Atomically write and sign the correlation result, read it back under a separate identity, and reconstruct it from the signed G002 envelope and all policy-required source snapshots. Replay empty G002 events, missing/extra trace IDs, mismatched roots or counts, forged result digest, upstream ERROR, upstream INSUFFICIENT_DATA, bare-findings input, missing source, stale coverage, duplicate evidence, threshold boundaries, and related/unrelated principals. An empty or unreconciled detector population remains INSUFFICIENT_DATA/UNKNOWN; a producer, signature, schema, or digest failure is ERROR/UNKNOWN; neither may become a clean correlation. Route verified alerts to AID-D-005.002; response execution remains in Isolate/Evict.
Regression fixture for G003 propagation
# File: tests/test_ai_service_c2_correlation_state.py
import hashlib
from detection.correlate_ai_service_c2 import (
CorrelationPolicy, VerifiedDetectorEnvelope, canonical,
correlate_detector_result, validate_detector_envelope,
)
def mixed_envelope() -> VerifiedDetectorEnvelope:
root = "b" * 64
finding = {
"finding_id": "c" * 64, "tenant_id": "tenant-a",
"principal_id": "principal-a", "provider": "provider-a", "score": 5.0,
"observed_at_unix": 20.0, "reasons": ["periodic_request_cadence"],
"sample_trace_ids": ["trace-1"], "evidence_ref": "sha256:" + "d" * 64,
}
value = {
"schema_version": "aidefend.ai-service-c2-detector-result.v2",
"control": "AID-D-005.009", "tenant_id": "tenant-a",
"source_id": "gateway-a", "window_start_unix": 10.0,
"window_end_unix": 30.0, "source_artifact_sha256": "d" * 64,
"upstream_control_status": "PASS", "policy_version": "fixture-1",
"policy_sha256": "sha256:" + "e" * 64,
"expected_population_count": 4, "observed_population_count": 4,
"expected_population_root_sha256": root,
"observed_population_root_sha256": root,
"population_reconciled": True,
"unevaluated_groups": [{"principal_id": "principal-b"}],
"findings": [finding], "control_status": "INSUFFICIENT_DATA",
"finding_status": "FINDING",
}
value["detector_result_sha256"] = hashlib.sha256(canonical(value)).hexdigest()
return VerifiedDetectorEnvelope(artifact_sha256="f" * 64, value=value)
def test_mixed_result_validates_and_propagates():
detector = mixed_envelope()
validated = validate_detector_envelope(
detector, tenant_id="tenant-a", window_start_unix=10.0, window_end_unix=30.0,
)
assert validated["control_status"] == "INSUFFICIENT_DATA"
assert validated["finding_status"] == "FINDING"
propagated = correlate_detector_result(
tenant_id="tenant-a", detector=detector, snapshots={},
window_start_unix=10.0, window_end_unix=30.0,
policy=CorrelationPolicy(
policy_version="fixture-1", required_source_ids=("endpoint",),
correlation_window_seconds=5, critical_score_threshold=4.0,
),
)
assert propagated["control_status"] == "INSUFFICIENT_DATA"
assert propagated["finding_status"] == "FINDING"Incomplete group coverage remains INSUFFICIENT_DATA, while verified findings from measured groups remain FINDING and are routed to AID-D-005.002 even if correlation sources are unavailable. Publish the coverage gap alongside the primary findings; never label a partially measured window NO_FINDING. ERROR in the primary detector cannot produce trusted findings.
Open-source tools
- OpenSearch
- Elasticsearch 8.16+ AGPLv3 source build
- Sigma CLI
- pySigma
- Zeek
- RITA (Active Countermeasures)
- Suricata
- OpenTelemetry
- Vector
- Fluent Bit
Commercial / hosted tools
- Splunk Enterprise Security
- Microsoft Sentinel
- Google Security Operations
- Datadog Cloud SIEM
- Palo Alto Networks Cortex XDR
- Palo Alto Networks Cortex XSIAM
- Netskope One AI Security
- Zscaler AI Security
Threat mappings
MITRE ATLAS
- AML.T0096 AI Service API
- AML.T0085 Data from AI Services
- AML.T0085.001 Data from AI Services: AI Agent Tools
- AML.T0086 Exfiltration via AI Agent Tool Invocation
- AML.T0025 Exfiltration via Cyber Means
- AML.T0114 AI Service Web Interface
- AML.T0072 Cyber Communication Channel (behavioral detection of AI-service traffic used as a command-and-control or relay channel)
MAESTRO
- Data Exfiltration (L2)
- Evasion of Detection (L5)
- Integration Risks (L7)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure
OWASP ML Top 10 2023
- N/A (AI-service C2 detection is primarily an operational monitoring control)
OWASP Top 10 for Agentic Applications 2026
- ASI02:2026 Tool Misuse and Exploitation
- ASI03:2026 Identity and Privilege Abuse
- ASI10:2026 Rogue Agents
NIST Adversarial Machine Learning 2025
- NISTAML.039 Compromising connected resources
- NISTAML.038 Data Extraction
- NISTAML.036 Leaking information from user interactions
Cisco Integrated AI Security and Safety Framework
- AITech-18.2 Malicious Workflows (correlated beacon cadence and encoded-payload network signals)
- AITech-8.2 Data Exfiltration / Exposure (possible encoded AI-service exfiltration channels from beacon and payload-shape correlation; not proof of sensitive content)
Google Secure AI Framework 2.0 - Risks
- SDD: Sensitive Data Disclosure
- IIC: Insecure Integrated Component
- RA: Rogue Actions
Databricks AI Security Framework 3.0
- Agents - Core 13.2: Tool Misuse
- Agents - Core 13.3: Privilege Compromise
- Agents - Core 13.13: Rogue Agents in Multi-Agent Systems