3edef38ba3
ops-readonly gets two read-only questions, waf_blocked and waf_rule_info. The rule ids per blocked request are in CrowdSec's alerts, not in its log, and alerts need machine credentials that can also write, so waf-export.py runs on the host and the container gets one more read-only mount. Workflow G lets a local 3B model draft CRS route exclusions as a Forgejo issue. The model never writes rule syntax, a code node checks every field against the tool output, the only write permission is an issue in one repo, trusted sources live in the workflow and the trigger needs a token. waf-propose.py does the same without a model: 5 of 5 false positives, no exclusion for an attack, 0.03 s. The model proposed an exclusion for nearly every attack it saw. Numbers and the traps we hit are in HARDENING.md.
32 lines
1.5 KiB
Python
32 lines
1.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
waf-propose (episode 14): the same proposal the advisor agent makes, without a model.
|
|
|
|
curl -s -H "x-ops-token: $T" http://ops-readonly:8098/read/waf_blocked | ./waf-propose.py 198.51.100.10
|
|
|
|
Reads the waf_blocked answer on stdin. Arguments: the source IPs you trust
|
|
(office, the box itself). A blocked request counts only if it came from one of
|
|
them, and a route only if it was blocked MIN_COUNT times. No trusted source
|
|
given means no proposal: request text alone proves nothing, the sender wrote it.
|
|
"""
|
|
import collections, json, sys
|
|
|
|
MIN_COUNT = 2
|
|
trusted = set(sys.argv[1:])
|
|
routes = collections.OrderedDict()
|
|
for r in json.load(sys.stdin)["requests"]:
|
|
if r["source_ip"] not in trusted:
|
|
continue
|
|
path = r["uri"].split("?")[0]
|
|
seg = path.strip("/").split("/")
|
|
# ponytail: route = first two path segments. A repo or topic name in there
|
|
# makes the exclusion per repo; widen by hand when that gets tedious.
|
|
prefix = "/" + "/".join(seg[:2]) + "/" if len(seg) > 2 else path
|
|
g = routes.setdefault((r["host"], prefix), {"count": 0, "ids": set()})
|
|
g["count"] += r["count"]
|
|
g["ids"].update(i for i in r["rule_ids"] if i.isdigit())
|
|
|
|
proposals = [{"classification": "false_positive", "host": h, "path": p, "match": "prefix", "rule_ids": sorted(g["ids"]),
|
|
"reason": "blocked %d times from a trusted source" % g["count"]}
|
|
for (h, p), g in routes.items() if g["count"] >= MIN_COUNT and g["ids"]]
|
|
json.dump({"proposals": proposals, "rejected": []}, sys.stdout, indent=1)
|