vulncheck/tests/test_netdisco_inventory.py

"""Netdisco inventory mapping — run: python tests/test_netdisco_inventory.py

Exercises the real production functions against the real API payload:

  * `NetdiscoClient.get_devices` over the JSON the deviceinventory report
    actually returns (the public demo instance's rows, verbatim, plus the four
    Aruba/HP rows from a customer export);
  * `netdisco_service.aruba_family` — which of the three Aruba firmware lines a
    row is, and the far more important question of when the answer is "none";
  * `netdisco_service.os_and_version` — what is written onto the asset, which
    is what both CVE paths then key off.
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.integrations.netdisco_client import NetdiscoClient
from app.services import netdisco_service as nd
from app.services import app_cve_scanner_service as cpe


# Four rows in the shape a real deviceinventory export has (two ProVision
# switches, two AOS-CX), in the API's own JSON spelling. Names, addresses and
# serials are placeholders; the models and firmware strings are what matters
# and are the genuine ones. `version` is the report's "OS Version" column —
# the firmware string with HPE's two-letter code-line prefix.
EXPORT = [
    {"device_name": "sw-access-01", "device_details": "Chassis", "ip": "192.0.2.25",
     "location": "Example site", "model": "JL557A|2930F-48G-PoE+-4SFP-740W",
     "serial": "SN00000000A1", "vendor": "hp", "os": "hp", "version": "WC.16.11.0016"},
    {"device_name": "sw-access-02", "device_details": "Chassis", "ip": "192.0.2.21",
     "location": "Example site", "model": "J9774A|2530-8G-PoEP",
     "serial": "SN00000000A2", "vendor": "hp", "os": "hp", "version": "YA.16.11.0027"},
    {"device_name": "sw-access-03", "device_details": "Chassis", "ip": "192.0.2.8",
     "location": '""""', "model": "R8N87A", "serial": "SN00000000B1",
     "vendor": "aruba", "os": "arubaos-cx", "version": "PL.10.08.1010"},
    {"device_name": "sw-access-04", "device_details": "Chassis", "ip": "192.0.2.20",
     "location": '""""', "model": "R8N87A", "serial": "SN00000000B2",
     "vendor": "aruba", "os": "arubaos-cx", "version": "PL.10.13.1005"},
]

# A row from the public demo instance, unauthenticated, verbatim.
DEMO_ROW = {"device_name": "leaf01", "vendor": "Cumulus Networks", "location": "leaves",
            "serial": "a0:00:00:00:00:11",
            "version": "Cumulus Linux 3.5.1 (Linux Kernel 4.1.33-1+cl3u11)",
            "model": "3.5.1|VX Chassis", "device_details": "leaf01",
            "ip": "192.168.0.11", "os": "cumulus"}

# A device Netdisco knows nothing about yet — no canonical address, so nothing
# to pin an asset to.
NO_IP_ROW = dict(DEMO_ROW, ip="", device_name="ghost")


def _client_devices(rows):
    """Run the REAL get_devices over a canned API answer."""
    c = NetdiscoClient(host="netdisco.local", api_key="948e17690498f7a3496510f7b3128d1a")
    c._get = lambda path: rows
    try:
        return c.get_devices()
    finally:
        c.close()


