napalm-qnap-qts/tools/sanitize.py

#!/usr/bin/env python3
"""Scrub identifying data out of harvested QNAP output before it becomes a fixture.

Fixtures live in a git repo forever. Serial numbers, MACs, public IPs and
hostnames from a real NAS have no business in one, and a reviewer cannot be
expected to spot every one of them by eye.

Replacements are stable within a run — the same serial always becomes the same
placeholder — so cross-references between files (a disk serial appearing in both
qcli_storage and get_hd_smartinfo) survive and the fixtures stay coherent.

    ./tools/sanitize.py tools/harvest-out/qts5
    ./tools/sanitize.py tools/harvest-out/qts5 --extra-host mynas --in-place
"""

from __future__ import annotations

import argparse
import pathlib
import re
import sys

# QNAP serials are alphanumeric runs; matching them generically would eat model
# names, so they are found via their labelled context instead.
SERIAL_CONTEXT = re.compile(
    r"(?i)\b(serial(?:\s*number)?|serial_no|sn)\b(\s*[:=]\s*|\s+)([A-Z0-9][A-Z0-9\-]{5,})"
)
DISK_SERIAL_FIELD = re.compile(r'(?i)\bSERIAL="([^"]*)"')

# One alternation rather than three passes. A MAC is also a valid match for the
# IPv6 pattern, so separate passes let the second one rewrite what the first
# just produced — and the placeholder MAC came back out as an IPv6 address.
# Alternation order is the precedence: most specific first.
ADDRESS = re.compile(
    r"(?P<mac>\b(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\b)"
    r"|(?P<ipv6>\b(?:[0-9A-Fa-f]{0,4}:){3,7}[0-9A-Fa-f]{0,4}\b)"
    r"|(?P<ipv4>\b(?:\d{1,3}\.){3}\d{1,3}\b)"
)

# Addresses that carry no information about the owner and make fixtures harder
# to read if replaced.
KEEP_IPS = {"0.0.0.0", "127.0.0.1", "255.255.255.255"}


class Scrubber:
    def __init__(self, extra_hosts: list[str] | None = None) -> None:
        self._maps: dict[str, dict[str, str]] = {}
        self._extra_hosts = [h for h in (extra_hosts or []) if h]

    def _placeholder(self, kind: str, value: str, template: str) -> str:
        bucket = self._maps.setdefault(kind, {})
        if value not in bucket:
            bucket[value] = template.format(n=len(bucket) + 1)
        return bucket[value]

    def _serial(self, value: str) -> str:
        return self._placeholder("serial", value, "SERIAL{n:03d}XXXX")

    def _mac(self, value: str) -> str:
        n = self._placeholder("mac", value.lower(), "{n}")
        return f"00:11:22:33:44:{int(n):02x}"

    def _ip(self, value: str) -> str:
        return self._placeholder("ip", value, "192.0.2.{n}")

    def _address(self, match: re.Match) -> str:
        if match.group("mac"):
            return self._mac(match.group("mac"))
        if match.group("ipv6"):
            return "2001:db8::1"
        value = match.group("ipv4")
        return value if value in KEEP_IPS else self._ip(value)

    def scrub(self, text: str) -> str:
        text = SERIAL_CONTEXT.sub(
            lambda m: f"{m.group(1)}{m.group(2)}{self._serial(m.group(3))}", text
        )
        text = DISK_SERIAL_FIELD.sub(lambda m: f'SERIAL="{self._serial(m.group(1))}"', text)
        text = ADDRESS.sub(self._address, text)
        for host in self._extra_hosts:
            text = re.sub(rf"(?i)\b{re.escape(host)}\b", "testnas", text)
        return text


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("directory", type=pathlib.Path)
    ap.add_argument(
        "--extra-host",
        action="append",
        default=[],
        help="hostname or share name to replace with 'testnas' (repeatable)",
    )
    ap.add_argument(
        "--in-place",
        action="store_true",
        help="overwrite the harvested files instead of writing a .sanitised sibling dir",
    )
    args = ap.parse_args(argv)

    if not args.directory.is_dir():
        print(f"not a directory: {args.directory}", file=sys.stderr)
        return 2

    scrubber = Scrubber(args.extra_host)
    dest = (
        args.directory
        if args.in_place
        else args.directory.with_name(args.directory.name + ".sanitised")
    )
    dest.mkdir(parents=True, exist_ok=True)

    count = 0
    for path in sorted(args.directory.glob("*.txt")):
        cleaned = scrubber.scrub(path.read_text(errors="replace"))
        (dest / path.name).write_text(cleaned)
        count += 1

    replaced = {k: len(v) for k, v in scrubber._maps.items()}
    print(f"sanitised {count} file(s) -> {dest}")
    print(f"replaced: {replaced or 'nothing'}")
    print("Read the output before committing — this catches patterns, not judgement.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())