OPNexus/OPNexus
1
0 Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases 39 Wiki Activity
Files
c662a5842626d46ca159d3a8cc3ad877dba39b97
OPNexus/backend/tests/test_hosts.py
T
Julian Boldt 20fd575420
Release / verify (push) Canceled after 0s
Details
CI / checks (push) Canceled after 0s
Details
CI / frontend (push) Canceled after 0s
Details
CI / backend (push) Canceled after 0s
Details
Release / test (push) Canceled after 0s
Details
Release / publish (backend) (push) Canceled after 0s
Details
Release / publish (frontend) (push) Canceled after 0s
Details
Release / publish (proxy) (push) Canceled after 0s
Details
Release v1.12.0 
Netzwerk-Verwaltung und Übernahme von Hand angelegter Regeln. Alles gegen die
OPNsense-Core-Quellen geprüft und live auf drei Test-Firewalls (26.7.1 und
26.7.6) getestet, siehe Handbuch „Netzwerk, DNS und DHCP“:

- **Neuer Bereich „Netzwerk“:** Inventar je Firewall (Schnittstellen mit
  Adressen und CARP-Zustand, VLANs, virtuelle IPs, Gateways, Routen, DNS-Einträge,
  DHCP) und Änderungen daran: anlegen, bearbeiten, löschen für VLANs, virtuelle
  IPs, Gateways, statische Routen, Schnittstellen-Zuweisung samt IP-Konfiguration,
  Unbound-DNS-Einträge, dnsmasq-Hosts sowie Kea-Subnetze und -Reservierungen.
- **Abgesichert:** Jede Änderung läuft über eine Vorschau mit Diff und einem Token,
  der den Firewall-Stand erfasst. Vor dem Schreiben wird ein Konfig-Backup
  gespeichert (ohne Backup keine Änderung), danach angewendet und zurückgelesen.
  OPNexus erkennt, über welche Schnittstelle es die Firewall erreicht; Änderungen,
  die diese Verbindung kappen können (Verwaltungs-Schnittstelle, Standard-Gateway,
  Standardroute, Route über die Firewall-Adresse), brauchen eine ausdrückliche
  Bestätigung mit Eingabe des Firewall-Namens. Es gibt bewusst **keinen**
  automatischen Rollback: OPNexus kann eine Änderung nicht zurücknehmen, wenn sie
  seine eigene Verbindung trennt. Das Backup vor der Änderung ist der Rückweg.
- **Schnittstellen-Löschen:** OPNsense entfernt dabei selbst alle Regeln, die die
  Schnittstelle nutzen. Die Vorschau listet sie, die Bestätigung ist Pflicht.
  Bereits auf der Firewall vorgemerkte Änderungen blockieren eine neue Änderung.
- **IP-Konfiguration per API gibt es erst ab OPNsense 26.7.6.** Auf 26.7.1 ist die
  Zuweisungsliste nur lesbar; OPNexus erkennt das an den Daten und sperrt die
  Änderung mit Begründung.
- **Von der Firewall übernehmen:** Regeln, die direkt in OPNsense angelegt wurden,
  lassen sich in OPNexus übernehmen, bearbeiten und ausrollen. Die Übernahme schreibt
  nichts auf die Firewall. Regeln, die das Modell verfälschen würde (invertierte
  Quelle/Ziel, mehrere Schnittstellen, keine Quick-Regel, …), werden mit Grund
  abgelehnt. Neu: Regeln haben ein Protokollieren-Flag (`log`, Migration 0003),
  damit es beim Bearbeiten nicht still verloren geht; nicht modellierte Felder
  (Gateway, Zustandsoptionen, …) bleiben auf der Firewall unverändert.
- **Geheimnisse:** Das CARP-Passwort (liefert OPNsense im Klartext) und der
  Kea-DDNS-Schlüssel werden nie zurückgegeben, gespeichert oder auditiert.
- **Host-Adresse änderbar:** Hosts › Bearbeiten hat jetzt das Feld „Adresse“. Nach
  einer Änderung der Verwaltungs-IP muss der Host nicht mehr gelöscht und neu
  angelegt werden; Verlauf und Zuordnungen bleiben erhalten. Meldet eine Änderung
  „angewendet, aber nicht prüfbar“, weist die Oberfläche darauf hin.
