website/scripts/demo/anonymize.py

#!/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"}

# 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:])")
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", {}))
        self.pool16 = iter(cfg.get("ipv4_pool16", [f"10.{n}" for n in range(20, 250, 10)]))
        self.public: dict[str, str] = {}
        # 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-])")
        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 a.is_private:
            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 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:
            self.public[ip] = next(self.public_pool)
        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}"

    # -- whole strings -------------------------------------------------------
    def text(self, s: str) -> str:
        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)
        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}:", "@"]
    for k in [*cfg.get("domains", []), *cfg.get("hostnames", {}), *cfg.get("terms", {})]:
        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)", re.I)
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()))
    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:
    needles = [n for n in originals if len(n) >= 4]
    if not needles:
        return 0
    pat = "|".join(re.escape(n) for n in needles)
    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 $1) FROM "{t}" WHERE "{c}"::text ~* $1 LIMIT 1',
                f"(?i)(.{{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())