From 26470676ce3352069c8863763b45be09885f3c5e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 15 Jul 2026 12:26:36 +0200 Subject: [PATCH] feat(trust): add get_certificates() for Trust store certificate inventory Reads certificates via POST /api/trust/cert/search, normalising each row to {name, issuer, valid_from, valid_to, in_use_by}. Field mapping (Unix timestamps for validity, %caref for the resolved issuer label) verified against a live OPNsense 24.7 instance. Never surfaces crt_payload/ prv_payload/csr_payload -- those carry private key material. --- napalm_opnsense/opnsense.py | 32 ++++++++++++++ tests/unit/test_driver.py | 88 +++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 113068f..5e50a66 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1059,6 +1059,38 @@ class OPNsenseDriver(FirewallDriver): }) return sorted(result, key=lambda x: x["name"].lower()) + def get_certificates(self) -> list[dict[str, Any]]: + """Return certificates from the OPNsense Trust store. + + Calls ``POST /api/trust/cert/search`` and normalises each row to + ``{name, issuer, valid_from, valid_to, in_use_by}`` (Unix + timestamps for the validity fields). Never includes + crt_payload/prv_payload/csr_payload -- those rows carry private key + material and must not leave the Trust store. + """ + try: + data = self._post("/api/trust/cert/search") + except Exception as exc: + logger.warning("get_certificates() failed: %s", exc) + return [] + + def _as_int(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + result: list[dict[str, Any]] = [] + for row in data.get("rows", []): + result.append({ + "name": row.get("commonname") or row.get("descr") or row.get("refid", ""), + "issuer": row.get("%caref") or "", + "valid_from": _as_int(row.get("valid_from")), + "valid_to": _as_int(row.get("valid_to")), + "in_use_by": _as_int(row.get("in_use")), + }) + return result + def get_dhcp_leases(self) -> list[dict[str, Any]]: """Return active DHCP leases from OPNsense. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 64d6d8b..1fdc561 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -972,6 +972,94 @@ class TestGetBgpNeighbors: assert key in peer + +# --------------------------------------------------------------------------- +# get_certificates() +# +# Field names verified 2026-07-15 against a live OPNsense 24.7 instance's +# actual POST /api/trust/cert/search response. Confirmed: valid_from/valid_to +# are Unix timestamps as strings (not a formatted date string), issuer comes +# from the resolved "%caref" label (not the raw "caref" ref-id), and rows +# also carry crt_payload/prv_payload/csr_payload -- get_certificates() must +# never surface those (private key material). +# --------------------------------------------------------------------------- + +CERT_SEARCH_RESPONSE = { + "rows": [ + { + "refid": "69d55db78963e", + "descr": "gw.home.example.com (ACME Client)", + "commonname": "gw.home.example.com", + "caref": "6a24988658ae6", + "%caref": "YR1 (ACME Client)", + "cert_type": "server_cert", + "in_use": "1", + "valid_from": "1780779726", + "valid_to": "1788555725", + "crt_payload": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----\n", + "prv_payload": "-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----\n", + }, + { + "refid": "def456", + "descr": "Internal cert", + "commonname": "", + "caref": "", + "%caref": "", + "cert_type": "server_cert", + "in_use": "0", + "valid_from": "1780000000", + "valid_to": "1790000000", + }, + ] +} + + +class TestGetCertificates: + def test_returns_one_entry_per_row(self, driver): + driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE + certs = driver.get_certificates() + assert len(certs) == 2 + + def test_calls_trust_cert_search(self, driver): + calls = [] + driver._post = lambda path, data=None: calls.append(path) or CERT_SEARCH_RESPONSE + driver.get_certificates() + assert calls == ["/api/trust/cert/search"] + + def test_maps_expected_fields(self, driver): + driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE + cert = driver.get_certificates()[0] + assert cert["name"] == "gw.home.example.com" + assert cert["issuer"] == "YR1 (ACME Client)" + assert cert["valid_from"] == 1780779726 + assert cert["valid_to"] == 1788555725 + assert cert["in_use_by"] == 1 + + def test_falls_back_to_descr_when_commonname_empty(self, driver): + driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE + cert = driver.get_certificates()[1] + assert cert["name"] == "Internal cert" + + def test_missing_caref_defaults_to_empty_string(self, driver): + driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE + cert = driver.get_certificates()[1] + assert cert["issuer"] == "" + + def test_never_includes_private_key_or_cert_payload(self, driver): + driver._post = lambda path, data=None: CERT_SEARCH_RESPONSE + for cert in driver.get_certificates(): + assert "prv_payload" not in cert + assert "crt_payload" not in cert + assert "csr_payload" not in cert + + def test_no_rows_returns_empty_list(self, driver): + driver._post = lambda path, data=None: {"rows": []} + assert driver.get_certificates() == [] + + def test_api_error_returns_empty_list(self, driver): + driver._post = lambda path, data=None: (_ for _ in ()).throw(Exception("404")) + assert driver.get_certificates() == [] + # --------------------------------------------------------------------------- # _post() # ---------------------------------------------------------------------------