OPNsense: manage aliases, rules and NAT from Python

In short. Create an API key, talk to /api/firewall/... with basic auth, and treat every write as three calls: write, apply, read back. Put a managed-by marker in every description so the tool can find and remove its own work. Make creates idempotent by searching for the marker plus a signature before adding. Then call the HA sync yourself, because the API will not. Interfaces stay in the GUI.

How do I add a firewall rule to OPNsense from a script

The API covers aliases, filter rules, port forwards and outbound NAT through the MVC endpoints under /api/firewall/. What it does not do is apply anything by itself, tell you whether the write landed in the shape you intended, or push the change to the backup node of an HA pair. Each of those is a call you make. Once the pattern is in a small wrapper it is boring, which is the goal.

What you need

Steps

  1. Build the client. One session, basic auth with the key as the username and the secret as the password, a base URL per node.
import hashlib, json, requests

class Fw:
    def __init__(self, host, key, secret, verify=True):
        self.base = f"https://{host}/api"
        self.s = requests.Session()
        self.s.auth = (key, secret)
        self.s.verify = verify

    def get(self, path, **params):
        r = self.s.get(f"{self.base}/{path}", params=params, timeout=30)
        r.raise_for_status()
        return r.json()

    def post(self, path, body=None):
        r = self.s.post(f"{self.base}/{path}", json=body or {}, timeout=30)
        r.raise_for_status()
        return r.json()
  1. Decide on a marker. Every object the tool creates carries managed-by:estate-tool in its description, followed by a short signature of the inputs. The marker is how the tool finds its own work later and how a human reading the rule list knows not to edit it by hand.
MARK = "managed-by:estate-tool"

def signature(**fields):
    raw = json.dumps(fields, sort_keys=True).encode()
    return hashlib.sha256(raw).hexdigest()[:12]

def description(label, **fields):
    return f"{label} [{MARK} {signature(**fields)}]"
  1. Create an alias, apply, and read it back. The alias endpoints are firewall/alias/addItem, firewall/alias/reconfigure to apply, and firewall/alias/getItem/<uuid> to read.
def ensure_alias(fw, name, hosts):
    desc = description(name, hosts=sorted(hosts))
    found = fw.get("firewall/alias/searchItem", searchPhrase=MARK)["rows"]
    for row in found:
        if row["name"] == name and desc in row["description"]:
            return row["uuid"]                      # already there, same inputs
    body = {"alias": {"enabled": "1", "name": name, "type": "host",
                      "content": "\n".join(sorted(hosts)), "description": desc}}
    uuid = fw.post("firewall/alias/addItem", body)["uuid"]
    fw.post("firewall/alias/reconfigure")
    back = fw.get(f"firewall/alias/getItem/{uuid}")["alias"]
    assert set(back["content"].split("\n")) == set(hosts), "alias read-back differs"
    return uuid

The search by marker then compare is the idempotency check. Run the tool twice and the second run creates nothing. Change a host and the signature changes, so the tool sees a different object and can update rather than duplicate.

  1. Create a filter rule the same way. The endpoints are firewall/filter/addRule, firewall/filter/apply and firewall/filter/getRule/<uuid>. Field values are the same strings the GUI uses: an interface's assigned name (wan, opt3), an alias name as a network, inet for IPv4.
def ensure_rule(fw, interface, dst_alias, port, label):
    desc = description(label, interface=interface, dst=dst_alias, port=port)
    for row in fw.get("firewall/filter/searchRule", searchPhrase=MARK)["rows"]:
        if desc in row["description"]:
            return row["uuid"]
    body = {"rule": {"enabled": "1", "action": "pass", "interface": interface,
                     "direction": "in", "ipprotocol": "inet", "protocol": "TCP",
                     "source_net": "any", "destination_net": dst_alias,
                     "destination_port": str(port), "description": desc}}
    uuid = fw.post("firewall/filter/addRule", body)["uuid"]
    fw.post("firewall/filter/apply")
    back = fw.get(f"firewall/filter/getRule/{uuid}")["rule"]
    assert back["destination_port"] == str(port), "rule read-back differs"
    return uuid

Port forwards and outbound NAT follow the same shape under firewall/source_nat and firewall/one_to_one, and the port forward endpoints in 25.7 live beside them; check /api/core/menu/search or the API reference for your release for the exact controller names, because they were still moving when this was written.

  1. Apply to the backup first, then the master, then sync. The API write lands on one node. On an HA pair nothing pushes it to the other. The order that has not bitten me: write and verify on the backup, write and verify on the master, then run the sync on the master. The sync is configctl filter sync on the master's shell; there is no stable API call for it, so the tool runs it over SSH with a key that can do only that.
for node in (backup, master):
    alias = ensure_alias(node, "T1_INGRESS", ["10.1.101.10"])
    ensure_rule(node, "wan", "T1_INGRESS", 443, "tenant1 ingress https")
run_ssh(master_host, "configctl filter sync")
  1. Make removal find everything. To retire an object, search for the marker, delete by uuid with delRule or delItem, apply, and then search again to confirm the row is gone. The tool refuses to delete an alias while any rule still references it, which it checks by searching the rule set for the alias name, including rules it did not create.

Verify it worked

The read-back assertions are the first check. The second is pfctl -sr | grep estate-tool on each node, which shows the rules as pf loaded them, not as the config file describes them. The third is the drift check from the HA post: fetch the rules from both nodes and diff. A tool that writes to one node and forgets the sync shows up there within fifteen minutes.

Gotchas

FAQ

Can I use the filter savepoint and rollback features? Yes, and for a tool that runs unattended they are worth it: take a savepoint before the apply, and if the read-back fails, revert to it. The endpoints are under firewall/filter/ alongside apply. Read your release's API reference for the exact names.

Why a signature in the description rather than a database? Because the firewall is the source of truth for what is on the firewall. A database that says a rule exists is a second thing to keep in sync. The signature lets the tool decide from the firewall alone whether its intent is already there.

Should the tool ever touch rules without the marker? Only to read them. A reference guard that checks whether an alias is still used must look at every rule, including hand-made ones. Modifying or deleting rules the tool did not create is how an "unused" address becomes a 26-hour outage.

Related