#!/usr/bin/env python3 """Turn a restored copy of a production netOrk database into demo data. anonymize.py [--dsn postgresql://...] [--map demo-map.json] [--dry-run] Run it against the LOCAL copy only; it refuses anything that is not localhost. It works on every text-like column of every table instead of a hand-kept list, so a table added in a later release is covered too: * domains every configured domain (e.g. corp.example.com, acme.io) becomes `example.demo`, subdomains kept: gw.home.corp.example.com -> gw.home.example.demo * IPv4 private addresses move to another /16 per /16, host part kept, so subnets and VLAN plans still line up; public addresses are mapped one by one into the documentation ranges * IPv6 global prefixes go to 2001:db8::/32, interface IDs are hashed * MAC the vendor prefix (OUI) is kept, so manufacturer lookups still work; the device part is hashed * e-mail local part hashed, domain example.demo * names hostnames, site names, VLAN names, user names ... from the map * secrets stored credentials, keys, tokens, TOTP and secret settings are emptied; one admin `netork` with a known password is left Every mapping is deterministic, so the same address always turns into the same fake one, across tables, JSON documents and log lines alike. At the end a leak report lists anything that still looks like the original. """ import argparse import asyncio import hashlib import ipaddress import json import os import re import sys from pathlib import Path import asyncpg DEFAULT_DSN = "postgresql://netork:demo@127.0.0.1:55432/netork" DEFAULT_MAP = Path.home() / ".config" / "netork-screenshots" / "demo-map.json" DEMO_DOMAIN = "example.demo" # Public reference data: large, and nothing in it is about the instance. SKIP_TABLES = { "alembic_version", "cwe_entries", "epss_scores", "nvd_cpe_matches", "nvd_cpe_products", "nvd_cve_requirements", "nvd_cves", "osv_affected", "osv_vulns", "oui_vendors", "service_templates", } TEXT_TYPES = {"text", "character varying", "jsonb", "json", "inet", "cidr", "macaddr", "ARRAY"} # Only these count as internal addresses to move; Python's is_private also # covers 0.0.0.0/8 and friends, which in practice are version numbers. PRIVATE_NETS = [ipaddress.IPv4Network(n) for n in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "100.64.0.0/10")] # Well-known public resolvers stay as they are; they say nothing about anyone. KEEP_PUBLIC = {"1.1.1.1", "1.0.0.1", "8.8.8.8", "8.8.4.4", "9.9.9.9", "149.112.112.112"} IPV4 = re.compile(r"(? str: return hashlib.sha256(value.encode()).hexdigest()[:n] class Mapper: def __init__(self, cfg: dict): # {"home.corp.example.com": "hq.example.demo", "corp.example.com": "example.demo"} self.domains: dict[str, str] = cfg.get("domains", {}) self.prefix16 = dict(cfg.get("ipv4_prefix16", {})) taken = set(self.prefix16.values()) pool = cfg.get("ipv4_pool16") or ( [f"10.{n}" for n in range(20, 256, 10)] + [f"10.{n}" for n in range(256) if n % 10] + [f"172.{n}" for n in range(16, 32)]) self.pool16 = iter(p for p in pool if p not in taken) self.public: dict[str, str] = {} self.public_used: set[str] = set() # Public-looking dotted quads are only mapped once they were seen as an # address (see collect_public); "kernel 6.8.0.45" is a version, not a host. self.known_public: set[str] = set(cfg.get("public_ips", [])) self.unmapped_public: dict[str, int] = {} self.public_pool = iter( [f"203.0.113.{n}" for n in range(10, 250)] + [f"198.51.100.{n}" for n in range(10, 250)]) names = {**cfg.get("hostnames", {}), **cfg.get("terms", {})} self.names = names self.names_re = None if names: alt = "|".join(re.escape(k) for k in sorted(names, key=len, reverse=True)) # A name is a whole token: not glued to letters, digits, '-' or '_'. self.names_re = re.compile(rf"(? str: a = ipaddress.IPv4Address(ip) if ip in KEEP_PUBLIC or a.is_loopback or a.is_multicast or a.is_unspecified \ or a.is_link_local or ip.startswith("255.") or a.is_reserved: return ip if any(a in net for net in PRIVATE_NETS): p = ".".join(ip.split(".")[:2]) if p not in self.prefix16: self.prefix16[p] = next(self.pool16) return self.prefix16[p] + "." + ".".join(ip.split(".")[2:]) if not a.is_global: return ip # 0.x, 192.0.0.x, benchmark ... : versions more often than hosts if ip not in self.known_public: self.unmapped_public[ip] = self.unmapped_public.get(ip, 0) + 1 return ip if ip not in self.public: fake = next(self.public_pool, None) probe = 0 while fake is None or fake in self.public_used: # Documentation ranges exhausted (CrowdSec alone brings tens of # thousands of attacker addresses): hash into the non-routable # benchmark range 198.18.0.0/15, probing on collision. n = int(h(f"{ip}/{probe}", 8), 16) % (2 ** 17) fake = f"198.{18 + (n >> 16)}.{(n >> 8) & 255}.{n & 255}" probe += 1 self.public_used.add(fake) self.public[ip] = fake return self.public[ip] def mac(self, m: str) -> str: sep = m[2] hexs = m.replace(sep, "") new = hexs[:6] + h(hexs.lower(), 6) new = new.upper() if hexs.isupper() else new.lower() return sep.join(new[i:i + 2] for i in range(0, 12, 2)) def mac_dot(self, m: str) -> str: hexs = m.replace(".", "") new = hexs[:6] + h(hexs.lower(), 6) return ".".join(new[i:i + 4] for i in range(0, 12, 4)) def ipv6(self, s: str) -> str: # "Data::" or "12:30:45" are no addresses; demand three real groups. if sum(1 for g in s.split(":") if g) < 3: return s try: a = ipaddress.IPv6Address(s) except ValueError: return s # a time like 12:30:45 or similar, not an address if a.is_loopback or a.is_unspecified or a.is_multicast: return s iid = h(a.packed[8:].hex(), 16) if a.is_link_local: prefix = "fe80:0000:0000:0000" elif a.is_private: # ULA fd00::/8, keep it ULA prefix = "fd00:" + h(a.packed[:8].hex(), 12) prefix = prefix[:4] + ":" + prefix[5:9] + ":" + prefix[9:13] + ":" + prefix[13:17].ljust(4, "0") else: p = h(a.packed[:8].hex(), 8) prefix = f"2001:0db8:{p[:4]}:{p[4:]}" full = prefix + ":" + ":".join(iid[i:i + 4] for i in range(0, 16, 4)) return str(ipaddress.IPv6Address(full)) def email(self, m: re.Match) -> str: e = m.group(0) if e.endswith("@" + DEMO_DOMAIN): return e return f"user-{h(e.lower(), 6)}@{DEMO_DOMAIN}" def reverse(self, m: re.Match) -> str: octets = m.group(1).rstrip(".").split(".")[::-1] # forward order if len(octets) < 2 or any(int(o) > 255 for o in octets): return m.group(0) padded = octets + ["0"] * (4 - len(octets)) mapped = self.ipv4(".".join(padded)).split(".")[:len(octets)] return ".".join(mapped[::-1]) + ".in-addr.arpa" # -- whole strings ------------------------------------------------------- def text(self, s: str) -> str: s = SECRET_JSON.sub(lambda m: m.group(0) if m.group(1) in SECRET_JSON_KEEP else f'"{m.group(1)}"{m.group(2)}""', s) s = REVERSE.sub(self.reverse, s) s = EMAIL.sub(self.email, s) if self.domain_re: s = self.domain_re.sub(lambda m: m.group(1) + self.domains[m.group(2).lower()], s) s = MAC.sub(lambda m: self.mac(m.group(1)), s) s = MAC_DOT.sub(lambda m: self.mac_dot(m.group(1)), s) s = IPV6.sub(lambda m: self.ipv6(m.group(1)), s) s = IPV4.sub(lambda m: self.ipv4(m.group(1)), s) if self.names_re: s = self.names_re.sub(lambda m: self.names[m.group(1)], s) for old, new in self.substrings.items(): s = s.replace(old, new) return s # Cheap server-side prefilter: only rows that could contain something to map. def prefilter(cfg: dict) -> str: parts = [r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", r"[0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-]", r"[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}\.", r"[0-9A-Fa-f]{1,4}::?[0-9A-Fa-f]{1,4}:", "@", r"in-addr\.arpa", r"(key|psk|passphrase|password|secret|token)\"\s*:"] for k in [*cfg.get("domains", []), *cfg.get("hostnames", {}), *cfg.get("terms", {}), *cfg.get("substrings", {})]: parts.append(re.escape(k)) return "|".join(parts) # Columns emptied wherever they occur, found by name so a new table is covered. SECRET_COLUMN = re.compile(r"(password|secret|private_key|api_key|apikey|token|passphrase|psk|ft_key|wpa_key)", re.I) # The same inside JSON and text: device snapshots carry Wi-Fi keys and the like. SECRET_JSON = re.compile( r'"((?:[A-Za-z0-9_]*_)?(?:key|psk|passphrase|password|passwd|secret|token|private_key|ft_key|sae_password))"' r'(\s*:\s*)"(?:[^"\\]|\\.)*"') SECRET_JSON_KEEP = {"public_key", "entry_key", "key_type", "is_secret", "ssh_key_id"} SECRET_KEEP = {"hashed_password", "token_version", "title_tokens", "disable_password_auth"} # Whole tables that only hold secrets or personal delivery data. SECRET_TABLES = ["user_ssh_keys", "user_backup_codes", "notification_deliveries", "notification_mutes", "notification_channels", "trusted_networks"] async def columns(con) -> list[tuple[str, str, str]]: rows = await con.fetch( "SELECT table_name, column_name, data_type FROM information_schema.columns " "WHERE table_schema = 'public' ORDER BY table_name, ordinal_position") return [(r[0], r[1], r[2]) for r in rows if r[0] not in SKIP_TABLES and r[2] in TEXT_TYPES] ADDRESS_COLUMN = re.compile(r"(^|_)(ip|ips|ip_address|address|addr|host|target|source|wan|gateway|peer|value)(_|$)") QUAD = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}" # A dotted quad reads as an address when it is a whole JSON string value (not # under a version-like key) or follows a word that introduces an address. AS_JSON_VALUE = re.compile(rf'(?:"([^"]*)"\s*:\s*)?"({QUAD})(?:/\d{{1,2}})?"') AS_PROSE = re.compile( rf"(?i)\b(?:from|to|ip|ipv4|addr|address|host|src|dst|source|peer|wan|gateway|gw|via|at|by|nameserver|server)\W{{1,3}}({QUAD})") VERSIONISH = re.compile(r"(?i)version|ver$|release|build|firmware|kernel|rev") def addresses_in(value: str, whole_column: bool) -> set[str]: found = set() if whole_column: found.update(IPV4.findall(value)) for key, ip in AS_JSON_VALUE.findall(value): if not (key and VERSIONISH.search(key)): found.add(ip) found.update(AS_PROSE.findall(value)) return found async def collect_public(con, mapper: Mapper) -> None: """Learn which public IPv4 addresses really are addresses.""" for t, c, dt in await columns(con): whole = dt in ("inet", "cidr") or bool(ADDRESS_COLUMN.search(c)) rows = await con.fetch( f'SELECT DISTINCT "{c}"::text AS v FROM "{t}" WHERE "{c}"::text ~ $1', QUAD) for r in rows: for ip in addresses_in(r["v"], whole): try: a = ipaddress.IPv4Address(ip) except ValueError: continue if a.is_global and ip not in KEEP_PUBLIC: mapper.known_public.add(ip) async def scrub_secrets(con, dry: bool) -> None: rows = await con.fetch( "SELECT c.table_name, c.column_name, c.is_nullable, c.data_type " "FROM information_schema.columns c JOIN information_schema.tables t " "ON t.table_name = c.table_name AND t.table_schema = c.table_schema " "WHERE c.table_schema = 'public' AND t.table_type = 'BASE TABLE'") for t, c, nullable, dt in rows: if t in SKIP_TABLES or c in SECRET_KEEP or not SECRET_COLUMN.search(c): continue if dt not in ("text", "character varying", "jsonb", "json", "bytea"): continue # flags like require_password are booleans value = "NULL" if nullable == "YES" else ("'{}'" if dt in ("jsonb", "json") else "''") if dt == "bytea" and nullable != "YES": value = "''::bytea" n = await con.fetchval(f'SELECT count(*) FROM "{t}" WHERE "{c}" IS NOT NULL') if n: print(f" {t}.{c}: {n} emptied") if not dry: await con.execute(f'UPDATE "{t}" SET "{c}" = {value}') # Settings flagged secret keep their key, lose their value. if await con.fetchval("SELECT to_regclass('public.settings') IS NOT NULL"): n = await con.fetchval("SELECT count(*) FROM settings WHERE is_secret") print(f" settings: {n} secret values emptied") if not dry: await con.execute("UPDATE settings SET value = '' WHERE is_secret") for t in SECRET_TABLES: if await con.fetchval("SELECT to_regclass($1) IS NOT NULL", f"public.{t}"): n = await con.fetchval(f'SELECT count(*) FROM "{t}"') print(f" {t}: {n} rows deleted") if not dry: await con.execute(f'DELETE FROM "{t}"') async def rewrite(con, mapper: Mapper, cfg: dict, dry: bool) -> None: pat = prefilter(cfg) by_table: dict[str, list[tuple[str, str]]] = {} for t, c, dt in await columns(con): by_table.setdefault(t, []).append((c, dt)) for table, cols in by_table.items(): for col, dt in cols: q = f'SELECT ctid, "{col}"::text AS v FROM "{table}" WHERE "{col}"::text ~ $1' rows = await con.fetch(q, pat) updates = [] for r in rows: new = mapper.text(r["v"]) if new != r["v"]: updates.append((new, r["ctid"])) if not updates: continue print(f" {table}.{col}: {len(updates)} rows") if dry: continue cast = {"jsonb": "::jsonb", "json": "::json", "inet": "::inet", "cidr": "::cidr", "macaddr": "::macaddr"}.get(dt, "") if dt == "ARRAY": udt = await con.fetchval( "SELECT udt_name FROM information_schema.columns " "WHERE table_name = $1 AND column_name = $2", table, col) cast = f"::{udt.lstrip('_')}[]" await con.executemany( f'UPDATE "{table}" SET "{col}" = $1{cast} WHERE ctid = $2', updates) async def reset_users(con, cfg: dict, dry: bool) -> None: sys.path.insert(0, str(Path(cfg["netork_src"]).expanduser())) from netork.core.security import hash_password # noqa: E402 admin = cfg.get("admin_from", "chris") password = cfg.get("admin_password", "netork-demo") users = await con.fetch("SELECT id, username FROM users ORDER BY username") print(f" users: {[u['username'] for u in users]}") if dry: return n = 0 for u in users: if u["username"] == admin: await con.execute( "UPDATE users SET username = 'netork', email = $2, hashed_password = $3, " "totp_secret = NULL, totp_enabled = false, token_version = token_version + 1 " "WHERE id = $1", u["id"], f"netork@{DEMO_DOMAIN}", hash_password(password)) else: n += 1 await con.execute( "UPDATE users SET username = $2, email = $3, hashed_password = $4, " "totp_secret = NULL, totp_enabled = false, is_active = false WHERE id = $1", u["id"], f"operator{n}", f"operator{n}@{DEMO_DOMAIN}", hash_password(os.urandom(16).hex())) # TOTP secrets are gone, so a role that demands MFA would lock everyone out. await con.execute("UPDATE roles SET require_mfa = false") role = await con.fetchval("SELECT id FROM roles WHERE lower(name) IN ('administrator', 'admin') LIMIT 1") if role: await con.execute("UPDATE users SET role_id = $1, is_superuser = true WHERE username = 'netork'", role) print(f" admin '{admin}' is now 'netork' / '{password}'") async def leak_report(con, cfg: dict, originals: list[str]) -> int: # Names are matched as written (FAMILY is a VLAN, "family" a JSON key); # leak_terms and domains in any case. names = [n for n in [*cfg.get("hostnames", {}), *cfg.get("terms", {}), *cfg.get("substrings", {})] if len(n) >= 4] loose = [n for n in [*cfg.get("domains", {}), *cfg.get("leak_terms", [])] if len(n) >= 4] # Postgres has no inline (?i:...), so spell case-insensitivity out: [mM][aA]... def anycase(t: str) -> str: return "".join(f"[{c.lower()}{c.upper()}]" if c.isalpha() else re.escape(c) for c in t) parts = [re.escape(n) for n in names] + [anycase(n) for n in loose] if not parts: return 0 pat = "|".join(parts) found = 0 for t, c, _ in await columns(con): n = await con.fetchval(f'SELECT count(*) FROM "{t}" WHERE "{c}"::text ~ $1', pat) if n: found += n sample = await con.fetchval( f'SELECT substring("{c}"::text from $2) FROM "{t}" WHERE "{c}"::text ~ $1 LIMIT 1', pat, f"(.{{0,30}}(?:{pat}).{{0,30}})") print(f" LEAK {t}.{c}: {n} rows, e.g. …{sample}…") return found async def main() -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--dsn", default=os.environ.get("DEMO_DSN", DEFAULT_DSN)) ap.add_argument("--map", type=Path, default=DEFAULT_MAP) ap.add_argument("--dry-run", action="store_true") ap.add_argument("--report-only", action="store_true", help="only run the leak report") args = ap.parse_args() host = re.search(r"@([^:/]+)", args.dsn) if not host or host.group(1) not in ("127.0.0.1", "localhost", "::1"): sys.exit("Refusing: this only runs against a local copy.") cfg = json.loads(args.map.read_text()) mapper = Mapper(cfg) originals = [*cfg.get("domains", []), *cfg.get("hostnames", {}), *cfg.get("terms", {}), *cfg.get("leak_terms", [])] con = await asyncpg.connect(args.dsn) try: if not args.report_only: async with con.transaction(): print("secrets:") await scrub_secrets(con, args.dry_run) print("users:") await reset_users(con, cfg, args.dry_run) await collect_public(con, mapper) print(f"public addresses seen as addresses: {len(mapper.known_public)}") print("rewriting:") await rewrite(con, mapper, cfg, args.dry_run) print("ipv4 /16 mapping:", json.dumps(mapper.prefix16)) print("public addresses mapped:", len(mapper.public)) if mapper.unmapped_public: top = sorted(mapper.unmapped_public.items(), key=lambda x: -x[1])[:40] print("left as is (versions? add real ones to public_ips in the map):") print(" " + ", ".join(f"{ip} ({n}x)" for ip, n in top)) print("leak report:") n = await leak_report(con, cfg, originals) if not args.report_only and mapper.unmapped_public: print(f" review: {len(mapper.unmapped_public)} public-looking dotted quads left as is (listed above)") print(" clean" if n == 0 else f" {n} rows still match") finally: await con.close() if __name__ == "__main__": asyncio.run(main())