def demo():
    # --- the client's own mapping ----------------------------------------
    devices = _client_devices(EXPORT + [DEMO_ROW, NO_IP_ROW])
    # A row without a canonical IP is dropped: Netdisco keys a device on that
    # address, so an asset created from such a row would fork a new one on
    # every sync.
    assert [d["name"] for d in devices] == [
        "sw-access-01", "sw-access-02", "sw-access-03", "sw-access-04", "leaf01"]
    assert devices[0] == {
        "ip": "192.0.2.25", "name": "sw-access-01", "details": "Chassis",
        "location": "Example site", "model": "JL557A|2930F-48G-PoE+-4SFP-740W",
        "serial": "SN00000000A1", "vendor": "hp", "os": "hp",
        "os_version": "WC.16.11.0016"}
    # Netdisco quotes an empty location as a literal '""""' in the CSV export
    # and passes it through the API the same way — an empty location, not a
    # location called '"'.
    assert devices[2]["location"] == ""

    # --- which firmware line each row is ----------------------------------
    fams = [nd.aruba_family(d["os"], d["vendor"], d["os_version"]) for d in devices]
    assert fams == ["aruba-switch", "aruba-switch", "aruba-cx", "aruba-cx", None]

    # The Mobility controllers: os slug and four-component version.
    assert nd.aruba_family("aos-w", "aruba", "8.13.1.0") == "aruba-os"
    assert nd.aruba_family("airos", "aruba", "10.7.2.2") == "aruba-os"

    # --- and when the answer has to be "none" -----------------------------
    # Ubiquiti's AirOS answers the SAME os slug as an Aruba controller. Read as
    # one, a 6.1.7 access point would be compared against ArubaOS 6.x ranges.
    assert nd.aruba_family("airos", "ubiquiti", "6.1.7") is None
    # "hp" is what every HP device answers, not just a ProVision switch. Only
    # the 15.x/16.x firmware lines HPE bounds its advisories by are claimed.
    assert nd.aruba_family("hp", "hp", "WC.16.11.0016") == "aruba-switch"
    assert nd.aruba_family("hp", "hp", "1.2.3") is None
    assert nd.aruba_family("hp", "hp", "H.10.98") is None
    # A CX row whose version came back truncated: two components cannot be
    # bounded against a three-component branch range.
    assert nd.aruba_family("arubaos-cx", "aruba", "10.13") is None
    assert nd.aruba_family("arubaos-cx", "aruba", "") is None
    # A controller version with only three components is not the four-part
    # Mobility scheme — no verdict rather than a guessed one.
    assert nd.aruba_family("aos-w", "aruba", "8.13.1") is None
    # Everything else Netdisco discovers.
    assert nd.aruba_family("cumulus", "Cumulus Networks", "Cumulus Linux 3.5.1") is None
    assert nd.aruba_family("ios", "cisco", "15.2(7)E3") is None

    # --- what lands on the asset -----------------------------------------
    # The canonical family label and the NUMERIC version: the letters are the
    # hardware code line, and no advisory bounds them.
    assert nd.os_and_version(devices[0]) == ("ArubaOS-Switch", "16.11.0016")
    assert nd.os_and_version(devices[1]) == ("ArubaOS-Switch", "16.11.0027")
    assert nd.os_and_version(devices[2]) == ("ArubaOS-CX", "10.08.1010")
    assert nd.os_and_version(devices[3]) == ("ArubaOS-CX", "10.13.1005")
    # A non-Aruba device keeps Netdisco's own strings, vendor first. Vendor
    # first is not cosmetic: Netdisco calls a Cisco switch's OS "ios", and a
    # bare "ios" on an asset is a product name that belongs to Apple.
    assert nd.os_and_version(devices[4]) == (
        "Cumulus Networks cumulus", "Cumulus Linux 3.5.1 (Linux Kernel 4.1.33-1+cl3u11)")
    # Classic IOS still keeps Netdisco's own strings and gets no verdict — it
    # numbers releases "15.2(7)E3", which no version scheme here reads.
    cisco = {"os": "ios", "vendor": "cisco", "os_version": "15.2(7)E3"}
    assert nd.os_and_version(cisco) == ("cisco ios", "15.2(7)E3")
    assert cpe._resolve_os("cisco ios") is None
    # IOS XE and IOS XR DO get a verdict now, and therefore a canonical label —
    # the raw "ios-xe" slug is never what lands on the asset. See
    # tests/test_cisco_firmware.py for the matching itself.
    xe = {"os": "ios-xe", "vendor": "cisco", "os_version": "17.15.4c"}
    assert nd.os_and_version(xe) == ("Cisco IOS XE", "17.15.4c")
    assert cpe._resolve_os("Cisco IOS XE")["cpe"] == "cpe:2.3:o:cisco:ios_xe"
    assert cpe._resolve_os("cisco ios-xe")["label"] == "Cisco IOS XE"
    # …and the platform filter has to agree with it: it decides whether an
    # Apple CVE is even considered for this asset, and it used to answer
    # "iphone_os" for anything with "ios" anywhere in the string.
    assert cpe._os_family("cisco ios") is None
    assert cpe._os_family("ios-xe") is None
    assert cpe._os_family("iOS 18.1") == "iphone_os"
    # …and the version that a Cisco asset carries would otherwise have been
    # readable as an Apple one, which is what makes the anchor above matter.
    assert cpe._clean_version("15.2(7)E3") == "15.2"

    # --- the scan side sees exactly what the sync wrote --------------------
    from app.services import cvelistv5_scan_service as c5
    for d in devices[:4]:
        label, version = nd.os_and_version(d)
        assert c5.aruba_key(label), label
        assert cpe._resolve_os(label)["label"] == label
        assert cpe._clean_version(version) == version

    # --- the client's URL and auth shape ----------------------------------
    # Plain HTTP on port 5000 is the DEFAULT, because Netdisco's own web server
    # speaks no TLS at all; HTTPS is a reverse proxy in front of it.
    c = NetdiscoClient(host="netdisco.local")
    assert c.base_url == "http://netdisco.local:5000"
    c.close()
    c = NetdiscoClient(host="https://netdisco.local/", port=443, use_https=True)
    assert c.base_url == "https://netdisco.local:443"
    c.close()
    # An API key short-circuits the login: it IS what a login returns.
    c = NetdiscoClient(host="nd", api_key="abc", username="u", password="p")
    c.login()
    assert c.api_key == "abc"
    c.close()
    # No credentials at all is a valid configuration — the demo instance
    # answers the report unauthenticated, and login() must not invent a call.
    c = NetdiscoClient(host="nd")
    c.login()
    assert not c.api_key
    c.close()

    print("test_netdisco_inventory: OK")


def test_netdisco_inventory():
    demo()


if __name__ == "__main__":
    demo()