From 2eb47ec8552274305e8e1f1811eec1d6329c3788 Mon Sep 17 00:00:00 2001 From: Admin <62771363+GoDeskio@users.noreply.github.com> Date: Sun, 1 Nov 2026 01:44:49 +0800 Subject: [PATCH] CMO: compliance evidence pack (#36) --- tools/compliance/compliance.py | 60 +++++++++++++++++++++++++++++ tools/compliance/test_compliance.py | 30 +++++++++++++++ 2 files changed, 90 insertions(+) create mode 100644 tools/compliance/compliance.py create mode 100644 tools/compliance/test_compliance.py diff --git a/tools/compliance/compliance.py b/tools/compliance/compliance.py new file mode 100644 index 00000000000..957584b1d84 --- /dev/null +++ b/tools/compliance/compliance.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Compliance pack: automated evidence for SOC 2 / HIPAA technical controls (plan item 36). + +Checks a valkey.conf (and optional live CONFIG GET output) against the +controls below and writes a JSON evidence record. Process controls (policies, +access reviews, vendor management) are listed as manual and need an owner. +This produces evidence; it is not an audit or a certification. +""" +import datetime +import hashlib +import json +import sys + +CHECKS = [ + # id, control refs, description, conf key, predicate + ("ENC-TRANSIT", "SOC2 CC6.7; HIPAA 164.312(e)(1)", "TLS on and plaintext port off", + lambda c: c.get("tls-port", "0") != "0" and c.get("port") == "0"), + ("ENC-REPL", "SOC2 CC6.7", "Replication over TLS", lambda c: c.get("tls-replication") == "yes"), + ("AUTH-DEFAULT", "SOC2 CC6.1; HIPAA 164.312(d)", "Default user disabled or ACL file used", + lambda c: "aclfile" in c or c.get("user default", "").startswith("off")), + ("AUTH-PROTECTED", "SOC2 CC6.6", "Protected mode on", lambda c: c.get("protected-mode", "yes") == "yes"), + ("AUDIT-LOG", "SOC2 CC7.2; HIPAA 164.312(b)", "Command log / slowlog enabled", + lambda c: int(c.get("slowlog-max-len", "128")) > 0), + ("BACKUP", "SOC2 A1.2; HIPAA 164.308(a)(7)", "AOF persistence on", lambda c: c.get("appendonly") == "yes"), + ("DANGEROUS-CMDS", "SOC2 CC6.8", "Debug/module commands not enabled", + lambda c: c.get("enable-debug-command", "no") == "no" and c.get("enable-module-command", "no") == "no"), +] +MANUAL = ["Security policies approved (CC1, CC5)", "Quarterly access reviews (CC6.2)", + "Vendor/BAA register (HIPAA 164.308(b))", "Incident response test (CC7.4)", + "Risk assessment (CC3; HIPAA 164.308(a)(1))", "Employee security training (CC1.4)"] + + +def parse_conf(text): + c = {} + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split(None, 1) + k = parts[0].lower() + v = parts[1].strip().strip('"') if len(parts) > 1 else "" + if k == "user": + name, _, rest = v.partition(" ") + c["user " + name] = rest + else: + c[k] = v + return c + + +def evaluate(conf_text, now=None): + c = parse_conf(conf_text) + res = [{"id": i, "controls": refs, "check": d, "pass": bool(p(c))} for i, refs, d, p in CHECKS] + return {"generated": (now or datetime.datetime.now(datetime.timezone.utc)).isoformat(), + "config_sha256": hashlib.sha256(conf_text.encode()).hexdigest(), + "automated": res, "passed": sum(r["pass"] for r in res), "total": len(res), + "manual": [{"control": m, "status": "needs owner"} for m in MANUAL]} + + +if __name__ == "__main__": + print(json.dumps(evaluate(open(sys.argv[1]).read()), indent=2)) diff --git a/tools/compliance/test_compliance.py b/tools/compliance/test_compliance.py new file mode 100644 index 00000000000..7274c421e45 --- /dev/null +++ b/tools/compliance/test_compliance.py @@ -0,0 +1,30 @@ +import unittest +import compliance as C + +GOOD = """port 0 +tls-port 6380 +tls-replication yes +aclfile /etc/valkey/users.acl +appendonly yes +""" + + +class T(unittest.TestCase): + def test_good(self): + r = C.evaluate(GOOD) + self.assertEqual(r["passed"], r["total"]) + self.assertEqual(len(r["config_sha256"]), 64) + self.assertTrue(all(m["status"] == "needs owner" for m in r["manual"])) + + def test_bad(self): + r = C.evaluate("port 6379\nprotected-mode no\nenable-debug-command yes\nuser default on nopass ~* +@all\n") + failed = {x["id"] for x in r["automated"] if not x["pass"]} + self.assertTrue({"ENC-TRANSIT", "AUTH-DEFAULT", "AUTH-PROTECTED", "BACKUP", "DANGEROUS-CMDS"} <= failed) + + def test_default_user_off(self): + r = C.evaluate("user default off\n") + self.assertTrue(next(x for x in r["automated"] if x["id"] == "AUTH-DEFAULT")["pass"]) + + +if __name__ == "__main__": + unittest.main()