Spaces:
Sleeping
Sleeping
File size: 8,041 Bytes
b77ed28 3f434eb b77ed28 3f434eb b77ed28 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 |
#!/usr/bin/env python3
import json
import sys
from pathlib import Path
LOG_TABLES = {"email_logs", "auth_logs", "netflow", "process_events", "alerts"}
def _err(msg):
print(f"ERROR: {msg}")
def _load_json(path):
with path.open() as f:
return json.load(f)
def validate_seed(seed):
errors = 0
users = {u["user_id"] for u in seed["entities"]["users"]}
hosts = {h["host_id"] for h in seed["entities"]["hosts"]}
domains = {d["domain"] for d in seed["entities"]["domains"]}
targets = {t["target_id"] for t in seed["entities"]["data_targets"]}
emails = {e["email_id"] for e in seed["seed_artifacts"]["emails"]}
log_templates = {t["template_id"]: t for t in seed["seed_artifacts"]["log_templates"]}
# Top-level references
if seed["patient_zero_host"] not in hosts:
_err("patient_zero_host not in entities.hosts")
errors += 1
if seed["compromised_user"] not in users:
_err("compromised_user not in entities.users")
errors += 1
if seed["attacker_domain"] not in domains:
_err("attacker_domain not in entities.domains")
errors += 1
if seed["data_target"] not in targets:
_err("data_target not in entities.data_targets")
errors += 1
# attack_plan references
ap = seed["attack_plan"]
if ap["phishing_email_id"] not in emails:
_err("attack_plan.phishing_email_id not in seed_artifacts.emails")
errors += 1
if ap["credentials_used"]["user_id"] not in users:
_err("attack_plan.credentials_used.user_id not in entities.users")
errors += 1
for hop in ap["lateral_path"]:
if hop["src_host"] not in hosts:
_err("attack_plan.lateral_path.src_host not in entities.hosts")
errors += 1
if hop["dst_host"] not in hosts:
_err("attack_plan.lateral_path.dst_host not in entities.hosts")
errors += 1
if ap["data_access"]["target_id"] not in targets:
_err("attack_plan.data_access.target_id not in entities.data_targets")
errors += 1
if ap["exfiltration"]["destination_domain"] not in domains:
_err("attack_plan.exfiltration.destination_domain not in entities.domains")
errors += 1
# timeline artifacts (attack_plan timeline + attack_graph initial_artifacts)
artifact_events = list(ap.get("timeline", []))
attack_graph = seed.get("attack_graph")
if attack_graph:
for art in attack_graph.get("initial_artifacts", []):
artifact_events.append({"step": art.get("step", 0), "artifacts": [art]})
for item in artifact_events:
for art in item["artifacts"]:
art_type = art["artifact_type"]
art_id = art["artifact_id"]
if art_type == "email":
if art_id not in emails:
_err("timeline artifact email not in seed_artifacts.emails")
errors += 1
elif art_type == "log_template":
if art_id not in log_templates:
_err("timeline artifact log_template not in seed_artifacts.log_templates")
errors += 1
elif art_type == "alert":
if art_id not in log_templates:
_err("timeline artifact alert not in seed_artifacts.log_templates")
errors += 1
else:
table = log_templates[art_id]["table"]
if table != "alerts":
_err("timeline artifact alert must reference log_template with table=alerts")
errors += 1
variant_action = art.get("variant_action_type")
if variant_action and variant_action not in {
"lateral_move",
"lateral_move_alt",
"exfiltrate",
"exfiltrate_alt",
}:
_err("timeline artifact variant_action_type not allowed")
errors += 1
# prompt injection mapping
injections = seed["prompt_injection_payloads"]
injection_ids = {p["injection_id"] for p in injections}
if len(injection_ids) != len(injections):
_err("prompt_injection_payloads injection_id must be unique")
errors += 1
template_by_injection = {
t.get("injection_id"): t for t in seed["seed_artifacts"]["log_templates"] if t.get("injection_id")
}
timeline_steps = {}
for item in artifact_events:
for art in item["artifacts"]:
if art["artifact_type"] in ("log_template", "alert"):
timeline_steps.setdefault(art["artifact_id"], item["step"])
for p in injections:
inj_id = p["injection_id"]
if p["surface"] == "email":
if not any(e.get("injection_id") == inj_id for e in seed["seed_artifacts"]["emails"]):
_err("email injection_id not referenced by any seed_artifacts.emails")
errors += 1
elif p["surface"] == "alert":
if not any(
t.get("injection_id") == inj_id and t.get("table") == "alerts"
for t in seed["seed_artifacts"]["log_templates"]
):
_err("alert injection_id not referenced by any alerts log_template")
errors += 1
else:
template = template_by_injection.get(inj_id)
if template:
step = timeline_steps.get(template.get("template_id"))
if step is not None:
expected = f"alert-{seed['scenario_id']}-{step}"
evidence_ids = set(p.get("evidence_ids", []))
if expected not in evidence_ids:
_err("alert injection evidence_ids missing expected alert id")
errors += 1
elif p["surface"] == "log":
template = template_by_injection.get(inj_id)
if not template:
_err("log injection_id not referenced by any log_template")
errors += 1
else:
table = template.get("table")
if table not in {"auth_logs", "netflow", "process_events"}:
_err("log injection_id must reference auth_logs/netflow/process_events template")
errors += 1
step = timeline_steps.get(template.get("template_id"))
if step is None:
_err("log injection template_id not present in timeline")
errors += 1
else:
expected = None
if table == "auth_logs":
expected = f"auth-{seed['scenario_id']}-{step}"
elif table == "netflow":
expected = f"flow-{seed['scenario_id']}-{step}"
elif table == "process_events":
expected = f"proc-{seed['scenario_id']}-{step}"
evidence_ids = set(p.get("evidence_ids", []))
if expected and expected not in evidence_ids:
_err("log injection evidence_ids missing expected row id")
errors += 1
# basic log_template table validation
for t in seed["seed_artifacts"]["log_templates"]:
if t["table"] not in LOG_TABLES:
_err(f"log_template table not recognized: {t['table']}")
errors += 1
return errors
def main(argv):
if len(argv) < 2:
candidates = [p for p in Path("data/seeds").glob("*.json") if "ground_truth" not in p.name]
else:
candidates = [Path(p) for p in argv[1:]]
total_errors = 0
for path in candidates:
seed = _load_json(path)
errs = validate_seed(seed)
if errs == 0:
print(f"OK: {path}")
total_errors += errs
if total_errors > 0:
print(f"Validation failed with {total_errors} error(s)")
sys.exit(1)
if __name__ == "__main__":
main(sys.argv)
|