- **Fehler aus den Live-Tests, die Mocks nicht gezeigt haben:** Jede Änderung einer
  virtuellen IP muss die Adresse (`network`) wiederholen, sonst leert OPNsense sie
  bzw. lehnt ab; die Prüfung nach dem Anwenden vergleicht jetzt den vollständigen
  Soll-Zustand statt nur der gesendeten Felder. Dazu: Gateway-Priorität als Zahl,
  Routen mit `enabled` statt `disabled`, Beschreibungsmuster der Schnittstellen,
  gesperrte Schnittstellen, Pflicht-Backup vor jeder Änderung.
- **Hinweis zu den Versionen:** Die IP-Konfiguration von Schnittstellen per API gibt
  es erst ab OPNsense 26.7.6. Mit 26.7.5 (die Prod-Firewalls) sind VLANs, virtuelle
  IPs, Gateways, Routen, DNS und DHCP verwaltbar, Schnittstellen nur lesbar.
2026-10-09 21:23:04 +02:00

590 lines
23 KiB
Python
Raw Blame History

import time
import httpx
import pytest
from tests.test_rules import _rule_payload
from tests.test_aliases import _alias_payload, _mock_alias_push
@pytest.mark.asyncio
async def test_delete_host_with_active_rule_deployment_rejected(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
host = (await client.post("/api/hosts", json={"name": "fw-k", "base_url": "https://fw-k.test", "api_key": "k", "api_secret": "s"})).json()
rule = (await client.post("/api/rules", json=_rule_payload([host["id"]]))).json()
respx_mock.post("https://fw-k.test/api/firewall/filter/addRule").mock(return_value=httpx.Response(200, json={"result": "saved", "uuid": "r1"}))
respx_mock.post("https://fw-k.test/api/firewall/filter/apply").mock(return_value=httpx.Response(200, json={"status": "ok"}))
await client.post(f"/api/rules/{rule['id']}/deploy")
await client.post(f"/api/rules/{rule['id']}/confirm")
resp = await client.delete(f"/api/hosts/{host['id']}")
assert resp.status_code == 409, resp.text
respx_mock.post("https://fw-k.test/api/firewall/filter/delRule/r1").mock(return_value=httpx.Response(200, json={"result": "deleted"}))
await client.delete(f"/api/rules/{rule['id']}") # undeploys + removes the rule first
assert (await client.delete(f"/api/hosts/{host['id']}")).status_code == 204
@pytest.mark.asyncio
async def test_delete_host_with_active_alias_deployment_rejected(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
host = (await client.post("/api/hosts", json={"name": "fw-l", "base_url": "https://fw-l.test", "api_key": "k", "api_secret": "s"})).json()
alias = (await client.post("/api/aliases", json=_alias_payload([host["id"]]))).json()
_mock_alias_push(respx_mock, "https://fw-l.test")
await client.post(f"/api/aliases/{alias['id']}/deploy")
await client.post(f"/api/aliases/{alias['id']}/confirm")
resp = await client.delete(f"/api/hosts/{host['id']}")
assert resp.status_code == 409, resp.text
respx_mock.post("https://fw-l.test/api/firewall/alias/delItem/alias-uuid-1").mock(return_value=httpx.Response(200, json={"result": "deleted"}))
await client.delete(f"/api/aliases/{alias['id']}")
assert (await client.delete(f"/api/hosts/{host['id']}")).status_code == 204
@pytest.mark.asyncio
async def test_create_list_delete_host(client, respx_mock):
respx_mock.get("https://fw1.test/api/diagnostics/interface/getVipStatus").mock(
return_value=httpx.Response(200, json={"rows": [{"interface": "wan", "vhid": "1", "status": "MASTER"}]})
)
resp = await client.post(
"/api/hosts",
json={
"name": "fw1",
"base_url": "https://fw1.test",
"api_key": "key123",
"api_secret": "secret123",
"verify_tls": False,
},
)
assert resp.status_code == 201
body = resp.json()
assert body["name"] == "fw1"
assert body["status"] == "online"
assert body["carp_status"] == "MASTER"
host_id = body["id"]
resp = await client.get("/api/hosts")
assert resp.status_code == 200
names = [h["name"] for h in resp.json()]
assert "fw1" in names
resp = await client.delete(f"/api/hosts/{host_id}")
assert resp.status_code == 204
resp = await client.get("/api/hosts")
assert resp.status_code == 200
assert all(h["id"] != host_id for h in resp.json())
@pytest.mark.asyncio
async def test_create_host_duplicate_name_rejected(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
payload = {
"name": "dupe",
"base_url": "https://fw2.test",
"api_key": "k",
"api_secret": "s",
}
resp1 = await client.post("/api/hosts", json=payload)
assert resp1.status_code == 201
resp2 = await client.post("/api/hosts", json=payload)
assert resp2.status_code == 409
@pytest.mark.asyncio
async def test_delete_unknown_host_404(client):
resp = await client.delete("/api/hosts/00000000-0000-0000-0000-000000000000")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_update_host_cluster_name(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-a", "base_url": "https://fw-a.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
assert resp.json()["cluster_name"] is None
resp = await client.patch(f"/api/hosts/{host_id}", json={"cluster_name": "ha-pair-1"})
assert resp.status_code == 200, resp.text
assert resp.json()["cluster_name"] == "ha-pair-1"
resp = await client.get(f"/api/hosts/{host_id}")
assert resp.json()["cluster_name"] == "ha-pair-1"
resp = await client.patch(f"/api/hosts/{host_id}", json={"clear_cluster_name": True})
assert resp.status_code == 200, resp.text
assert resp.json()["cluster_name"] is None
@pytest.mark.asyncio
async def test_live_rules_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-b", "base_url": "https://fw-b.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-b.test/api/firewall/filter/searchRule").mock(
return_value=httpx.Response(
200,
json={
"total": 1,
"rowCount": 1,
"current": 1,
"rows": [
{
"uuid": "abc-1",
"description": "allow ssh",
"action": "pass",
"%action": "Pass",
"direction": "in",
"%direction": "In",
"interface": "lan",
"%interface": "LAN",
"protocol": "tcp",
"%protocol": "TCP",
"source_net": "",
"source_port": "",
"destination_net": "",
"destination_port": "22",
"enabled": "1",
"is_automatic": False,
}
],
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/live-rules")
assert resp.status_code == 200, resp.text
rules = resp.json()
assert len(rules) == 1
rule = rules[0]
assert rule["description"] == "allow ssh"
assert rule["action_label"] == "Pass"
assert rule["interface_label"] == "LAN"
assert rule["source_net"] == "any" # empty string normalized to "any"
assert rule["destination_port"] == "22"
assert rule["enabled"] is True
assert rule["is_automatic"] is False
@pytest.mark.asyncio
async def test_live_rules_host_unreachable_returns_502(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-c", "base_url": "https://fw-c.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-c.test/api/firewall/filter/searchRule").mock(
side_effect=httpx.ConnectError("connection refused")
)
resp = await client.get(f"/api/hosts/{host_id}/live-rules")
assert resp.status_code == 502
@pytest.mark.asyncio
async def test_firmware_info_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-d", "base_url": "https://fw-d.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
# Real shape captured live from an OPNsense 26.x host — the actual
# response also includes a `package` array with hundreds of entries,
# trimmed here since the endpoint only extracts product_id/product_version.
respx_mock.get("https://fw-d.test/api/core/firmware/info").mock(
return_value=httpx.Response(200, json={"product_id": "opnsense", "product_version": "26.7.3_11", "package": []})
)
resp = await client.get(f"/api/hosts/{host_id}/firmware-info")
assert resp.status_code == 200, resp.text
assert resp.json() == {"product_id": "opnsense", "product_version": "26.7.3_11"}
@pytest.mark.asyncio
async def test_firmware_info_host_unreachable_returns_502(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-e", "base_url": "https://fw-e.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-e.test/api/core/firmware/info").mock(
side_effect=httpx.ConnectError("connection refused")
)
resp = await client.get(f"/api/hosts/{host_id}/firmware-info")
assert resp.status_code == 502
@pytest.mark.asyncio
async def test_interfaces_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-f", "base_url": "https://fw-f.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
# Real shape captured live from an OPNsense 25.x host's
# GET /firewall/filter/getRule (no uuid) — trimmed to the `interface`
# field, which is what this endpoint reads.
respx_mock.get("https://fw-f.test/api/firewall/filter/getRule").mock(
return_value=httpx.Response(
200,
json={
"rule": {
"interface": {
"wan": {"value": "WAN", "selected": 0},
"opt1": {"value": "DMZ", "selected": 0},
}
}
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/interfaces")
assert resp.status_code == 200, resp.text
options = resp.json()
assert {"name": "wan", "label": "WAN"} in options
assert {"name": "opt1", "label": "DMZ"} in options
@pytest.mark.asyncio
async def test_interfaces_host_unreachable_returns_502(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-g", "base_url": "https://fw-g.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-g.test/api/firewall/filter/getRule").mock(
side_effect=httpx.ConnectError("connection refused")
)
resp = await client.get(f"/api/hosts/{host_id}/interfaces")
assert resp.status_code == 502
@pytest.mark.asyncio
async def test_aliases_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-h", "base_url": "https://fw-h.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
# Real shape captured live from an OPNsense 25.x host's
# GET /firewall/alias/searchItem — trimmed to the fields this endpoint
# reads (name/type/description/content/enabled); the real response
# also carries hit counters and other fields this endpoint ignores.
respx_mock.get("https://fw-h.test/api/firewall/alias/searchItem").mock(
return_value=httpx.Response(
200,
json={
"total": 2,
"rowCount": 2,
"current": 1,
"rows": [
{"name": "Web_Backends", "type": "host", "%type": "Host(s)", "description": "Webserver Backends", "content": "10.10.120.10", "enabled": "1"},
{"name": "__opt2_network", "type": "internal", "%type": "Network(s)", "description": "DC_MGMT net", "content": "", "enabled": "1"},
],
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/aliases")
assert resp.status_code == 200, resp.text
assert resp.json() == [
{"name": "Web_Backends", "type": "host", "description": "Webserver Backends", "content": "10.10.120.10", "enabled": True},
{"name": "__opt2_network", "type": "internal", "description": "DC_MGMT net", "content": "", "enabled": True},
]
@pytest.mark.asyncio
async def test_aliases_host_unreachable_returns_502(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-i", "base_url": "https://fw-i.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-i.test/api/firewall/alias/searchItem").mock(
side_effect=httpx.ConnectError("connection refused")
)
resp = await client.get(f"/api/hosts/{host_id}/aliases")
assert resp.status_code == 502
@pytest.mark.asyncio
async def test_protocols_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-j", "base_url": "https://fw-j.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
# Real shape captured live from an OPNsense 25.x host's
# GET /firewall/filter/getRule — trimmed to a few of the ~140 entries
# (protocol names are IANA constants, not host-specific config).
respx_mock.get("https://fw-j.test/api/firewall/filter/getRule").mock(
return_value=httpx.Response(
200,
json={
"rule": {
"protocol": {
"any": {"value": "any", "selected": 1},
"TCP": {"value": "TCP", "selected": 0},
"TCP/UDP": {"value": "TCP/UDP", "selected": 0},
"GRE": {"value": "GRE", "selected": 0},
}
}
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/protocols")
assert resp.status_code == 200, resp.text
assert {"name": "TCP/UDP", "label": "TCP/UDP"} in resp.json()
assert {"name": "GRE", "label": "GRE"} in resp.json()
@pytest.mark.asyncio
async def test_certificates_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-m", "base_url": "https://fw-m.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
now = int(time.time())
respx_mock.get("https://fw-m.test/api/trust/cert/search").mock(
return_value=httpx.Response(
200,
json={
"total": 3,
"rowCount": 3,
"current": 1,
"rows": [
{
"uuid": "cert-expired",
"descr": "old-webgui",
"commonname": "fw-primary.local",
"caref": "",
"%caref": "",
"valid_from": str(now - 400 * 86400),
"valid_to": str(now - 5 * 86400),
"in_use": "1",
"rfc3280_purpose": "id-kp-serverAuth",
},
{
"uuid": "cert-soon",
"descr": "webgui",
"commonname": "fw-primary.local",
"caref": "ca-uuid-1",
"%caref": "Internal CA",
"valid_from": str(now - 300 * 86400),
"valid_to": str(now + 10 * 86400),
"in_use": "1",
"rfc3280_purpose": "id-kp-serverAuth",
},
{
"uuid": "cert-valid",
"descr": "openvpn-server",
"commonname": "vpn.example.com",
"caref": "",
"%caref": "",
"valid_from": str(now - 10 * 86400),
"valid_to": str(now + 400 * 86400),
"in_use": "0",
"rfc3280_purpose": "id-kp-serverAuth",
},
],
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/certificates")
assert resp.status_code == 200, resp.text
certs = {c["uuid"]: c for c in resp.json()}
expired = certs["cert-expired"]
assert expired["status"] == "expired"
assert expired["issuer"] == "Selbstsigniert"
assert expired["in_use"] is True
soon = certs["cert-soon"]
assert soon["status"] == "expiring_soon"
assert soon["issuer"] == "Internal CA"
valid = certs["cert-valid"]
assert valid["status"] == "valid"
assert valid["common_name"] == "vpn.example.com"
assert valid["in_use"] is False
@pytest.mark.asyncio
async def test_cas_parses_real_shape(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-ca", "base_url": "https://fw-ca.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
now = int(time.time())
respx_mock.get("https://fw-ca.test/api/trust/ca/search").mock(
return_value=httpx.Response(
200,
json={
"total": 1,
"rowCount": 1,
"current": 1,
"rows": [
{
"uuid": "ca-uuid-1",
# OpenVPN cert/ca and Trust-Store caref reference this id, not the uuid
"refid": "6ac8a49fa1589",
"descr": "Internal CA",
"commonname": "Internal Root CA",
"valid_from": str(now - 300 * 86400),
"valid_to": str(now + 400 * 86400),
}
],
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/cas")
assert resp.status_code == 200, resp.text
cas = resp.json()
assert len(cas) == 1
assert cas[0]["uuid"] == "ca-uuid-1"
assert cas[0]["refid"] == "6ac8a49fa1589"
assert cas[0]["common_name"] == "Internal Root CA"
assert cas[0]["status"] == "valid"
@pytest.mark.asyncio
async def test_certificates_csr_pending_returns_unknown_status(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-n", "base_url": "https://fw-n.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
# A not-yet-signed CSR entry: actionPostLoadingEvent() only parses X.509
# fields when `crt`/`csr` is non-empty, so these come back blank.
respx_mock.get("https://fw-n.test/api/trust/cert/search").mock(
return_value=httpx.Response(
200,
json={
"total": 1,
"rowCount": 1,
"current": 1,
"rows": [
{
"uuid": "csr-pending",
"descr": "pending-request",
"commonname": "",
"caref": "",
"%caref": "",
"valid_from": "",
"valid_to": "",
"in_use": "0",
}
],
},
)
)
resp = await client.get(f"/api/hosts/{host_id}/certificates")
assert resp.status_code == 200, resp.text
cert = resp.json()[0]
assert cert["status"] == "unknown"
assert cert["valid_to"] is None
assert cert["days_until_expiry"] is None
assert cert["common_name"] is None
@pytest.mark.asyncio
async def test_certificates_host_unreachable_returns_502(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-o", "base_url": "https://fw-o.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-o.test/api/trust/cert/search").mock(
side_effect=httpx.ConnectError("connection refused")
)
resp = await client.get(f"/api/hosts/{host_id}/certificates")
assert resp.status_code == 502
@pytest.mark.asyncio
async def test_firmware_info_broken_http_framing_gives_actionable_message(client, respx_mock):
"""OPNsense 26.7 mit aktivem kTLS liefert Antworten > ~64 KiB beschädigt
(httpx: RemoteProtocolError). OPNexus soll das als 502 melden und im Log
den bekannten Workaround nennen statt nur „Verbindung fehlgeschlagen“."""
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
resp = await client.post(
"/api/hosts",
json={"name": "fw-ktls", "base_url": "https://fw-ktls.test", "api_key": "k", "api_secret": "s"},
)
host_id = resp.json()["id"]
respx_mock.get("https://fw-ktls.test/api/core/firmware/info").mock(
side_effect=httpx.RemoteProtocolError("malformed chunk footer: bytearray(b'fl')")
)
resp = await client.get(f"/api/hosts/{host_id}/firmware-info")
assert resp.status_code == 502
from app.models.host import OpnsenseHost
from app.services import opnsense_client
host = OpnsenseHost(name="fw-ktls", base_url="https://fw-ktls.test", api_key="k", verify_tls=False)
host.id = __import__("uuid").UUID(host_id)
with pytest.raises(opnsense_client.OpnsenseAPIError) as info:
await opnsense_client.get_firmware_info(host)
assert "kern.ipc.tls.enable=0" in str(info.value)
assert "malformed chunk footer" in str(info.value)
@pytest.mark.asyncio
async def test_host_address_can_be_changed_after_the_firewall_moved(client, respx_mock):
respx_mock.get(url__regex=r".*/getVipStatus").mock(return_value=httpx.Response(200, json={"rows": []}))
host = (await client.post("/api/hosts", json={"name": "fw-moved", "base_url": "https://10.9.9.1", "api_key": "k", "api_secret": "s"})).json()
resp = await client.patch(f"/api/hosts/{host['id']}", json={"base_url": "https://10.9.9.2"})
assert resp.status_code == 200, resp.text
assert resp.json()["base_url"] == "https://10.9.9.2" and resp.json()["id"] == host["id"]
for bad in ("ftp://x", "http://10.9.9.3", "https://user:[email protected]", "https://10.9.9.3/?a=b"):
assert (await client.patch(f"/api/hosts/{host['id']}", json={"base_url": bad})).status_code == 422, bad
assert next(h for h in (await client.get("/api/hosts")).json() if h["id"] == host["id"])["base_url"] == "https://10.9.9.2"
Reference in new issue View Git Blame Copy Permalink