Add firewall rule helpers and structured API errors

README.md

@@ -67,6 +67,7 @@ except Exception as e:
 # events = controller.get_unifi_site_event(site_name)
 # rogue_aps = controller.get_unifi_site_rogueap(site_name)
 # networks = controller.get_unifi_site_networkconf(site_name)
+# firewall_rules = controller.get_unifi_site_firewallrule(site_name)
 # report = controller.devices_report(site_names=['site1', 'site2'])
 
 # Exporting data:
@@ -93,6 +94,39 @@ except Exception as e:
 
 ---
 
+## Firewall Rules
+
+Firewall rule support follows the same raw/typed pattern as the rest of the client:
+
+```python
+rules = controller.get_unifi_site_firewallrule("default", raw=False)
+for rule in rules:
+    print(rule.name, rule.ruleset, rule.action, rule.protocol, rule.dst_port)
+```
+
+Mutating helpers are available for callers that deliberately want to manage rules:
+
+```python
+controller.create_unifi_site_firewallrule(
+    "default",
+    {
+        "name": "Block example traffic",
+        "enabled": True,
+        "action": "drop",
+        "ruleset": "WAN_OUT",
+        "protocol": "tcp",
+        "dst_port": "25",
+    },
+)
+```
+
+> **Caution:** Firewall rule endpoints are private UniFi Network APIs. Valid fields,
+> rule index conventions, and controller-side validation can vary across UniFi
+> Network versions. Prefer read-only listing first, keep writes explicit, and inspect
+> `UnifiAPIError.response_json` when the controller rejects a payload.
+
+---
+
 ## Data Models
 
 The library automatically maps JSON API responses to Python data classes located in `unifi_controller_api.models`. Key models include:
@@ -103,6 +137,7 @@ The library automatically maps JSON API responses to Python data classes located
 *   `UnifiClient`: Represents a connected client (wired or wireless).
 *   `UnifiWlanConf`: Represents a Wireless LAN configuration.
 *   `UnifiNetworkConf`: Represents a Network configuration.
+*   `UnifiFirewallRule`: Represents a firewall rule.
 *   `UnifiAlarm`: Represents a controller alarm.
 *   `UnifiEvent`: Represents a controller event.
 *   `UnifiRogueAp`: Represents a detected rogue access point.

tests/test_api_error_details.py

@@ -0,0 +1,101 @@
+import requests
+import pytest
+
+from unifi_controller_api import UnifiController
+from unifi_controller_api.exceptions import UnifiAPIError
+
+
+class ErrorResponse:
+    status_code = 400
+    text = '{"meta":{"rc":"error","msg":"api.err.InvalidValue"},"data":[]}'
+
+    def json(self):
+        return {"meta": {"rc": "error", "msg": "api.err.InvalidValue"}, "data": []}
+
+    def raise_for_status(self):
+        raise requests.HTTPError("400 Client Error", response=self)  # type: ignore[arg-type]
+
+
+class NonJsonErrorResponse:
+    status_code = 502
+    text = "bad gateway"
+
+    def json(self):
+        raise ValueError("not json")
+
+    def raise_for_status(self):
+        raise requests.HTTPError("502 Server Error", response=self)  # type: ignore[arg-type]
+
+
+class NoResponseSession:
+    cookies = {}
+
+    def request(self, method, url, **kwargs):
+        raise requests.ConnectionError("connection failed")
+
+
+class ErrorSession:
+    cookies = {}
+
+    def __init__(self, response):
+        self.response = response
+
+    def request(self, method, url, **kwargs):
+        return self.response
+
+
+def make_controller(session):
+    controller = UnifiController.__new__(UnifiController)
+    controller.controller_url = "https://controller.example"
+    controller.original_controller_url = "https://controller.example"
+    controller.is_udm_pro = False
+    controller.session = session
+    controller.verify_ssl = True
+    controller.auth_retry_enabled = False
+    controller.auth_retry_count = 1
+    controller.auth_retry_delay = 0.1  # type: ignore[assignment]
+    controller.request_timeout = None
+    return controller
+
+
+def test_unifi_api_error_preserves_json_response_details():
+    controller = make_controller(ErrorSession(ErrorResponse()))
+
+    with pytest.raises(UnifiAPIError) as excinfo:
+        controller._invoke_api_call("POST", "https://controller.example/api/test", {})
+
+    err = excinfo.value
+    assert err.method == "POST"
+    assert err.url == "https://controller.example/api/test"
+    assert err.status_code == 400
+    assert err.response_json is not None
+    assert err.response_json["meta"]["msg"] == "api.err.InvalidValue"
+    assert "api.err.InvalidValue" in str(err)
+
+
+def test_unifi_api_error_preserves_non_json_response_text():
+    controller = make_controller(ErrorSession(NonJsonErrorResponse()))
+
+    with pytest.raises(UnifiAPIError) as excinfo:
+        controller._invoke_api_call("PUT", "https://controller.example/api/test", {})
+
+    err = excinfo.value
+    assert err.method == "PUT"
+    assert err.status_code == 502
+    assert err.response_text == "bad gateway"
+    assert err.response_json is None
+    assert "bad gateway" in str(err)
+
+
+def test_unifi_api_error_handles_request_failure_without_response():
+    controller = make_controller(NoResponseSession())
+
+    with pytest.raises(UnifiAPIError) as excinfo:
+        controller._invoke_api_call("DELETE", "https://controller.example/api/test")
+
+    err = excinfo.value
+    assert err.method == "DELETE"
+    assert err.url == "https://controller.example/api/test"
+    assert err.status_code is None
+    assert err.response_json is None
+    assert "connection failed" in str(err)

