From 38834100d1de72f90a294f4ebbbf2231abaecbab Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 26 Sep 2026 07:33:18 +0200 Subject: [PATCH] feat(screenshots): real netOrk screenshots from an anonymized demo copy The site has shown hand-built JSX mockups of the UI so far. This adds the tooling to replace them with screenshots of the real application: - scripts/demo/up.sh restores a pg_dump of a production database into a local Postgres and starts netOrk (a pinned release, default v0.28.0) with only the API and the UI: no worker, no beat, no Redis, a random encryption key. Nothing polls and nothing can reach a device. - scripts/demo/anonymize.py rewrites every text, JSON and address column of every table: domains to example.demo, private IPv4 per /16 with the host part kept, public addresses into the documentation ranges, MACs with the vendor prefix kept, e-mail addresses and configured names. Secrets are emptied by column name, one admin "netork" is left. It refuses non-local databases and ends with a leak report. The real-to-demo name map lives outside the repo. - scripts/screenshots/capture.py drives headless Chromium through a declarative list of pages, logs in to the demo copy by itself, and aborts every non-GET API request, so taking screenshots cannot change anything. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 4 + scripts/demo/README.md | 25 +++ scripts/demo/anonymize.py | 383 +++++++++++++++++++++++++++++++++ scripts/demo/up.sh | 73 +++++++ scripts/screenshots/capture.py | 199 +++++++++++++++++ scripts/screenshots/shots.py | 37 ++++ 6 files changed, 721 insertions(+) create mode 100644 scripts/demo/README.md create mode 100755 scripts/demo/anonymize.py create mode 100755 scripts/demo/up.sh create mode 100644 scripts/screenshots/capture.py create mode 100644 scripts/screenshots/shots.py diff --git a/.gitignore b/.gitignore index 8154ad0..e4587c7 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,7 @@ memory/ # Deploy target + registry token deploy.env + +# Screenshot tooling: bytecode, and database dumps that hold production data +__pycache__/ +*.dump diff --git a/scripts/demo/README.md b/scripts/demo/README.md new file mode 100644 index 0000000..d99bd63 --- /dev/null +++ b/scripts/demo/README.md @@ -0,0 +1,25 @@ +# Demo instance for screenshots + +The website shows real netOrk screens, taken from a local copy of a production +database with every hostname, domain, address, MAC and name replaced. + +``` +pg_dump -Fc ... > netork.dump # on the production host, by hand +scripts/demo/up.sh restore netork.dump # fresh local DB + anonymize.py +scripts/demo/up.sh start # API :8000, UI http://127.0.0.1:5173 +scripts/screenshots/capture.py --list-devices +scripts/screenshots/capture.py --var ap= --var switch= --var server= +``` + +Log in as `netork` / `netork-demo`. + +- Only the API and the UI run. There is no worker, no beat and no Redis, so + nothing polls or reaches a device. Stored credentials are emptied, and the + encryption key is random per start. +- The mapping from real to demo names lives outside the repo in + `~/.config/netork-screenshots/demo-map.json`, because it lists the real names. + Domains become `example.demo`. +- `anonymize.py` ends with a leak report. Read it before taking screenshots, + and look at every image before committing it. +- The dump file itself holds production data: keep it out of the repo and + delete it when done. diff --git a/scripts/demo/anonymize.py b/scripts/demo/anonymize.py new file mode 100755 index 0000000..c04fa96 --- /dev/null +++ b/scripts/demo/anonymize.py @@ -0,0 +1,383 @@ +#!/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"(? 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"(? 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()) diff --git a/scripts/demo/up.sh b/scripts/demo/up.sh new file mode 100755 index 0000000..af8a3e7 --- /dev/null +++ b/scripts/demo/up.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +# Local netOrk demo instance for website screenshots. +# +# up.sh restore fresh demo DB from a pg_dump -Fc file, then anonymize +# up.sh start API on :8000 and UI on :5173 (foreground, Ctrl-C stops) +# up.sh stop stop the demo database container +# +# Only the API and the UI run: no Celery worker, no beat, no Redis. Nothing +# polls, nothing reboots, nothing reaches a device. Stored credentials are +# emptied by anonymize.py and the encryption key is a fresh random one, so +# even a leftover value could not be decrypted. +set -euo pipefail + +HERE="$(cd "$(dirname "$0")" && pwd)" +DEMO="${NETORK_DEMO_DIR:-$HOME/.cache/netork-demo}" +SRC="$DEMO/src" +VENV="${NETORK_VENV:-$HOME/dev/NetOrk/.venv}" +VERSION="${NETORK_DEMO_VERSION:-v0.28.0}" +NETORK_REPO="${NETORK_REPO:-$HOME/dev/NetOrk}" +DB=netork-demo-db +PORT=55432 + +ensure_src() { + if [ ! -d "$SRC/netork" ]; then + mkdir -p "$SRC" + git -C "$NETORK_REPO" archive "$VERSION" | tar -x -C "$SRC" + fi +} + +ensure_db() { + if ! docker ps --format '{{.Names}}' | grep -qx "$DB"; then + docker start "$DB" 2>/dev/null || docker run -d --name "$DB" \ + -p 127.0.0.1:$PORT:5432 -e POSTGRES_DB=netork -e POSTGRES_USER=netork \ + -e POSTGRES_PASSWORD=demo -v netork-demo-pg:/var/lib/postgresql/data postgres:16-alpine + until docker exec "$DB" pg_isready -U netork -q; do sleep 1; done + fi +} + +case "${1:-}" in + restore) + dump="${2:?usage: up.sh restore }" + ensure_src; ensure_db + docker exec "$DB" psql -U netork -d postgres -q \ + -c "DROP DATABASE IF EXISTS netork WITH (FORCE)" -c "CREATE DATABASE netork" + docker exec -i "$DB" pg_restore -U netork -d netork --no-owner --no-privileges < "$dump" \ + || echo "pg_restore reported errors (often only missing roles/extensions); checking ..." + got=$(docker exec "$DB" psql -U netork -tA -c "SELECT version_num FROM alembic_version") + want=$(cd "$SRC" && PATH="$VENV/bin:$PATH" alembic heads 2>/dev/null | awk '{print $1}') + echo "dump schema: $got $VERSION head: $want" + [ "$got" = "$want" ] || echo "WARNING: schema differs from $VERSION; screens may not match the release." + "$VENV/bin/python" "$HERE/anonymize.py" + ;; + start) + ensure_src; ensure_db + [ -d "$SRC/ui/node_modules" ] || (cd "$SRC/ui" && npm ci --no-audit --no-fund) + export DATABASE_URL="postgresql+asyncpg://netork:demo@127.0.0.1:$PORT/netork" + export ENVIRONMENT=development + export SECRET_KEY="$(openssl rand -hex 32)" + export CREDENTIAL_ENCRYPTION_KEY="$("$VENV/bin/python" -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())')" + # Nothing listens on port 1: no task can be queued, so no worker could act. + export REDIS_URL=redis://127.0.0.1:1/0 CELERY_BROKER_URL=redis://127.0.0.1:1/0 CELERY_RESULT_BACKEND=redis://127.0.0.1:1/1 + cd "$SRC" + "$VENV/bin/uvicorn" netork.api.main:app --host 127.0.0.1 --port 8000 & + api=$! + trap 'kill $api 2>/dev/null' EXIT + cd ui && npx vite --host 127.0.0.1 --port 5173 --strictPort + ;; + stop) + docker stop "$DB" + ;; + *) + sed -n '2,12p' "$0"; exit 1 ;; +esac diff --git a/scripts/screenshots/capture.py b/scripts/screenshots/capture.py new file mode 100644 index 0000000..8e109e2 --- /dev/null +++ b/scripts/screenshots/capture.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Take real screenshots of a running netOrk instance for the website. + +Normally that instance is the local demo copy from scripts/demo (anonymized +production data), which this script logs into on its own: + + capture.py --list-devices # prints IDs to pick for --var + capture.py --var ap= --var server= [--only name ...] + +Against a real instance, log in by hand and cover what must not be seen: + + NETORK_URL=https://... capture.py --login + NETORK_URL=https://... capture.py --mask --var ... + +While capturing, every request to the API that is not a GET is aborted, so +taking screenshots cannot change anything on the instance. +""" + +import argparse +import json +import os +import re +import sys +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +from playwright.sync_api import Page, sync_playwright + +from shots import SHOTS + +BASE = os.environ.get("NETORK_URL", "http://127.0.0.1:5173").rstrip("/") +LOCAL = re.match(r"https?://(127\.0\.0\.1|localhost)[:/]", BASE + "/") is not None +# The demo instance's admin (see scripts/demo/anonymize.py). +USER = os.environ.get("NETORK_USER", "netork") +PASSWORD = os.environ.get("NETORK_PASSWORD", "netork-demo") +STATE = Path(os.environ.get( + "NETORK_STATE", Path.home() / ".cache" / "netork-screenshots" / "state.json")) +# One term per line: site names, customer names, domains ... never committed. +MASK_FILE = Path(os.environ.get( + "NETORK_MASK_FILE", Path.home() / ".config" / "netork-screenshots" / "mask.txt")) +OUT = Path(__file__).resolve().parents[2] / "public" / "screenshots" + +VIEWPORT = {"width": 1600, "height": 1000} + +# Any IPv4 address that is not RFC 1918, loopback or link-local. +PUBLIC_IPV4 = re.compile( + r"\b(?!10\.)(?!127\.)(?!169\.254\.)(?!192\.168\.)(?!172\.(?:1[6-9]|2\d|3[01])\.)" + r"(?:25[0-5]|2[0-4]\d|1?\d?\d)(?:\.(?:25[0-5]|2[0-4]\d|1?\d?\d)){3}\b") +EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+") + + +def mask_terms() -> list[str]: + if not MASK_FILE.exists(): + return [] + return [t.strip() for t in MASK_FILE.read_text().splitlines() + if t.strip() and not t.startswith("#")] + + +def login() -> None: + STATE.parent.mkdir(parents=True, exist_ok=True) + with sync_playwright() as p: + browser = p.chromium.launch(headless=False) + ctx = browser.new_context(ignore_https_errors=True, viewport=VIEWPORT) + page = ctx.new_page() + page.goto(f"{BASE}/login") + print("Log in in the browser window (10 minutes) ...", flush=True) + page.wait_for_function( + "() => localStorage.getItem('token') && !location.pathname.startsWith('/login')", + timeout=600_000) + ctx.storage_state(path=STATE) + STATE.chmod(0o600) + browser.close() + print(f"Session saved to {STATE}") + + +def token() -> str: + if LOCAL: + body = urllib.parse.urlencode({"username": USER, "password": PASSWORD}).encode() + try: + with urllib.request.urlopen(f"{BASE}/api/v1/auth/token", body) as res: + return json.load(res)["access_token"] + except urllib.error.URLError as e: + sys.exit(f"Login as {USER} at {BASE} failed: {e} (is scripts/demo/up.sh start running?)") + state = json.loads(STATE.read_text()) + for origin in state.get("origins", []): + for item in origin.get("localStorage", []): + if item["name"] == "token": + return item["value"] + sys.exit("No token in the saved session; run --login first.") + + +def list_devices() -> None: + with sync_playwright() as p: + req = p.request.new_context( + base_url=BASE, ignore_https_errors=True, + extra_http_headers={"Authorization": f"Bearer {token()}"}) + res = req.get("/api/v1/devices/") + if not res.ok: + sys.exit(f"{res.status}: {res.text()[:200]} (session expired? run --login)") + for d in res.json(): + print(f"{d.get('id')} {d.get('driver') or '-':18} " + f"{d.get('device_type') or '-':20} {d.get('hostname')}") + + +def settle(page: Page) -> None: + """Wait until the page has finished loading its data.""" + try: + page.wait_for_load_state("networkidle", timeout=15_000) + except Exception: + pass # pages that poll never go fully idle + try: + page.wait_for_function( + "() => !document.querySelector('.animate-spin, .animate-pulse')", timeout=15_000) + except Exception: + print(" still loading after 15 s, taking the shot anyway") + page.wait_for_timeout(800) + + +def capture(variables: dict[str, str], only: set[str], mask: bool) -> None: + OUT.mkdir(parents=True, exist_ok=True) + terms = mask_terms() + tok = token() if LOCAL else None + blocked: list[str] = [] + + def guard(route): + if route.request.method in ("GET", "HEAD", "OPTIONS"): + route.continue_() + else: + blocked.append(f"{route.request.method} {route.request.url}") + route.abort() + + with sync_playwright() as p: + browser = p.chromium.launch() + ctx = browser.new_context( + storage_state=None if LOCAL else STATE, ignore_https_errors=True, + viewport=VIEWPORT, device_scale_factor=2, color_scheme="dark") + if tok: + ctx.add_init_script(f"localStorage.setItem('token', {json.dumps(tok)})") + ctx.route("**/api/**", guard) + page = ctx.new_page() + for shot in SHOTS: + if only and shot.name not in only: + continue + try: + path = shot.path.format(**variables) + except KeyError as e: + print(f"skip {shot.name}: needs --var {e.args[0]}=") + continue + print(f"{shot.name}: {path}") + page.goto(f"{BASE}{path}") + if page.url.rstrip("/").endswith("/login"): + sys.exit("Session expired; run --login again.") + page.wait_for_selector(shot.wait_for, timeout=20_000) + settle(page) + # The release notes dialog after an upgrade; dismissing it only + # writes localStorage in this throwaway browser context. + got_it = page.get_by_role("button", name="Got it") + if got_it.is_visible(): + got_it.click() + page.wait_for_timeout(300) + masks = [page.locator(s) for s in shot.mask] + if mask: + masks += [page.get_by_text(PUBLIC_IPV4), page.get_by_text(EMAIL)] + masks += [page.get_by_text(t) for t in terms] + clip = None + if shot.height: + clip = {"x": 0, "y": 0, "width": VIEWPORT["width"], "height": shot.height} + page.screenshot(path=OUT / f"{shot.name}.png", full_page=shot.full_page, + clip=clip, mask=masks, mask_color="#334155", animations="disabled") + browser.close() + if blocked: + print("Blocked non-GET requests (nothing was sent):") + for b in sorted(set(blocked)): + print(f" {b}") + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--login", action="store_true", help="log in and save the session") + ap.add_argument("--list-devices", action="store_true", help="print device IDs") + ap.add_argument("--var", action="append", default=[], metavar="NAME=VALUE", + help="fill a {placeholder} in the shot paths") + ap.add_argument("--only", nargs="*", default=[], help="only these shot names") + ap.add_argument("--mask", action="store_true", + help="cover public IPs, e-mails and the mask-file terms (real instances)") + args = ap.parse_args() + if args.login: + login() + elif args.list_devices: + list_devices() + else: + capture(dict(v.split("=", 1) for v in args.var), set(args.only), args.mask) + + +if __name__ == "__main__": + main() diff --git a/scripts/screenshots/shots.py b/scripts/screenshots/shots.py new file mode 100644 index 0000000..736c0e6 --- /dev/null +++ b/scripts/screenshots/shots.py @@ -0,0 +1,37 @@ +"""The screenshots the website uses, as data. + +Each shot is one page of the netOrk UI. `path` may contain `{placeholders}` +that are filled from `--var name=value` on the command line (device IDs +differ per instance, so they are never hard-coded here). Device detail +sections are addressed through the URL hash the UI itself writes +(`#security/assessment`, `#config`, ...), so no clicking is needed. + +`mask` lists extra CSS selectors to cover on top of the automatic masks +(public IPv4 addresses, e-mail addresses, and the terms from the mask file). +""" + +from dataclasses import dataclass, field + + +@dataclass +class Shot: + name: str + path: str + # Selector that must be visible before the shot is taken. + wait_for: str = "main" + mask: list[str] = field(default_factory=list) + full_page: bool = False + # Crop height in CSS pixels; None keeps the viewport height. + height: int | None = None + + +SHOTS: list[Shot] = [ + Shot("devices", "/devices"), + Shot("device-detail", "/devices/{ap}#networking/interfaces"), + Shot("vlans", "/vlans"), + Shot("device-security", "/devices/{server}#security/assessment"), + Shot("vulnerabilities", "/vulnerabilities"), + Shot("dashboard", "/"), + Shot("audit-log", "/audit-log"), + Shot("service-checks", "/monitoring/checks"), +]