napalm-amt/napalm_amt/record.py

"""Record a real AMT's answers as test fixtures: ``python -m napalm_amt.record HOST``.

Makes every read the driver makes and writes each raw answer under the name the tests'
mock serves it by (``<Class>.get.xml``, ``<Class>.pull.xml``, ``<Class>.<Method>.xml``),
plus a ``COMMANDS.txt`` and the unauthenticated landing page with its headers (what
discovery sees). It never switches power and never writes to AMT.

The password comes from ``--password-file`` (a file on the host, the way netOrk keeps
secrets) or the ``AMT_PASSWORD`` environment variable, never from the command line.
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from collections.abc import Sequence
from pathlib import Path

import requests

from napalm_amt.wsman import PORT, WsmanError, WsmanTransport

#: Everything the driver reads: (class, operation). Methods with input are handled below.
READS: list[tuple[str, str]] = [
    ("AMT_GeneralSettings", "Get"),
    ("AMT_SetupAndConfigurationService", "Get"),
    ("AMT_RedirectionService", "Get"),
    ("IPS_KVMRedirectionSettingData", "Get"),
    ("AMT_EthernetPortSettings", "Enumerate"),
    ("AMT_TLSSettingData", "Enumerate"),
    ("CIM_KVMRedirectionSAP", "Enumerate"),
    ("CIM_SoftwareIdentity", "Enumerate"),
    ("CIM_BIOSElement", "Enumerate"),
    ("CIM_Chassis", "Enumerate"),
    ("CIM_Card", "Enumerate"),
    ("CIM_ComputerSystemPackage", "Enumerate"),
    ("CIM_Processor", "Enumerate"),
    ("CIM_Chip", "Enumerate"),
    ("CIM_PhysicalMemory", "Enumerate"),
    ("CIM_MediaAccessDevice", "Enumerate"),
    ("CIM_AssociatedPowerManagementService", "Enumerate"),
    # Not read by the driver yet; recorded to see what this AMT knows.
    ("CIM_Fan", "Enumerate"),
    ("CIM_Sensor", "Enumerate"),
    ("AMT_SetupAndConfigurationService", "GetUuid"),
]
#: Values that are secrets, or close to one, replaced before anything is written.
_REDACT = re.compile(r"(<(\w+:)?(ProvisioningServerOTP|RFBPassword)>)[^<]+(</)")


class _Recorder:
    def __init__(self, out: Path) -> None:
        self.out = out
        self.pages: dict[str, int] = {}
        self.commands: list[str] = []

    def __call__(self, resource: str, operation: str, content: bytes) -> None:
        if operation == "Enumerate" and b"Fault" not in content:
            return  # the context only; the instances come with Pull
        suffix = operation.lower() if operation in ("Get", "Pull", "Enumerate") else operation
        key = f"{resource}.{suffix}"
        self.pages[key] = self.pages.get(key, 0) + 1
        name = f"{key}.xml" if self.pages[key] == 1 else f"{key}.{self.pages[key]}.xml"
        text = _REDACT.sub(r"\1REDACTED\4", content.decode("utf-8", errors="replace"))
        (self.out / name).write_text(text)
        self.commands.append(f"{name:<55} {operation} {resource}")


def record(transport: WsmanTransport, out: Path) -> list[str]:
    """Make every read in ``READS`` plus the event log; return the files written."""
    out.mkdir(parents=True, exist_ok=True)
    recorder = _Recorder(out)
    transport.recorder = recorder
    try:
        for resource, operation in READS:
            try:
                if operation == "Get":
                    transport.get(resource)
                elif operation == "Enumerate":
                    transport.enumerate(resource)
                else:
                    transport.invoke(resource, operation, check=False)
            except WsmanError as exc:
                print(f"{resource} {operation}: {exc}", file=sys.stderr)
        start = transport.invoke("AMT_MessageLog", "PositionToFirstRecord", check=False)
        if start.get("ReturnValue") == "0":
            transport.invoke(
                "AMT_MessageLog",
                "GetRecords",
                [
                    ("IterationIdentifier", start.get("IterationIdentifier", "1")),
                    ("MaxReadRecords", 390),
                ],
                check=False,
            )
    finally:
        transport.recorder = None
    (out / "COMMANDS.txt").write_text(
        "# Recorded with python -m napalm_amt.record\n" + "\n".join(recorder.commands) + "\n"
    )
    return [c.split()[0] for c in recorder.commands]


def _landing_page(host: str, port: int, tls: bool, out: Path) -> None:
    """The page discovery fetches, without credentials, with its status line and headers."""
    url = f"{'https' if tls else 'http'}://{host}:{port}/"
    try:
        # Recorded as discovery sees it: no credentials go out, and AMT's own certificate
        # on 16993 is self-signed unless one was provisioned.
        response = requests.get(url, timeout=10, verify=False, allow_redirects=True)  # nosec B501
    except requests.RequestException as exc:
        print(f"landing page: {exc}", file=sys.stderr)
        return
    headers = "\n".join(f"{k}: {v}" for k, v in response.headers.items())
    (out / "landing_page.txt").write_text(
        f"GET {url} -> {response.status_code}\n{headers}\n\n{response.text}"
    )


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="python -m napalm_amt.record",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("host")
    parser.add_argument("--user", default="admin")
    parser.add_argument("--password-file", type=Path)
    parser.add_argument("--port", type=int, default=PORT)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--no-landing-page", action="store_true")
    parser.add_argument(
        "--insecure", action="store_true", help="do not verify AMT's TLS certificate (port 16993)"
    )
    args = parser.parse_args(argv)

    if args.password_file:
        password = args.password_file.read_text().strip()
    elif os.environ.get("AMT_PASSWORD"):
        password = os.environ["AMT_PASSWORD"]
    else:
        parser.error("give --password-file or set AMT_PASSWORD")

    transport = WsmanTransport(
        args.host, args.user, password, port=args.port, verify=not args.insecure
    )
    try:
        written = record(transport, args.out)
    finally:
        transport.close()
    if not args.no_landing_page:
        _landing_page(args.host, args.port, transport.tls, args.out)
    print(f"{len(written)} files in {args.out}")
    return 0


if __name__ == "__main__":
    sys.exit(main())