From 51ba69a54137b3a9e1e8150d4bbd7bd2011bacc4 Mon Sep 17 00:00:00 2001 From: leetcrypt Date: Wed, 22 Jul 2026 09:04:06 -0700 Subject: [PATCH] flock_tap: audit-integrity + data-minimization for cloud-vs-local verdicts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes the passive tap a defensible auditing instrument for the "does this camera phone home to Flock?" question, while shrinking what it retains about bystanders. Nothing here widens third-party capture. - Unified verdict vocabulary. A single private _classify() folds flow-stat and device-level signals into exactly one of CLOUD_CONNECTED / LOCAL_STATION / INDETERMINATE. classify_device/classify_camera become thin wrappers. Previously the two classifiers returned divergent sets (OFFLINE_OR_UNMONITORED, NO_DATA, UNKNOWN) that fell out of every summary bucket, so totals never reconciled. - Report counts now reconcile. Summary buckets partition every tracked IP; counts_reconcile asserts cloud+local+indeterminate == total_ips_tracked. - Explainable cloud verdicts. Record known-Flock IPs a device contacted (cloud_ip_contacts) and surface CLOUD_IP(...) evidence, plus a TLS_CATEGORY(...) fallback, so a cloud verdict is never unexplained. - Bounded captures. Per-device raw sample lists cap at max_samples (default 500, --tap-max-samples). Exact integer counters increment regardless, so verdict inputs and report totals stay precise past the cap. - Bystander protection. FRP auth detection stores a non-content descriptor (matched keyword + payload length) instead of 64 raw payload bytes. Optional --tap-redact-ips salt-hashes non-Flock destination IPs in stored samples; known-Flock IPs stay legible so cloud evidence is readable. - Metadata-only destination_summary per cloud device (DNS names, TLS SNIs, known-Flock IPs) — who it talks to, never payload contents. Tests extend the merged suite with audit-integrity coverage (vocabulary agreement, bucket reconciliation, indeterminate counting, CLOUD_IP evidence, sample capping vs exact counters, FRP descriptor, IP redaction, metadata-only summary). Stale-vocabulary assertions updated to INDETERMINATE. Co-Authored-By: Claude Opus 4.6 --- flock_tap.py | 254 +++++++++++++++++++++++++++------------- tests/test_flock_tap.py | 107 ++++++++++++++++- 2 files changed, 276 insertions(+), 85 deletions(-) diff --git a/flock_tap.py b/flock_tap.py index f6dc000..df4a0f2 100644 --- a/flock_tap.py +++ b/flock_tap.py @@ -42,6 +42,12 @@ FLOCK_CLOUD_IPS = [ "104.18.16.189", "104.18.17.189", ] +# ── Canonical verdict vocabulary (single source of truth) ── +# Every device maps to exactly one of these so audit summary counts reconcile. +VERDICT_CLOUD = "CLOUD_CONNECTED" +VERDICT_LOCAL = "LOCAL_STATION" +VERDICT_INDETERMINATE = "INDETERMINATE" + # Scapy availability try: from scapy.all import sniff, IP, TCP, UDP, DNS, DNSQR, Raw, conf @@ -72,7 +78,8 @@ class FlockTrafficTap: """ def __init__(self, interface=None, pcap=None, pipe=False, verbose=False, - output_file=None, timeout=None, filter_expr=None): + output_file=None, timeout=None, filter_expr=None, + max_samples=500, redact_non_flock_ips=False): self.interface = interface self.pcap = pcap self.pipe = pipe @@ -81,6 +88,16 @@ class FlockTrafficTap: self.timeout = timeout self.filter_expr = filter_expr + # Data-minimization knobs (see AUDIT_INTEGRITY_SPEC.md). + # max_samples caps per-device raw sample lists so a long audit does not + # accrete an unbounded pile of raw records; integer counters below stay + # exact regardless. redact_non_flock_ips salt-hashes non-Flock dst IPs + # in stored samples to protect bystanders (off by default: hashing + # removes reproducibility a Flock-side auditor may need). + self.max_samples = max_samples + self.redact_non_flock_ips = redact_non_flock_ips + self._redact_salt = os.urandom(16) + # ── flow_stats: matches pseudocode structure ── self.flow_stats = defaultdict(lambda: { "cloud_dns": 0, @@ -102,6 +119,10 @@ class FlockTrafficTap: "frp_tunnels": [], "tls_snis": [], "http_requests": [], + "cloud_ip_contacts": [], + # Exact running totals — always incremented even after the sample + # lists above stop growing at max_samples, so counts/verdicts hold. + "counts": {"dns": 0, "conn": 0, "frp": 0, "sni": 0, "flock_dns": 0}, "bytes_up": 0, "bytes_down": 0, "packets_seen": 0, @@ -170,38 +191,57 @@ class FlockTrafficTap: # CLASSIFICATION — matches pseudocode interface # ═══════════════════════════════════════════════════ + def _classify(self, ip): + """Single source of truth for a device verdict. + + Folds flow-stat signals and device-level raw fallback into exactly one + canonical verdict so audit summary counts always reconcile. + """ + stats = self.flow_stats[ip] + # Cloud signals from the per-flow counters. + if (stats.get("cloud_dns", 0) > 0 or stats.get("frp_tunnel") + or stats.get("s3_uploads", 0) > 0 or stats.get("auth_tls", 0) > 0 + or stats.get("cloud_api", 0) > 0): + return VERDICT_CLOUD + + # Cloud signals from the raw device store (fallback when flow_stats is + # sparse — e.g. SNI/FRP seen but no counter set). + dev = self.devices.get(ip, {}) + if dev: + has_flock_sni = any( + "flock" in s.get("sni", "").lower() or "auth0" in s.get("sni", "").lower() + for s in dev.get("tls_snis", []) + ) + if has_flock_sni or dev.get("frp_tunnels") or dev.get("cloud_ip_contacts"): + return VERDICT_CLOUD + + # Traffic seen but no cloud signal → stays on-prem. + if stats.get("connections", 0) > 0 or (dev and dev.get("connections")): + return VERDICT_LOCAL + + return VERDICT_INDETERMINATE + + # Public names kept as thin wrappers so existing callers/tests stay valid. def classify_device(self, camera_ip): - """Generate a full traffic profile for a camera (from pseudocode).""" - stats = self.flow_stats[camera_ip] - if stats.get("cloud_dns", 0) > 0 or stats.get("frp_tunnel"): - return "CLOUD_CONNECTED" - if stats.get("s3_uploads", 0) > 0 or stats.get("auth_tls", 0) > 0: - return "CLOUD_CONNECTED" - if stats.get("cloud_api", 0) > 0: - return "CLOUD_CONNECTED" - if stats.get("connections", 0) > 0: - return "LOCAL_STATION" - return "OFFLINE_OR_UNMONITORED" + return self._classify(camera_ip) def classify_camera(self, ip): - """Return verdict for a single camera IP (raw data fallback).""" - verdict = self.classify_device(ip) - if verdict != "OFFLINE_OR_UNMONITORED": - return verdict + return self._classify(ip) - dev = self.devices.get(ip, {}) - if not dev: - return "NO_DATA" - has_flock_sni = any( - "flock" in s.get("sni", "").lower() or "auth0" in s.get("sni", "").lower() - for s in dev.get("tls_snis", []) - ) - has_frp = len(dev.get("frp_tunnels", [])) > 0 - if has_flock_sni or has_frp: - return "CLOUD_CONNECTED" - if dev.get("connections"): - return "LOCAL_STATION" - return "UNKNOWN" + # ── data-minimization helpers ── + def _append_capped(self, lst, item): + """Append a raw sample only while under the cap. Counters (kept + separately) stay exact, so bounding samples never skews verdicts.""" + if len(lst) < self.max_samples: + lst.append(item) + + def _redact_ip(self, ip): + """Salt-hash a non-Flock IP for bystander protection. Known-Flock IPs + pass through unchanged so cloud evidence stays legible.""" + if not self.redact_non_flock_ips or ip in self.known_flock_ips: + return ip + import hashlib + return "redacted:" + hashlib.sha256(self._redact_salt + ip.encode()).hexdigest()[:12] # ═══════════════════════════════════════════════════ # PACKET HANDLER — scapy @@ -246,7 +286,10 @@ class FlockTrafficTap: pass self.on_dns_query(qname, src, resolved_ips, ts) - dev["dns_queries"].append({ + dev["counts"]["dns"] += 1 + if "flock" in qname.lower() or "auth0" in qname.lower(): + dev["counts"]["flock_dns"] += 1 + self._append_capped(dev["dns_queries"], { "timestamp": ts, "src": src, "query": qname, "resolved_ips": resolved_ips, }) @@ -271,11 +314,14 @@ class FlockTrafficTap: # ── Connection to a known Flock cloud IP (no SNI/DNS needed) ── if dst in self.known_flock_ips: self.flow_stats[src]["cloud_api"] += 1 + if dst not in dev["cloud_ip_contacts"]: + dev["cloud_ip_contacts"].append(dst) if tcp.flags & 0x02: # SYN - dev["connections"].append({ + dev["counts"]["conn"] += 1 + self._append_capped(dev["connections"], { "timestamp": ts, "src": src, "sport": sport, - "dst": dst, "dport": dport, "bytes": ip.len, + "dst": self._redact_ip(dst), "dport": dport, "bytes": ip.len, }) # ── FRP Tunnel Detection ── @@ -294,8 +340,9 @@ class FlockTrafficTap: # Port-based signal is evaluated per-packet (fires on the SYN too). if dport in self.frp_ports: self.flow_stats[src]["frp_tunnel"] = True - dev["frp_tunnels"].append({ - "timestamp": ts, "src": src, "dst": dst, + dev["counts"]["frp"] += 1 + self._append_capped(dev["frp_tunnels"], { + "timestamp": ts, "src": src, "dst": self._redact_ip(dst), "port": dport, "type": "frp_tunnel", }) if self.verbose: @@ -306,12 +353,19 @@ class FlockTrafficTap: # packet — not only on the SYN (SYN packets carry no payload). if tcp.haslayer(Raw): raw = tcp[Raw].load - if b"frp" in raw.lower() or b"auth" in raw.lower() or b"proxy_type" in raw.lower(): + low = raw.lower() + matched = next((k for k in (b"frp", b"auth", b"proxy_type") if k in low), None) + if matched: self.flow_stats[src]["frp_tunnel"] = True - dev["frp_tunnels"].append({ - "timestamp": ts, "src": src, "dst": dst, + dev["counts"]["frp"] += 1 + # Store a non-content descriptor, not the raw bytes: which + # keyword matched + payload length. Proves FRP detection + # without retaining any of the payload (bystander protection). + self._append_capped(dev["frp_tunnels"], { + "timestamp": ts, "src": src, "dst": self._redact_ip(dst), "port": dport, "type": "frp_auth_payload", - "payload_preview": raw[:64].decode(errors="replace"), + "matched_keyword": matched.decode(), + "payload_len": len(raw), }) if self.verbose: self.log(f"{C.R}FRP_AUTH {src} -> {dst}:{dport} - auth payload{C.END}") @@ -325,7 +379,11 @@ class FlockTrafficTap: sni = self._extract_sni(raw) if sni: self.on_tls_sni(sni, src, dst, ts) - dev["tls_snis"].append({"timestamp": ts, "src": src, "dst": dst, "sni": sni}) + dev["counts"]["sni"] += 1 + self._append_capped(dev["tls_snis"], { + "timestamp": ts, "src": src, + "dst": self._redact_ip(dst), "sni": sni, + }) if self.verbose: cat = self._categorize_sni(sni) c = C.R if cat else C.CY @@ -491,14 +549,18 @@ class FlockTrafficTap: if parsed["type"] == "dns": query = parsed.get("query", "") if query: - dev["dns_queries"].append({ + dev["counts"]["dns"] += 1 + is_flock = "flock" in query.lower() or "auth0" in query.lower() + if is_flock: + dev["counts"]["flock_dns"] += 1 + self._append_capped(dev["dns_queries"], { "timestamp": time.time(), "src": src, "query": query, "resolved_ips": [], }) self.seen_domains.add(query) - if "flock" in query.lower(): + if is_flock: self.flow_stats[src]["cloud_dns"] += 1 if self.verbose: self.log(f"{C.R}Flock DNS {src} -> {query}{C.END}") @@ -506,9 +568,10 @@ class FlockTrafficTap: self.log(f"{C.CY}DNS {src} -> {query}{C.END}") if parsed["type"] == "syn" and dport in self.frp_ports: self.flow_stats[src]["frp_tunnel"] = True - dev["frp_tunnels"].append({ + dev["counts"]["frp"] += 1 + self._append_capped(dev["frp_tunnels"], { "timestamp": time.time(), "src": src, - "dst": dst, "port": dport, "type": "frp_tunnel", + "dst": self._redact_ip(dst), "port": dport, "type": "frp_tunnel", }) if self.verbose: self.log(f"{C.R}FRP {src} -> {dst}:{dport}{C.END}") @@ -536,18 +599,12 @@ class FlockTrafficTap: "summary": {}, } - cloud_count = 0 - local_count = 0 - unknown_count = 0 + counts = {VERDICT_CLOUD: 0, VERDICT_LOCAL: 0, VERDICT_INDETERMINATE: 0} - for ip in sorted(set(list(self.devices.keys()) + list(self.flow_stats.keys()))): - verdict = self.classify_camera(ip) - if verdict == "CLOUD_CONNECTED": - cloud_count += 1 - elif verdict == "LOCAL_STATION": - local_count += 1 - else: - unknown_count += 1 + all_ips = sorted(set(list(self.devices.keys()) + list(self.flow_stats.keys()))) + for ip in all_ips: + verdict = self._classify(ip) + counts[verdict] += 1 # Always include flow_stats report["flow_stats"][ip] = dict(self.flow_stats[ip]) @@ -556,8 +613,10 @@ class FlockTrafficTap: if dev.get("packets_seen", 0) < 5: continue - snis = list(set(s["sni"] for s in dev.get("tls_snis", []))) - dns_list = list(set(q["query"] for q in dev.get("dns_queries", []))) + c = dev.get("counts", {}) + snis = sorted(set(s["sni"] for s in dev.get("tls_snis", []))) + dns_list = sorted(set(q["query"] for q in dev.get("dns_queries", []))) + cloud_ips = list(dev.get("cloud_ip_contacts", [])) report["devices"][ip] = { "verdict": verdict, @@ -566,31 +625,44 @@ class FlockTrafficTap: "bytes_down": dev["bytes_down"], "first_seen": dev["first_seen"], "last_seen": dev["last_seen"], - "dns_queries_count": len(dev["dns_queries"]), - "tls_snis_count": len(dev.get("tls_snis", [])), - "frp_tunnels_count": len(dev.get("frp_tunnels", [])), - "connections_count": len(dev.get("connections", [])), + # Exact counters (not len() of the capped sample lists). + "dns_queries_count": c.get("dns", len(dev["dns_queries"])), + "tls_snis_count": c.get("sni", len(dev.get("tls_snis", []))), + "frp_tunnels_count": c.get("frp", len(dev.get("frp_tunnels", []))), + "connections_count": c.get("conn", len(dev.get("connections", []))), + "samples_capped_at": self.max_samples, "flow_stats": dict(self.flow_stats[ip]), - "tls_snis": snis, + # Metadata-only destination evidence — who the device talks to, + # never payload contents. + "destination_summary": { + "dns_domains": dns_list, + "tls_snis": snis, + "flock_cloud_ips": cloud_ips, + }, "frp_tunnels": dev.get("frp_tunnels", []), - "dns_domains": dns_list, } report["summary"] = { - "total_ips_tracked": len(self.devices), - "cloud_connected": cloud_count, - "local_station": local_count, - "unknown": unknown_count, + "total_ips_tracked": len(all_ips), + "cloud_connected": counts[VERDICT_CLOUD], + "local_station": counts[VERDICT_LOCAL], + "indeterminate": counts[VERDICT_INDETERMINATE], + # Invariant: the three buckets partition every tracked IP. + "counts_reconcile": ( + counts[VERDICT_CLOUD] + counts[VERDICT_LOCAL] + + counts[VERDICT_INDETERMINATE] == len(all_ips) + ), "flock_domains_resolved": sorted( - d for d in self.seen_domains if "flock" in d.lower() + d for d in self.seen_domains + if "flock" in d.lower() or "auth0" in d.lower() ), "total_flock_queries": sum( - 1 for dev in self.devices.values() - for q in dev.get("dns_queries", []) - if "flock" in q.get("query", "").lower() + dev.get("counts", {}).get("flock_dns", 0) + for dev in self.devices.values() ), "total_frp_tunnels": sum( - len(dev.get("frp_tunnels", [])) for dev in self.devices.values() + dev.get("counts", {}).get("frp", 0) + for dev in self.devices.values() ), } return report @@ -633,15 +705,16 @@ class FlockTrafficTap: print(f" Duration: {elapsed:.1f}s") print(f" Packets: {self.packet_count:,}") - cloud_cams = [] - for ip in sorted(self.devices): - if self.classify_camera(ip) == "CLOUD_CONNECTED": - cloud_cams.append(ip) + # Partition every tracked IP into exactly one bucket (counts reconcile). + verdicts = {ip: self._classify(ip) for ip in self.devices} + cloud_cams = [ip for ip, v in sorted(verdicts.items()) if v == VERDICT_CLOUD] + local_n = sum(1 for v in verdicts.values() if v == VERDICT_LOCAL) + indet_n = sum(1 for v in verdicts.values() if v == VERDICT_INDETERMINATE) print(f"\n{C.CY}Classification:{C.END}") print(f" Cloud-connected: {len(cloud_cams)}") - print(f" Local station: {sum(1 for ip in self.devices if self.classify_camera(ip) == 'LOCAL_STATION')}") - print(f" Unknown: {sum(1 for ip in self.devices if self.classify_camera(ip) == 'UNKNOWN')}") + print(f" Local station: {local_n}") + print(f" Indeterminate: {indet_n}") if cloud_cams: print(f"\n{C.R}Cloud-Connected Cameras:{C.END}") @@ -649,17 +722,27 @@ class FlockTrafficTap: dev = self.devices[ip] evidence = [] dns_list = list(set(q["query"] for q in dev.get("dns_queries", []) - if "flock" in q["query"].lower())) + if "flock" in q["query"].lower() or "auth0" in q["query"].lower())) snis = list(set(s["sni"] for s in dev.get("tls_snis", []) if "flock" in s["sni"].lower() or "auth0" in s["sni"].lower())) frps = dev.get("frp_tunnels", []) + cloud_ips = dev.get("cloud_ip_contacts", []) if dns_list: evidence.append(f"DNS({', '.join(dns_list[:3])})") if snis: evidence.append(f"SNI({', '.join(snis[:3])})") if frps: - evidence.append(f"FRP({len(frps)} tunnels)") - print(f" {C.R}[!] {ip:<16}{C.END} {' | '.join(evidence[:3])}") + evidence.append(f"FRP({dev.get('counts', {}).get('frp', len(frps))} tunnels)") + if cloud_ips: + evidence.append(f"CLOUD_IP({', '.join(cloud_ips[:3])})") + if not evidence: + # Cloud via an S3/auth/api SNI category counter with no + # retained hostname sample — name the signal so the verdict + # is never unexplained. + fs = self.flow_stats[ip] + sig = [k for k in ("s3_uploads", "auth_tls", "cloud_api") if fs.get(k)] + evidence.append(f"TLS_CATEGORY({', '.join(sig)})" if sig else "cloud signal") + print(f" {C.R}[!] {ip:<16}{C.END} {' | '.join(evidence[:4])}") # All Flock domains seen flock_domains = sorted( @@ -671,8 +754,8 @@ class FlockTrafficTap: for d in flock_domains: print(f" ﹒ {d}") - # FRP tunnels - total_frp = sum(len(dev.get("frp_tunnels", [])) for dev in self.devices.values()) + # FRP tunnels (exact counter; the per-tunnel list below is a capped sample) + total_frp = sum(dev.get("counts", {}).get("frp", 0) for dev in self.devices.values()) if total_frp > 0: print(f"\n{C.R}FRP Tunnels: {total_frp}{C.END}") for ip, dev in sorted(self.devices.items()): @@ -729,12 +812,17 @@ def main(): parser.add_argument("--tap-verbose", "-v", action="store_true", help="Verbose output") parser.add_argument("--tap-timeout", type=int, default=None, help="Capture duration (seconds)") parser.add_argument("--tap-filter", default=None, help="BPF filter expression") + parser.add_argument("--tap-max-samples", type=int, default=500, + help="Cap on per-device raw samples kept (data minimization). Default 500.") + parser.add_argument("--tap-redact-ips", action="store_true", + help="Salt-hash non-Flock destination IPs in stored samples (bystander protection).") args = parser.parse_args() tap = FlockTrafficTap( interface=args.tap_interface, pcap=args.tap_pcap, pipe=args.tap_pipe, verbose=args.tap_verbose, output_file=args.tap_output, timeout=args.tap_timeout, - filter_expr=args.tap_filter, + filter_expr=args.tap_filter, max_samples=args.tap_max_samples, + redact_non_flock_ips=args.tap_redact_ips, ) if not args.tap_interface and not args.tap_pcap and not args.tap_pipe: parser.print_help() diff --git a/tests/test_flock_tap.py b/tests/test_flock_tap.py index fbe68cf..3df2ffc 100644 --- a/tests/test_flock_tap.py +++ b/tests/test_flock_tap.py @@ -131,13 +131,15 @@ def test_classify_device_local_station(tap): def test_classify_device_offline(tap): - assert tap.classify_device("ip") == "OFFLINE_OR_UNMONITORED" + # Unified vocabulary: no traffic → INDETERMINATE (was OFFLINE_OR_UNMONITORED). + assert tap.classify_device("ip") == "INDETERMINATE" # ── classify_camera (fallback via device data) ────────────────────────────── def test_classify_camera_no_data(tap): - assert tap.classify_camera("10.0.0.99") == "NO_DATA" + # Unified vocabulary: no data → INDETERMINATE (was NO_DATA). + assert tap.classify_camera("10.0.0.99") == "INDETERMINATE" def test_classify_camera_flock_sni_fallback(tap): @@ -227,3 +229,104 @@ def test_known_flock_ip_correlation(tap): tap._handle_packet_scapy(pkt) assert tap.flow_stats["10.0.0.5"]["cloud_api"] >= 1 assert tap.classify_device("10.0.0.5") == "CLOUD_CONNECTED" + + +# ── audit-integrity: unified verdict vocabulary ───────────────────────────── + +def test_device_and_camera_classifiers_agree(tap): + """Both public names fold into one canonical verdict.""" + tap.flow_stats["10.0.0.5"]["cloud_dns"] = 1 + assert tap.classify_device("10.0.0.5") == tap.classify_camera("10.0.0.5") == "CLOUD_CONNECTED" + + +# ── audit-integrity: report buckets partition every tracked IP ────────────── + +def test_summary_buckets_reconcile(tap): + tap.start_time = 0 + tap.flow_stats["10.0.0.1"]["cloud_dns"] = 1 # cloud + tap.flow_stats["10.0.0.2"]["connections"] = 2 # local + _ = tap.flow_stats["10.0.0.3"] # indeterminate (touched only) + s = tap.generate_report()["summary"] + assert s["counts_reconcile"] is True + assert s["cloud_connected"] + s["local_station"] + s["indeterminate"] == s["total_ips_tracked"] + + +def test_indeterminate_device_still_counted(tap): + """Regression: the old report dropped NO_DATA/OFFLINE from every bucket.""" + tap.start_time = 0 + _ = tap.flow_stats["10.0.0.9"] + assert tap.generate_report()["summary"]["indeterminate"] == 1 + + +# ── audit-integrity: CLOUD_IP evidence is recorded ────────────────────────── + +@requires_scapy +def test_cloud_ip_contact_recorded_for_evidence(tap): + dst = flock_tap.FLOCK_CLOUD_IPS[0] + pkt = _rebuild(IP(src="192.168.1.120", dst=dst) / TCP(sport=47000, dport=443, flags="A")) + tap._handle_packet_scapy(pkt) + assert dst in tap.devices["192.168.1.120"]["cloud_ip_contacts"] + + +# ── audit-integrity: memory capping keeps counters exact ──────────────────── + +@requires_scapy +def test_samples_capped_but_counter_exact(): + tap = FlockTrafficTap(max_samples=3) + for i in range(10): + pkt = _rebuild(IP(src="192.168.1.130", dst="10.0.0.20") / + TCP(sport=48000 + i, dport=8080, flags="S")) + tap._handle_packet_scapy(pkt) + dev = tap.devices["192.168.1.130"] + assert len(dev["connections"]) == 3 # sample list bounded + assert dev["counts"]["conn"] == 10 # counter exact + rep = tap.generate_report() + assert rep["devices"]["192.168.1.130"]["connections_count"] == 10 + + +# ── audit-integrity: FRP auth stores a descriptor, not raw payload ────────── + +@requires_scapy +def test_frp_auth_stores_descriptor_not_payload(tap): + pkt = _rebuild(IP(src="192.168.1.140", dst="10.0.0.30") / + TCP(sport=49000, dport=8080, flags="PA") / + Raw(load=b'{"proxy_type":"tcp","secret":"topsecret"}')) + tap._handle_packet_scapy(pkt) + rec = [f for f in tap.devices["192.168.1.140"]["frp_tunnels"] + if f["type"] == "frp_auth_payload"][0] + assert "payload_preview" not in rec + assert rec["matched_keyword"] == "proxy_type" + assert rec["payload_len"] == 41 + assert "topsecret" not in str(rec) # secret bytes must not be retained + + +# ── audit-integrity: optional non-Flock IP redaction ──────────────────────── + +@requires_scapy +def test_redaction_hashes_non_flock_ip_keeps_flock_clear(): + tap = FlockTrafficTap(redact_non_flock_ips=True) + flock = flock_tap.FLOCK_CLOUD_IPS[0] + p1 = _rebuild(IP(src="192.168.1.150", dst="8.8.8.8") / TCP(sport=50000, dport=8080, flags="S")) + tap._handle_packet_scapy(p1) + assert tap.devices["192.168.1.150"]["connections"][0]["dst"].startswith("redacted:") + p2 = _rebuild(IP(src="192.168.1.151", dst=flock) / TCP(sport=50001, dport=7000, flags="S")) + tap._handle_packet_scapy(p2) + assert tap.devices["192.168.1.151"]["frp_tunnels"][0]["dst"] == flock + + +# ── audit-integrity: destination summary is metadata-only ─────────────────── + +@requires_scapy +def test_destination_summary_is_metadata_only(tap): + tap.start_time = 0 + for i in range(6): # clear the <5-packet filter + tap._handle_packet_scapy( + _rebuild(IP(src="192.168.1.160", dst="10.0.0.40") / + TCP(sport=51000 + i, dport=8080, flags="S"))) + tap.on_tls_sni("api.flocksafety.com", "192.168.1.160", "10.0.0.40") + tap.devices["192.168.1.160"]["tls_snis"].append( + {"timestamp": 0, "src": "192.168.1.160", "dst": "10.0.0.40", "sni": "api.flocksafety.com"}) + tap.devices["192.168.1.160"]["counts"]["sni"] += 1 + dsum = tap.generate_report()["devices"]["192.168.1.160"]["destination_summary"] + assert set(dsum.keys()) == {"dns_domains", "tls_snis", "flock_cloud_ips"} + assert "api.flocksafety.com" in dsum["tls_snis"] -- 2.53.0