Writing a collector

A collector knows how to obtain a platform's artifacts, once per transport. It never imports Netmiko or httpx directly: it receives a session conforming to the protocols of backupmssp.transports. That is what makes it possible to exercise it against captured outputs, with no device.

1. Declare the platform

In domain/platforms.py, add an entry to the registry with its artifacts and its normalisation rules:

PlatformId.ACME_OS: PlatformDef(
    id=PlatformId.ACME_OS,
    label="ACME OS", vendor="ACME",
    transports=(Transport.API, Transport.SSH),
    preferred=Transport.API,
    netmiko_device_type="acme_os",
    artifacts=(
        ArtifactSpec("running-config", "Running configuration",
                     ArtifactClass.TEXT,
                     scrub=(r"^! generated on ", r"^checksum: "),
                     suffix=".cfg"),
    ),
)

The normalisation patterns are the part that decides whether the tool is usable: without them, every run produces a difference and nobody reads the alerts any more.

What normalisation touches — and does not touch. It serves two purposes: the fingerprint that detects change, and the comparison of two versions. It never touches the text that is kept: the platform stores and returns the raw text (canonical(), line endings unified, nothing removed). A set passwd ENC … line is volatile for the fingerprint — FortiOS re-encrypts every secret with a fresh salt on every read — and indispensable to restoring.

Two forms of pattern: scrub, one line at a time, and scrub_blocks, (start, end) pairs that exclude a whole block — the case of encrypted private keys, whose base64 changes entirely on every read, whereas certificates are stable and stay compared.

An OS version can change the HTTP method. FortiOS's backup endpoint answers GET up to 7.x and requires POST from 8.0, where GET returns 405. The collector follows the code the device returns rather than a version table to maintain: TransportError.status carries the HTTP code, and 405 says precisely “not this method”.

2. Write the collector

@register
class AcmeCollector(Collector):
    platform_id = PlatformId.ACME_OS

    def make_api_session(self, target):
        return HttpxApi(target, base_path="/api/v1")

    def collect_ssh(self, target, session, workdir):
        body = session.send("show running-config")
        return [self._text(Transport.SSH, "running-config", body)]

    def collect_api(self, target, session, workdir):
        body = session.get("/config/running")
        return [self._text(Transport.API, "running-config", body)]

Raise TransportUnavailable when the transport cannot be established — that is what triggers the switch to the fallback. Raise TransportError when a command fails on a transport that was nonetheless established: replaying on the other transport would give nothing better and would hide a genuine defect.

3. Capture the outputs

A single trial window is enough to harvest captures reusable indefinitely. The files live in tests/fixtures/<platform>/<normalised command>.txt, and tests/fixtures/<platform>/api/<key>.txt for the API.

The name derives from the command. A command carrying a name generated at run time is normalised beforehand — see ReplaySsh.DEFAULT_ALIASES. Sensitive parameters (key, token, password) are excluded from API keys: rotating a secret does not invalidate the captures.

4. Test

def test_collecte_acme(replay, tmp_path):
    outcome, artifacts = CollectorRunner(replay).run(
        target_for("acme_os"), preference=Transport.SSH, workdir=tmp_path)
    assert outcome.transport_used == "ssh"

And above all, the test that matters: two reads of an identical configuration at two different moments must give the same fingerprint, while a real change must produce a different one. See tests/test_normalisation.py.

5. Check the output rather than keeping it as it comes

A truncated output looks like a valid output. On a FortiGate, two consecutive reads returned 1.4 MB and then 997 bytes: a prompt or a pagination badly recognised cuts the output short without any error being raised. Kept, that output becomes a perfectly plausible “modified” version.

A collector must therefore check the document's expected shape before returning it. For FortiOS, the configuration begins with #config-version= and ends with end; any departure raises TransportError.

The same check applies to the API path: an error body returned with a 200 code would otherwise become a “configuration”.

What not to do in a template

Jinja escapes automatically for the HTML context. Interpolating a Python value straight into JavaScript therefore produces entities — an apostrophe becomes an HTML entity — and the entire <script> block stops being parseable. Every function it defines disappears, with no server error and no trace in the logs: a button that does nothing at all.

The only filter whose output is marked safe is tojson:

{# wrong: breaks the script #}    const t = {{ liste }};
{# right #}                       const t = {{ liste | tojson }};

tests/test_web.py has each page's JavaScript parsed by node --check and checks that every function called from an onclick is indeed defined.