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.
2026-07-15 12:26:36 +02:00
parent 62424cfd71
commit 26470676ce
2 changed files with 120 additions and 0 deletions
@@ -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.

@@ -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()
 # ---------------------------------------------------------------------------