tests/test_api_response_handling.py

@@ -33,3 +33,16 @@ def test_process_api_response_rejects_missing_data_key():
 def test_process_api_response_raises_api_error_for_unifi_error_payload():
     with pytest.raises(UnifiAPIError, match="api.err.Invalid"):
         process({"meta": {"rc": "error", "msg": "api.err.Invalid"}, "data": []})
+
+
+def test_process_api_response_preserves_request_method_for_error_payload():
+    controller = UnifiController.__new__(UnifiController)
+    response = cast(
+        Any,
+        FakeResponse({"meta": {"rc": "error", "msg": "api.err.Invalid"}, "data": []}),
+    )
+
+    with pytest.raises(UnifiAPIError) as excinfo:
+        controller._process_api_response(response, "/api/test", method="POST")
+
+    assert excinfo.value.method == "POST"

tests/test_firewall_rules.py

@@ -0,0 +1,154 @@
+from unifi_controller_api import UnifiController, UnifiFirewallRule
+
+
+class FakeResponse:
+    status_code = 200
+    text = ""
+
+    def __init__(self, payload):
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+    def raise_for_status(self):
+        return None
+
+
+class FirewallSession:
+    cookies = {}
+
+    def __init__(self):
+        self.get_calls = []
+        self.request_calls = []
+        self.get_payload = {
+            "meta": {"rc": "ok"},
+            "data": [
+                {
+                    "_id": "rule-1",
+                    "name": "Allow DNS",
+                    "enabled": True,
+                    "action": "accept",
+                    "ruleset": "LAN_OUT",
+                    "rule_index": 20000,
+                    "protocol": "tcp_udp",
+                    "dst_port": "53",
+                    "undocumented_field": "preserved",
+                }
+            ],
+        }
+
+    def get(self, url, **kwargs):
+        self.get_calls.append((url, kwargs))
+        return FakeResponse(self.get_payload)
+
+    def request(self, method, url, **kwargs):
+        self.request_calls.append((method, url, kwargs))
+        payload = kwargs.get("json") or {}
+        return FakeResponse(
+            {
+                "meta": {"rc": "ok"},
+                "data": [{"_id": "rule-2", **payload}],
+            }
+        )
+
+
+def make_controller(session=None):
+    controller = UnifiController.__new__(UnifiController)
+    controller.controller_url = "https://controller.example/proxy/network"
+    controller.original_controller_url = "https://controller.example"
+    controller.is_udm_pro = False
+    controller.session = session or FirewallSession()  # type: ignore[assignment]
+    controller.verify_ssl = True
+    controller.auth_retry_enabled = False
+    controller.auth_retry_count = 1
+    controller.auth_retry_delay = 0.1  # type: ignore[assignment]
+    controller.request_timeout = None
+    return controller
+
+
+def test_get_firewall_rules_returns_raw_data_by_default():
+    session = FirewallSession()
+    controller = make_controller(session)
+
+    rules = controller.get_unifi_site_firewallrule("default")
+
+    assert isinstance(rules[0], dict)
+    assert rules[0]["name"] == "Allow DNS"
+    assert session.get_calls[0][0] == (
+        "https://controller.example/proxy/network/api/s/default/rest/firewallrule"
+    )
+
+
+def test_get_firewall_rules_maps_typed_model_and_preserves_extra_fields():
+    controller = make_controller(FirewallSession())
+
+    rules = controller.get_unifi_site_firewallrule("default", raw=False)
+
+    assert isinstance(rules[0], UnifiFirewallRule)
+    assert rules[0].name == "Allow DNS"
+    assert rules[0].dst_port == "53"
+    assert rules[0]._extra_fields["undocumented_field"] == "preserved"
+
+
+def test_get_firewall_rule_by_id_uses_rule_specific_endpoint():
+    session = FirewallSession()
+    controller = make_controller(session)
+
+    controller.get_unifi_site_firewallrule("default", firewall_rule_id="rule-1")
+
+    assert session.get_calls[0][0].endswith(
+        "/api/s/default/rest/firewallrule/rule-1"
+    )
+
+
+def test_create_firewall_rule_posts_payload_and_maps_response():
+    session = FirewallSession()
+    controller = make_controller(session)
+
+    rules = controller.create_unifi_site_firewallrule(
+        "default",
+        {"name": "Block SMTP", "action": "drop"},
+        raw=False,
+        protocol="tcp",
+        dst_port="25",
+    )
+
+    method, url, kwargs = session.request_calls[0]
+    assert method == "POST"
+    assert url.endswith("/api/s/default/rest/firewallrule")
+    assert kwargs["json"] == {
+        "name": "Block SMTP",
+        "action": "drop",
+        "protocol": "tcp",
+        "dst_port": "25",
+    }
+    assert isinstance(rules[0], UnifiFirewallRule)
+    assert rules[0]._id == "rule-2"
+    assert rules[0].dst_port == "25"
+
+
+def test_update_firewall_rule_puts_payload_to_rule_endpoint():
+    session = FirewallSession()
+    controller = make_controller(session)
+
+    controller.update_unifi_site_firewallrule(
+        "default", "rule-1", {"enabled": False}
+    )
+
+    method, url, kwargs = session.request_calls[0]
+    assert method == "PUT"
+    assert url.endswith("/api/s/default/rest/firewallrule/rule-1")
+    assert kwargs["json"] == {"enabled": False}
+
+
+def test_delete_firewall_rule_uses_delete_method():
+    session = FirewallSession()
+    controller = make_controller(session)
+
+    controller.delete_unifi_site_firewallrule("default", "rule-1")
+
+    method, url, kwargs = session.request_calls[0]
+    assert method == "DELETE"
+    assert url.endswith("/api/s/default/rest/firewallrule/rule-1")
+    assert "json" not in kwargs

