#!/usr/bin/env python3 """Sentinet - passive network monitor (single-file edition). Download, then run: python3 sentinet-passive.py Opens a local dashboard in your browser showing your apps' real outbound connections - process -> host -> volume, live. No install, no dependencies, no proxy, no root. macOS only. It is metadata-only by design: it shows WHO is talking to the network, WHERE, and HOW MUCH. It never reads the contents of your traffic (that is encrypted inside each app), so every flow is honestly marked "content unavailable", and blocking here is advisory (observe, not enforce). Stop with Ctrl+C. """ from __future__ import annotations import json import os import queue import re import socket import subprocess import sys import threading import time import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer DASHBOARD_HTML = '\n\n\n\n\nSentinet — live network monitor\n\n\n\n\n\n\n\n
\n
\n
Sentinet
\n
connecting…
\n \n
\n
apps
0
\n
conns
0
\n
up
0 B/s
\n
down
0 B/s
\n
blocked
0
\n
\n
\n
\n\n
\n
\n
\n ⌕\n \n
\n
\n \n \n \n \n
\n \n
\n\n
\n \n \n \n \n \n \n \n \n \n \n \n
Process / HostVolume ↑↓PayloadVerdictAction
\n
Waiting for traffic
\n
\n
\n\n\n\n\n\n\n' # ===================== rules / verdict ===================== _SYSTEM_NAMES = { "mDNSResponder", "trustd", "apsd", "nsurlsessiond", "rapportd", "cloudd", "syncdefaultsd", "identityservicesd", "AssetCacheLocatorService", "configd", "softwareupdated", "akd", "geod", "timed", "locationd", "remindd", "commerce", "appstoreagent", "mdmclient", "networkserviceproxy", "amsengagementd", "symptomsd", "airportd", } _SYSTEM_PREFIXES = ("com.apple.", "system_", "_system") def is_system_process(name): if not name: return False return name in _SYSTEM_NAMES or name.startswith(_SYSTEM_PREFIXES) class RuleSet: def __init__(self): self._lock = threading.Lock() self._hosts = set() self._procs = set() def block(self, host=None, process=None): with self._lock: if host: self._hosts.add(host) if process: self._procs.add(process) def unblock(self, host=None, process=None): with self._lock: if host: self._hosts.discard(host) if process: self._procs.discard(process) def snapshot(self): with self._lock: return {"hosts": sorted(self._hosts), "processes": sorted(self._procs)} def is_blocked(self, host, process): with self._lock: return host in self._hosts or (process is not None and process in self._procs) RULES = RuleSet() def decide(c): if is_system_process(c["process"]): return "system" if RULES.is_blocked(c["host"], c["process"]): return "blocked" if c["findings"]: return "leak" if not c["payload_visible"]: return "locked" return "allow" # ===================== store + SSE bus ===================== class Store: def __init__(self): self._lock = threading.Lock() self._conns = {} self._subs = [] self.mode = "passive" self._tu = self._td = self._lu = self._ld = 0 self._last = time.time() def subscribe(self): q = queue.Queue(maxsize=1000) with self._lock: self._subs.append(q) return q def unsubscribe(self, q): with self._lock: if q in self._subs: self._subs.remove(q) def _publish(self, ev): with self._lock: subs = list(self._subs) for q in subs: try: q.put_nowait(ev) except queue.Full: pass def record(self, process, pid, host, port, scheme="https", bytes_up=0, bytes_down=0, payload_visible=False): cid = process + "@" + host now = time.time() with self._lock: c = self._conns.get(cid) if c is None: c = {"cid": cid, "process": process, "pid": pid, "host": host, "port": port, "scheme": scheme, "direction": "outbound", "bytes_up": 0, "bytes_down": 0, "payload_visible": payload_visible, "verdict": "locked", "findings": [], "requests": 0, "first_seen": now, "last_seen": now} self._conns[cid] = c c["bytes_up"] += bytes_up c["bytes_down"] += bytes_down c["requests"] += 1 c["last_seen"] = now c["pid"] = pid or c["pid"] if payload_visible: c["payload_visible"] = True c["verdict"] = decide(c) self._tu += bytes_up self._td += bytes_down snap = dict(c) self._publish({"type": "conn", "conn": snap}) def reverdict_all(self): with self._lock: conns = list(self._conns.values()) for c in conns: c["verdict"] = decide(c) snaps = [dict(c) for c in conns] for s in snaps: self._publish({"type": "conn", "conn": s}) def snapshot(self): with self._lock: return [dict(c) for c in self._conns.values()] def tick_stats(self): now = time.time() with self._lock: dt = max(now - self._last, 1e-3) up = (self._tu - self._lu) / dt dn = (self._td - self._ld) / dt self._lu, self._ld, self._last = self._tu, self._td, now conns = len(self._conns) stats = {"type": "stats", "mode": self.mode, "up_bps": up, "down_bps": dn, "connections": conns, "blocked": RULES.snapshot()} self._publish(stats) STORE = Store() def start_stats_ticker(): def loop(): while True: time.sleep(1.0) STORE.tick_stats() threading.Thread(target=loop, name="stats", daemon=True).start() # ===================== passive engine (nettop + lsof) ===================== _CONN_RE = re.compile(r"->(\d{1,3}(?:\.\d{1,3}){3}):(\d+)") _ESC_RE = re.compile(r"\\x([0-9a-fA-F]{2})") _CONN_INTERVAL = 2.0 _BYTES_INTERVAL = 6.0 _latest = {} _latest_lock = threading.Lock() _dns_cache = {} _dns_lock = threading.Lock() _dns_q = queue.Queue() def _unescape(s): return _ESC_RE.sub(lambda m: chr(int(m.group(1), 16)), s) def _dns_worker(): while True: ip = _dns_q.get() host = ip try: host = socket.gethostbyaddr(ip)[0] except Exception: host = ip with _dns_lock: _dns_cache[ip] = host def _hostname(ip): with _dns_lock: if ip in _dns_cache: return _dns_cache[ip] _dns_cache[ip] = ip _dns_q.put(ip) return ip def _poll_connections(): conns = {} try: out = subprocess.run(["lsof", "-nP", "-i", "+c", "0"], capture_output=True, text=True, timeout=4.0).stdout except (subprocess.SubprocessError, OSError): return conns for line in out.splitlines(): if line.startswith("COMMAND"): continue tokens = line.split() if len(tokens) < 8: continue pid_idx = next((i for i, t in enumerate(tokens) if i > 0 and t.isdigit()), None) if not pid_idx: continue proc = _unescape(" ".join(tokens[:pid_idx])) try: pid = int(tokens[pid_idx]) except ValueError: continue is_udp = "UDP" in tokens if not is_udp and "ESTABLISHED" not in line: continue m = _CONN_RE.search(line) if not m: continue ip, port = m.group(1), int(m.group(2)) if ip.startswith("127.") or ip == "::1": continue kind = "UDP" if is_udp else "TCP" conns.setdefault(pid, (proc, set()))[1].add((ip, port, kind)) return conns def _poll_bytes(): stats = {} try: out = subprocess.run(["nettop", "-P", "-L", "1", "-x"], capture_output=True, text=True, timeout=10.0).stdout except (subprocess.SubprocessError, OSError): return stats for line in out.splitlines(): f = line.split(",") if len(f) < 6 or "." not in f[1]: continue try: pid = int(f[1].rsplit(".", 1)[1]) stats[pid] = (int(f[4]), int(f[5])) except (ValueError, IndexError): continue return stats def _scheme(port, proto): return "https" if port == 443 else "http" if port == 80 else proto.lower() def _connections_loop(): while True: try: conns = _poll_connections() with _latest_lock: _latest.clear() _latest.update(conns) for pid, (proc, flows) in conns.items(): if is_system_process(proc) or not flows: continue for ip, port, proto in flows: STORE.record(proc, pid, _hostname(ip), port, _scheme(port, proto), 0, 0, payload_visible=False) except Exception: pass time.sleep(_CONN_INTERVAL) def _bytes_loop(): prev = _poll_bytes() time.sleep(_BYTES_INTERVAL) while True: try: now = _poll_bytes() with _latest_lock: snap = {pid: (p, set(f)) for pid, (p, f) in _latest.items()} for pid, (cin, cout) in now.items(): if pid not in prev or pid not in snap: continue din = max(0, cin - prev[pid][0]) dout = max(0, cout - prev[pid][1]) proc, flows = snap[pid] if not flows or (din == 0 and dout == 0): continue si, so = din // len(flows), dout // len(flows) for ip, port, proto in flows: STORE.record(proc, pid, _hostname(ip), port, _scheme(port, proto), so, si, payload_visible=False) prev = now except Exception: pass time.sleep(_BYTES_INTERVAL) def start_passive_monitor(): for fn, nm in ((_dns_worker, "dns"), (_connections_loop, "conns"), (_bytes_loop, "bytes")): threading.Thread(target=fn, name=nm, daemon=True).start() # ===================== dashboard server ===================== class Handler(BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): path = self.path.split("?", 1)[0] if path == "/": self._bytes(DASHBOARD_HTML.encode(), "text/html") elif path == "/api/state": self._json({"connections": STORE.snapshot(), "blocked": RULES.snapshot()}) elif path == "/events": self._events() else: self.send_error(404) def do_POST(self): if self.path != "/api/block": self.send_error(404) return n = int(self.headers.get("Content-Length", 0)) try: body = json.loads(self.rfile.read(n) or b"{}") except json.JSONDecodeError: self.send_error(400) return host, proc = body.get("host"), body.get("process") if body.get("action") == "unblock": RULES.unblock(host=host, process=proc) else: RULES.block(host=host, process=proc) STORE.reverdict_all() self._json({"ok": True, "blocked": RULES.snapshot()}) def _bytes(self, data, ctype): self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _json(self, obj, code=200): self._bytes(json.dumps(obj).encode(), "application/json") def _events(self): self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.end_headers() q = STORE.subscribe() try: self._sse(json.dumps({"type": "snapshot", "mode": STORE.mode, "connections": STORE.snapshot(), "blocked": RULES.snapshot()})) while True: try: self._sse(json.dumps(q.get(timeout=15))) except queue.Empty: self.wfile.write(b": keepalive\n\n") self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass finally: STORE.unsubscribe(q) def _sse(self, payload): self.wfile.write(("data: " + payload + "\n\n").encode()) self.wfile.flush() class _QuietServer(ThreadingHTTPServer): daemon_threads = True def handle_error(self, request, client_address): exc = sys.exc_info()[1] if isinstance(exc, (ConnectionResetError, BrokenPipeError)): return super().handle_error(request, client_address) def start_server(host, port): httpd = _QuietServer((host, port), Handler) threading.Thread(target=httpd.serve_forever, name="dashboard", daemon=True).start() return httpd # ===================== main ===================== def main(): import argparse p = argparse.ArgumentParser(description="Sentinet - passive network monitor (single file)") p.add_argument("--port", type=int, default=8787) p.add_argument("--host", default="127.0.0.1") p.add_argument("--no-browser", action="store_true") args = p.parse_args() if sys.platform != "darwin": print("[sentinet] Note: passive mode uses macOS tools (nettop, lsof). " "On other systems it will show no data.") start_server(args.host, args.port) start_stats_ticker() start_passive_monitor() url = "http://%s:%d" % (args.host, args.port) print("[sentinet] passive monitor running -> " + url) print("[sentinet] showing your apps' real outbound connections (metadata only).") print("[sentinet] payload is never read; blocking here is advisory. Ctrl+C to stop.") if not args.no_browser: try: webbrowser.open(url) except Exception: pass try: while True: time.sleep(1) except KeyboardInterrupt: print("\n[sentinet] bye.") if __name__ == "__main__": main()