Christian ManivongandClaude Opus 5.5
369f66afdc
- shots.py crops to a region (clip) or to what one or more elements cover (element, pad), per-shot viewport; capture.py writes the published sizes to src/data/screenshots.json so the page reserves the right space. - anonymize.py no longer empties secrets that netOrk compares with each other (Wi-Fi keys on an SSID against the key read from the access point). Emptying them invented passphrase "drift" that never existed; a keyed hash keeps equal equal, reverses nothing, and its key lives for one run. - scripts/check/site.py checks the built site in both languages at four widths: sideways overflow, one h1, images with alt and size, console errors, requests to other origins, links to unknown routes, old-URL redirects, language detection, and word counts against the budgets. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
471 lines
22 KiB
Python
Executable File
471 lines
22 KiB
Python
Executable File
#!/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 secrets
|
|
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"(?<![\d.])((?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})(?!\d|\.\d)")
|
|
MAC = re.compile(r"(?<![0-9A-Fa-f:-])([0-9A-Fa-f]{2}([:-])(?:[0-9A-Fa-f]{2}\2){4}[0-9A-Fa-f]{2})(?![0-9A-Fa-f:-])")
|
|
MAC_DOT = re.compile(r"(?<![0-9A-Fa-f.])([0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4})(?![0-9A-Fa-f.])")
|
|
IPV6 = re.compile(r"(?<![0-9A-Fa-f:])((?:[0-9A-Fa-f]{0,4}:){2,7}[0-9A-Fa-f]{0,4})(?![0-9A-Fa-f:])")
|
|
# Reverse zones and PTR names: 8.22.172.in-addr.arpa is 172.22.8.0/24.
|
|
REVERSE = re.compile(r"(?<![\d.])((?:\d{1,3}\.){1,4})in-addr\.arpa", re.I)
|
|
EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@([A-Za-z0-9-]+\.)+[A-Za-z]{2,}")
|
|
|
|
|
|
def h(value: str, n: int) -> 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"(?<![\w-])({alt})(?![\w-])")
|
|
# Plain substrings, for names glued into identifiers (HOME_OFFICE_MGMT_NET).
|
|
self.substrings: dict[str, str] = cfg.get("substrings", {})
|
|
self.domain_re = None
|
|
if self.domains:
|
|
alt = "|".join(re.escape(d) for d in sorted(self.domains, key=len, reverse=True))
|
|
# Lazy prefix, so the longest configured domain wins.
|
|
self.domain_re = re.compile(rf"(?<![\w-])((?:[\w-]+\.)*?)({alt})(?![\w-])", re.I)
|
|
|
|
# -- single values -------------------------------------------------------
|
|
def ipv4(self, ip: str) -> 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 -------------------------------------------------------
|
|
@staticmethod
|
|
def _secret(m: re.Match) -> str:
|
|
name, sep = m.group(1), m.group(2)
|
|
if name in SECRET_JSON_KEEP:
|
|
return m.group(0)
|
|
raw = m.group(0)[m.group(0).index(sep) + len(sep) + 1:-1]
|
|
if not raw or not COMPARED_SECRET.match(name):
|
|
return f'"{name}"{sep}""'
|
|
value = json.loads(f'"{raw}"') # the value as the column would hold it
|
|
return f'"{name}"{sep}"{secret_token(value)}"'
|
|
|
|
def text(self, s: str) -> str:
|
|
s = SECRET_JSON.sub(self._secret, 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"}
|
|
|
|
# Secrets netOrk compares with each other (the Wi-Fi key stored on an SSID against
|
|
# the key read from the access point). Emptying them would invent drift that never
|
|
# existed, so they become a keyed hash instead: equal stays equal, nothing can be
|
|
# reversed, and the key lives only for this run.
|
|
COMPARED_SECRET = re.compile(r"^(passphrase|psk|ft_key|wpa_key|key|sae_password)$", re.I)
|
|
RUN_KEY = secrets.token_hex(32)
|
|
|
|
|
|
def secret_token(value: str) -> str:
|
|
"""Same formula as the SQL in scrub_secrets: md5(run key || value)."""
|
|
return "demo-" + hashlib.md5((RUN_KEY + value).encode()).hexdigest()[:16]
|
|
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
|
|
n = await con.fetchval(f'SELECT count(*) FROM "{t}" WHERE "{c}" IS NOT NULL')
|
|
if not n:
|
|
continue
|
|
if COMPARED_SECRET.match(c) and dt in ("text", "character varying"):
|
|
print(f" {t}.{c}: {n} replaced by keyed hash")
|
|
if not dry:
|
|
await con.execute(
|
|
f'UPDATE "{t}" SET "{c}" = \'demo-\' || left(md5($1 || "{c}"), 16) '
|
|
f'WHERE "{c}" IS NOT NULL AND "{c}" <> \'\'', RUN_KEY)
|
|
continue
|
|
value = "NULL" if nullable == "YES" else ("'{}'" if dt in ("jsonb", "json") else "''")
|
|
if dt == "bytea" and nullable != "YES":
|
|
value = "''::bytea"
|
|
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())
|