tests/test_package_smoke.py

@@ -3,7 +3,13 @@
 from importlib import resources
 
 import unifi_controller_api as api
-from unifi_controller_api.models import UnifiDevice, UnifiHealth, UnifiPortConf, UnifiSite
+from unifi_controller_api.models import (
+    UnifiDevice,
+    UnifiFirewallRule,
+    UnifiHealth,
+    UnifiPortConf,
+    UnifiSite,
+)
 
 
 def test_public_imports_are_available():
@@ -12,6 +18,7 @@ def test_public_imports_are_available():
     assert UnifiDevice is not None
     assert UnifiHealth is not None
     assert UnifiPortConf is not None
+    assert api.UnifiFirewallRule is UnifiFirewallRule
 
 
 def test_packaged_device_model_database_is_valid_json():

unifi_controller_api/__init__.py

@@ -8,7 +8,8 @@
 from .api_client import UnifiController
 from .models import (
     UnifiSite, UnifiDevice, LLDPEntry, UnifiClient,
-    UnifiEvent, UnifiAlarm, UnifiWlanConf, UnifiRogueAp, UnifiNetworkConf
+    UnifiEvent, UnifiAlarm, UnifiWlanConf, UnifiRogueAp, UnifiNetworkConf,
+    UnifiFirewallRule
 )
 from .export import export_csv, export_json, to_dict_list
 from .exceptions import (
@@ -30,6 +31,7 @@
     "UnifiWlanConf",
     "UnifiRogueAp",
     "UnifiNetworkConf",
+    "UnifiFirewallRule",
     "export_csv",
     "export_json",
     "to_dict_list",