Document opening Waters Guest WiFi at North Attleboro to translate #2
+12
-4
@@ -153,7 +153,15 @@ curl -si -N -H 'Connection: Upgrade' -H 'Upgrade: websocket' \
|
|||||||
|
|
||||||
Then open the page on a phone: for the LAN-only setup on the guest WiFi, for a
|
Then open the page on a phone: for the LAN-only setup on the guest WiFi, for a
|
||||||
published one **with WiFi turned off**. That is the real test. If the page loads
|
published one **with WiFi turned off**. That is the real test. If the page loads
|
||||||
but captions never appear, the WebSocket headers are missing. If phones on
|
but captions never appear, the WebSocket headers are missing. `/operator`
|
||||||
guest WiFi cannot connect at all, look at client isolation on that network
|
returning 404 through the proxy is by design; staff use
|
||||||
before anything else. `/operator` returning 404 through the proxy is by design;
|
`http://10.1.7.39:3100/operator` on the LAN.
|
||||||
staff use `http://10.1.7.39:3100/operator` on the LAN.
|
|
||||||
|
**Guest WiFi on UniFi.** Guest networks block every private range by default,
|
||||||
|
so the proxy is unreachable from them until two things change: a firewall
|
||||||
|
policy that allows the guest zone to reach the proxy on 80/443, and the
|
||||||
|
proxy's range removed from the guest hotspot's *Post-Authorization
|
||||||
|
Restrictions*. The second one is easy to miss: the access points enforce that
|
||||||
|
list themselves, so with only the firewall policy a phone works for a minute or
|
||||||
|
two and then times out. North Attleboro's fix, with scripts, is in
|
||||||
|
[`examples/waters-church-na/`](examples/waters-church-na/README.md) section 3.
|
||||||
|
|||||||
@@ -75,7 +75,33 @@ Known limits:
|
|||||||
so the effect is minor; if audio lag is reported, that is the first thing to
|
so the effect is minor; if audio lag is reported, that is the first thing to
|
||||||
revisit with a hand-written nginx include.
|
revisit with a hand-written nginx include.
|
||||||
|
|
||||||
## 3. After the proxy answers
|
## 3. Waters Guest WiFi at North Attleboro — done 2026-10-15
|
||||||
|
|
||||||
|
The Guest network (VLAN 25, 192.168.16.0/24) is in the UniFi **Hotspot** zone.
|
||||||
|
Two things stood between a phone on Waters Guest and `10.1.1.4`:
|
||||||
|
|
||||||
|
| Blocker | Fix |
|
||||||
|
| --- | --- |
|
||||||
|
| Zone default "Block All Traffic" Hotspot → Internal | User policy "Guest to translate.watersmedia.org (Boba-Fett web)", ALLOW Hotspot → Internal, destination `10.1.1.4` TCP 80 and 443, index 10000 (evaluated first). Created with [`zone-allow-policy.py`](zone-allow-policy.py). |
|
||||||
|
| Guest "Post-Authorization Restrictions" listed 10.0.0.0/8. The list is enforced twice: as system policies on the gateway (index 30001, which the allow rule sits above) **and on the access points**, which no gateway policy can override. Symptom: a fresh guest client worked for about two minutes, then timed out even to `http://10.1.1.4/`. | Removed 10.0.0.0/8 from the list with [`guest-restrictions.py`](guest-restrictions.py). The Integration API refuses to edit the system policies; the script edits the legacy `guest_access` setting (`restricted_subnet_1..3`) and the four gateway policies regenerate from it. 192.168.0.0/16 and 172.16.0.0/12 stay. Guests are still cut off from every other 10.x host by the zone default block. |
|
||||||
|
| The Waters Guest SSID only broadcast from the AP "The Vault" | Added Auditorium Left, Auditorium Right, Main Lobby Entrance, Admin Offices - U7 and Creative Offices - U7 with [`wifi-broadcast-devices.py`](wifi-broadcast-devices.py). |
|
||||||
|
|
||||||
|
Guest clients get DNS from the gateway (DHCP hands out no custom DNS), so the
|
||||||
|
local record resolves for them. The SSID keeps client isolation on, which only
|
||||||
|
separates guest devices from each other, not from the gateway.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python3 docs/examples/waters-church-na/zone-allow-policy.py north-attleboro Hotspot 10.1.1.4 80,443 --name "Guest to translate.watersmedia.org (Boba-Fett web)" --apply
|
||||||
|
python3 docs/examples/waters-church-na/guest-restrictions.py north-attleboro --remove 10.0.0.0/8 --apply
|
||||||
|
python3 docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" --list
|
||||||
|
```
|
||||||
|
|
||||||
|
To undo: delete the allow policy in Settings → Policy Engine (or Firewall) on
|
||||||
|
the NA console, put 10.0.0.0/8 back under the guest hotspot's
|
||||||
|
Post-Authorization Restrictions, and remove the APs with
|
||||||
|
`wifi-broadcast-devices.py --remove`.
|
||||||
|
|
||||||
|
## 4. After the proxy answers
|
||||||
|
|
||||||
1. On the NA LAN open `http://10.1.7.39:3100/operator`, enter the operator PIN,
|
1. On the NA LAN open `http://10.1.7.39:3100/operator`, enter the operator PIN,
|
||||||
and set **Public address** to `http://translate.watersmedia.org` (no trailing
|
and set **Public address** to `http://translate.watersmedia.org` (no trailing
|
||||||
|
|||||||
+145
@@ -0,0 +1,145 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Remove a subnet from the guest "Post-Authorization Restrictions" on one UniFi console.
|
||||||
|
|
||||||
|
UniFi blocks guest (Hotspot zone) clients from the private ranges listed in the
|
||||||
|
guest hotspot's Post-Authorization Restrictions. That list is enforced in two
|
||||||
|
places: as system firewall policies on the gateway (one per zone pair, which a
|
||||||
|
user policy can sit above) and on the access points themselves, which a gateway
|
||||||
|
policy cannot override. To let guests reach one internal host you therefore
|
||||||
|
remove its range from the restriction list and rely on the zone firewall's
|
||||||
|
default Hotspot -> Internal block plus an explicit allow (zone-allow-policy.py).
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
docs/examples/waters-church-na/guest-restrictions.py north-attleboro --show
|
||||||
|
docs/examples/waters-church-na/guest-restrictions.py north-attleboro --remove 10.0.0.0/8 # dry run
|
||||||
|
docs/examples/waters-church-na/guest-restrictions.py north-attleboro --remove 10.0.0.0/8 --apply
|
||||||
|
|
||||||
|
It updates every "Post-Authorization Restrictions" policy whose source is the
|
||||||
|
Hotspot zone through the Network Integration API, then the legacy
|
||||||
|
guest_access setting (restricted_subnet_1..3) that the access points read.
|
||||||
|
|
||||||
|
Needs the per-console Network API key at ~/.config/unifi/<site>-network-api-key
|
||||||
|
(see unifi-dns-record.py). The key is only ever sent in the X-API-KEY header.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import ssl
|
||||||
|
import sys
|
||||||
|
import urllib.error
|
||||||
|
import urllib.request
|
||||||
|
|
||||||
|
SITES = {
|
||||||
|
"apollo-beach": "10.3.0.1",
|
||||||
|
"north-attleboro": "10.1.0.1",
|
||||||
|
"woonsocket": "10.2.0.1",
|
||||||
|
"norwood": "10.4.0.1",
|
||||||
|
"southcoast": "10.5.0.1",
|
||||||
|
}
|
||||||
|
KEY_DIR = os.path.expanduser("~/.config/unifi")
|
||||||
|
TIMEOUT = 20
|
||||||
|
POLICY_NAME = "Post-Authorization Restrictions"
|
||||||
|
|
||||||
|
|
||||||
|
class Console:
|
||||||
|
def __init__(self, name: str):
|
||||||
|
self.name = name
|
||||||
|
self.gateway = SITES[name]
|
||||||
|
with open(os.path.join(KEY_DIR, f"{name}-network-api-key")) as f:
|
||||||
|
self.key = f.read().strip()
|
||||||
|
self.ctx = ssl.create_default_context()
|
||||||
|
self.ctx.check_hostname = False
|
||||||
|
self.ctx.verify_mode = ssl.CERT_NONE
|
||||||
|
|
||||||
|
def request(self, method: str, path: str, body: dict | None = None):
|
||||||
|
url = f"https://{self.gateway}/proxy/network{path}"
|
||||||
|
data = json.dumps(body).encode() if body is not None else None
|
||||||
|
req = urllib.request.Request(url, data=data, method=method, headers={
|
||||||
|
"X-API-KEY": self.key, "Accept": "application/json",
|
||||||
|
"Content-Type": "application/json"})
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req, timeout=TIMEOUT, context=self.ctx) as r:
|
||||||
|
raw = r.read().decode()
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
raise SystemExit(f"{method} {path} -> HTTP {e.code}: {e.read().decode(errors='replace')[:600]}")
|
||||||
|
return json.loads(raw) if raw else None
|
||||||
|
|
||||||
|
def site_id(self) -> str:
|
||||||
|
return self.request("GET", "/integration/v1/sites")["data"][0]["id"]
|
||||||
|
|
||||||
|
def zones(self, sid: str) -> dict[str, str]:
|
||||||
|
return {z["id"]: z["name"] for z in self.request("GET", f"/integration/v1/sites/{sid}/firewall/zones?limit=200")["data"]}
|
||||||
|
|
||||||
|
def restriction_policies(self, sid: str, zones: dict[str, str]) -> list[dict]:
|
||||||
|
pols = self.request("GET", f"/integration/v1/sites/{sid}/firewall/policies?limit=500")["data"]
|
||||||
|
return [p for p in pols if p.get("name") == POLICY_NAME and zones.get(p["source"]["zoneId"]) == "Hotspot"]
|
||||||
|
|
||||||
|
def guest_access(self) -> dict:
|
||||||
|
for s in self.request("GET", "/api/s/default/get/setting")["data"]:
|
||||||
|
if s.get("key") == "guest_access":
|
||||||
|
return s
|
||||||
|
raise SystemExit("no guest_access setting found")
|
||||||
|
|
||||||
|
|
||||||
|
def items_of(p: dict) -> list[str]:
|
||||||
|
f = (p.get("destination") or {}).get("trafficFilter") or {}
|
||||||
|
return [i["value"] for i in (f.get("ipAddressFilter") or {}).get("items", [])]
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||||
|
ap.add_argument("site", choices=sorted(SITES))
|
||||||
|
ap.add_argument("--show", action="store_true", help="print the current restrictions and exit")
|
||||||
|
ap.add_argument("--remove", help="subnet to remove, e.g. 10.0.0.0/8")
|
||||||
|
ap.add_argument("--apply", action="store_true", help="actually write the change")
|
||||||
|
a = ap.parse_args()
|
||||||
|
|
||||||
|
c = Console(a.site)
|
||||||
|
sid = c.site_id()
|
||||||
|
zones = c.zones(sid)
|
||||||
|
pols = c.restriction_policies(sid, zones)
|
||||||
|
ga = c.guest_access()
|
||||||
|
legacy = {k: v for k, v in ga.items() if k.startswith("restricted_subnet_")}
|
||||||
|
|
||||||
|
print(f"{a.site} ({c.gateway}) guest Post-Authorization Restrictions:")
|
||||||
|
for p in pols:
|
||||||
|
print(f" gateway policy Hotspot -> {zones.get(p['destination']['zoneId']):8} index {p['index']}: {', '.join(items_of(p))}")
|
||||||
|
print(f" legacy guest_access (access points): {json.dumps(legacy)}")
|
||||||
|
if a.show or not a.remove:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
targets = [p for p in pols if a.remove in items_of(p)]
|
||||||
|
legacy_hits = {k: v for k, v in legacy.items() if v == a.remove}
|
||||||
|
if not targets and not legacy_hits:
|
||||||
|
print(f"\n{a.remove} is not in any restriction list; nothing to do")
|
||||||
|
return 0
|
||||||
|
print(f"\nwould remove {a.remove} from legacy field(s) {list(legacy_hits)}; the {len(targets)} gateway policy(ies) listing it regenerate from that setting")
|
||||||
|
if not a.apply:
|
||||||
|
print("DRY RUN, add --apply to write it")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
# The gateway policies are SYSTEM_DEFINED and the Integration API refuses to
|
||||||
|
# edit them (api.firewall.policy.operation.system-defined-firewall-policy-update-forbidden).
|
||||||
|
# They are generated from the legacy guest_access restricted_subnet_* fields, so
|
||||||
|
# that setting is the one lever; the policies and the APs follow it.
|
||||||
|
if not legacy_hits:
|
||||||
|
print(" no legacy restricted_subnet field holds it; the gateway policies cannot be edited directly")
|
||||||
|
return 1
|
||||||
|
# Re-pack the remaining subnets into slots 1..n and blank the rest.
|
||||||
|
remaining = [v for k, v in sorted(legacy.items()) if v and v != a.remove]
|
||||||
|
payload = {f"restricted_subnet_{i}": (remaining[i - 1] if i <= len(remaining) else "") for i in range(1, 4)}
|
||||||
|
print(f" POST /api/s/default/set/setting/guest_access {json.dumps(payload)}")
|
||||||
|
c.request("POST", "/api/s/default/set/setting/guest_access", payload)
|
||||||
|
ga2 = c.guest_access()
|
||||||
|
print(" legacy guest_access now:", json.dumps({k: v for k, v in ga2.items() if k.startswith("restricted_subnet_")}))
|
||||||
|
for p in c.restriction_policies(sid, zones):
|
||||||
|
print(f" gateway policy Hotspot -> {zones.get(p['destination']['zoneId']):8} now: {', '.join(items_of(p))}")
|
||||||
|
print("done; access points re-provision within a minute, guests should reconnect")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
+140
@@ -0,0 +1,140 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Change which access points broadcast a WiFi network (SSID) on one UniFi console.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" --list
|
||||||
|
docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" \
|
||||||
|
--add "Auditorium Left,Auditorium Right,Main Lobby Entrance" # dry run
|
||||||
|
docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" \
|
||||||
|
--add "Auditorium Left,Auditorium Right,Main Lobby Entrance" --apply # write it
|
||||||
|
docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" --remove "The Vault" --apply
|
||||||
|
docs/examples/waters-church-na/wifi-broadcast-devices.py north-attleboro "Waters Guest" --all --apply # every AP
|
||||||
|
|
||||||
|
Names are matched exactly against the console's device names (--list prints
|
||||||
|
them). Without --apply nothing is written: the script prints the APs the SSID
|
||||||
|
is on now and the list it would send. The whole broadcast object is PUT back
|
||||||
|
with only broadcastingDeviceFilter changed.
|
||||||
|
|
||||||
|
Needs the per-console Network API key at ~/.config/unifi/<site>-network-api-key
|
||||||
|
(see unifi-dns-record.py). The key is only ever sent in the X-API-KEY header.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import ssl
|
||||||
|
import sys
|
||||||
|
import urllib.error
|
||||||
|
import urllib.request
|
||||||
|
|
||||||
|
SITES = {
|
||||||
|
"apollo-beach": "10.3.0.1",
|
||||||
|
"north-attleboro": "10.1.0.1",
|
||||||
|
"woonsocket": "10.2.0.1",
|
||||||
|
"norwood": "10.4.0.1",
|
||||||
|
"southcoast": "10.5.0.1",
|
||||||
|
}
|
||||||
|
KEY_DIR = os.path.expanduser("~/.config/unifi")
|
||||||
|
TIMEOUT = 20
|
||||||
|
|
||||||
|
|
||||||
|
class Console:
|
||||||
|
def __init__(self, name: str):
|
||||||
|
self.name = name
|
||||||
|
self.gateway = SITES[name]
|
||||||
|
with open(os.path.join(KEY_DIR, f"{name}-network-api-key")) as f:
|
||||||
|
self.key = f.read().strip()
|
||||||
|
self.ctx = ssl.create_default_context()
|
||||||
|
self.ctx.check_hostname = False
|
||||||
|
self.ctx.verify_mode = ssl.CERT_NONE
|
||||||
|
|
||||||
|
def request(self, method: str, path: str, body: dict | None = None):
|
||||||
|
url = f"https://{self.gateway}/proxy/network/integration/v1{path}"
|
||||||
|
data = json.dumps(body).encode() if body is not None else None
|
||||||
|
req = urllib.request.Request(url, data=data, method=method, headers={
|
||||||
|
"X-API-KEY": self.key, "Accept": "application/json",
|
||||||
|
"Content-Type": "application/json"})
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req, timeout=TIMEOUT, context=self.ctx) as r:
|
||||||
|
raw = r.read().decode()
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
raise SystemExit(f"{method} {path} -> HTTP {e.code}: {e.read().decode(errors='replace')[:600]}")
|
||||||
|
return json.loads(raw) if raw else None
|
||||||
|
|
||||||
|
def site_id(self) -> str:
|
||||||
|
return self.request("GET", "/sites")["data"][0]["id"]
|
||||||
|
|
||||||
|
def access_points(self, sid: str) -> dict[str, str]:
|
||||||
|
devs = self.request("GET", f"/sites/{sid}/devices?limit=200")["data"]
|
||||||
|
return {d["name"]: d["id"] for d in devs if "accessPoint" in (d.get("features") or [])}
|
||||||
|
|
||||||
|
def broadcast(self, sid: str, ssid: str) -> dict:
|
||||||
|
for b in self.request("GET", f"/sites/{sid}/wifi/broadcasts?limit=200")["data"]:
|
||||||
|
if b.get("name") == ssid:
|
||||||
|
return b
|
||||||
|
raise SystemExit(f"no WiFi network named {ssid!r} on {self.name}")
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||||
|
ap.add_argument("site", choices=sorted(SITES))
|
||||||
|
ap.add_argument("ssid", help="WiFi network name as shown in the console")
|
||||||
|
ap.add_argument("--add", default="", help="comma-separated AP names to add")
|
||||||
|
ap.add_argument("--remove", default="", help="comma-separated AP names to remove")
|
||||||
|
ap.add_argument("--all", action="store_true", help="broadcast on every AP (clears the device filter)")
|
||||||
|
ap.add_argument("--list", action="store_true", help="print the APs and where the SSID broadcasts, then exit")
|
||||||
|
ap.add_argument("--apply", action="store_true", help="actually write the change")
|
||||||
|
a = ap.parse_args()
|
||||||
|
|
||||||
|
c = Console(a.site)
|
||||||
|
sid = c.site_id()
|
||||||
|
aps = c.access_points(sid)
|
||||||
|
by_id = {v: k for k, v in aps.items()}
|
||||||
|
b = c.broadcast(sid, a.ssid)
|
||||||
|
filt = b.get("broadcastingDeviceFilter") or {}
|
||||||
|
current = list(filt.get("deviceIds", [])) if filt.get("type") == "DEVICES" else None
|
||||||
|
|
||||||
|
def names(ids):
|
||||||
|
return ", ".join(by_id.get(i, i) for i in ids)
|
||||||
|
|
||||||
|
print(f"{a.site}: {a.ssid!r} broadcasts on " + ("ALL access points" if current is None else f"{len(current)} AP(s): {names(current)}"))
|
||||||
|
if a.list:
|
||||||
|
print("\nAccess points on this console:")
|
||||||
|
for n in sorted(aps):
|
||||||
|
print(f" {'*' if current is None or aps[n] in current else ' '} {n}")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
add = [s.strip() for s in a.add.split(",") if s.strip()]
|
||||||
|
rem = [s.strip() for s in a.remove.split(",") if s.strip()]
|
||||||
|
unknown = [n for n in add + rem if n not in aps]
|
||||||
|
if unknown:
|
||||||
|
raise SystemExit(f"unknown AP name(s): {', '.join(unknown)}; use --list")
|
||||||
|
if not (add or rem or a.all):
|
||||||
|
ap.error("nothing to change: give --add, --remove or --all")
|
||||||
|
|
||||||
|
new_b = {k: v for k, v in b.items() if k not in ("id", "metadata")}
|
||||||
|
if a.all:
|
||||||
|
new_b.pop("broadcastingDeviceFilter", None)
|
||||||
|
summary = "ALL access points"
|
||||||
|
else:
|
||||||
|
ids = list(current or [])
|
||||||
|
for n in add:
|
||||||
|
if aps[n] not in ids:
|
||||||
|
ids.append(aps[n])
|
||||||
|
ids = [i for i in ids if i not in {aps[n] for n in rem}]
|
||||||
|
new_b["broadcastingDeviceFilter"] = {"type": "DEVICES", "deviceIds": ids}
|
||||||
|
summary = f"{len(ids)} AP(s): {names(ids)}"
|
||||||
|
print(f"would broadcast on {summary}")
|
||||||
|
if not a.apply:
|
||||||
|
print("DRY RUN, add --apply to write it")
|
||||||
|
return 0
|
||||||
|
out = c.request("PUT", f"/sites/{sid}/wifi/broadcasts/{b['id']}", new_b)
|
||||||
|
f2 = (out or {}).get("broadcastingDeviceFilter") or {}
|
||||||
|
print("UPDATED: now on " + ("ALL access points" if f2.get("type") != "DEVICES" else names(f2.get("deviceIds", []))))
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
+160
@@ -0,0 +1,160 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Create a zone-based firewall policy on one UniFi console that allows TCP from a
|
||||||
|
source zone to one destination IP on given ports (e.g. guest WiFi -> a LAN web host).
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
docs/examples/waters-church-na/zone-allow-policy.py north-attleboro Hotspot 10.1.1.4 80,443 \
|
||||||
|
--name "Guest to translate.watersmedia.org (Boba-Fett web)" # dry run
|
||||||
|
docs/examples/waters-church-na/zone-allow-policy.py north-attleboro Hotspot 10.1.1.4 80,443 \
|
||||||
|
--name "Guest to translate.watersmedia.org (Boba-Fett web)" --apply # create it
|
||||||
|
docs/examples/waters-church-na/zone-allow-policy.py north-attleboro --list-zones
|
||||||
|
|
||||||
|
Without --apply nothing is written: the script prints the zones it resolved, any
|
||||||
|
existing policy with the same name, and the exact JSON it would POST. The
|
||||||
|
destination zone is found from the console's networks (the one whose subnet
|
||||||
|
contains the IP). User-defined policies are evaluated before the system ones
|
||||||
|
(index >= 30000), so the allow lands above "Post-Authorization Restrictions".
|
||||||
|
|
||||||
|
Needs the per-console Network API key at ~/.config/unifi/<site>-network-api-key
|
||||||
|
(see unifi-dns-record.py). The key is only ever sent in the X-API-KEY header.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import ipaddress
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import ssl
|
||||||
|
import sys
|
||||||
|
import urllib.error
|
||||||
|
import urllib.request
|
||||||
|
|
||||||
|
SITES = {
|
||||||
|
"apollo-beach": "10.3.0.1",
|
||||||
|
"north-attleboro": "10.1.0.1",
|
||||||
|
"woonsocket": "10.2.0.1",
|
||||||
|
"norwood": "10.4.0.1",
|
||||||
|
"southcoast": "10.5.0.1",
|
||||||
|
}
|
||||||
|
KEY_DIR = os.path.expanduser("~/.config/unifi")
|
||||||
|
TIMEOUT = 20
|
||||||
|
|
||||||
|
|
||||||
|
class Console:
|
||||||
|
def __init__(self, name: str):
|
||||||
|
self.name = name
|
||||||
|
self.gateway = SITES[name]
|
||||||
|
with open(os.path.join(KEY_DIR, f"{name}-network-api-key")) as f:
|
||||||
|
self.key = f.read().strip()
|
||||||
|
self.ctx = ssl.create_default_context()
|
||||||
|
self.ctx.check_hostname = False
|
||||||
|
self.ctx.verify_mode = ssl.CERT_NONE
|
||||||
|
|
||||||
|
def request(self, method: str, path: str, body: dict | None = None):
|
||||||
|
url = f"https://{self.gateway}/proxy/network{path}"
|
||||||
|
data = json.dumps(body).encode() if body is not None else None
|
||||||
|
req = urllib.request.Request(url, data=data, method=method, headers={
|
||||||
|
"X-API-KEY": self.key, "Accept": "application/json",
|
||||||
|
"Content-Type": "application/json"})
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen(req, timeout=TIMEOUT, context=self.ctx) as r:
|
||||||
|
raw = r.read().decode()
|
||||||
|
except urllib.error.HTTPError as e:
|
||||||
|
raise SystemExit(f"{method} {path} -> HTTP {e.code}: {e.read().decode(errors='replace')[:600]}")
|
||||||
|
return json.loads(raw) if raw else None
|
||||||
|
|
||||||
|
def site_id(self) -> str:
|
||||||
|
return self.request("GET", "/integration/v1/sites")["data"][0]["id"]
|
||||||
|
|
||||||
|
def zones(self, sid: str) -> dict[str, str]:
|
||||||
|
data = self.request("GET", f"/integration/v1/sites/{sid}/firewall/zones?limit=200")["data"]
|
||||||
|
return {z["name"]: z["id"] for z in data}
|
||||||
|
|
||||||
|
def zone_for_ip(self, sid: str, ip: str) -> tuple[str, str]:
|
||||||
|
"""Zone id and network name whose subnet contains ip (via the legacy networkconf)."""
|
||||||
|
nets = {n["id"]: n for n in self.request("GET", f"/integration/v1/sites/{sid}/networks?limit=200")["data"]}
|
||||||
|
addr = ipaddress.ip_address(ip)
|
||||||
|
for n in self.request("GET", "/api/s/default/rest/networkconf")["data"]:
|
||||||
|
subnet = n.get("ip_subnet")
|
||||||
|
if not subnet:
|
||||||
|
continue
|
||||||
|
if addr in ipaddress.ip_interface(subnet).network:
|
||||||
|
zone_id = nets.get(n.get("external_id", ""), {}).get("zoneId")
|
||||||
|
if zone_id:
|
||||||
|
return zone_id, n["name"]
|
||||||
|
raise SystemExit(f"no gateway network on {self.name} contains {ip}")
|
||||||
|
|
||||||
|
def policies(self, sid: str) -> list[dict]:
|
||||||
|
return self.request("GET", f"/integration/v1/sites/{sid}/firewall/policies?limit=500")["data"]
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||||
|
ap.add_argument("site", choices=sorted(SITES))
|
||||||
|
ap.add_argument("source_zone", nargs="?", help="source zone name, e.g. Hotspot")
|
||||||
|
ap.add_argument("dest_ip", nargs="?", help="destination IPv4 address")
|
||||||
|
ap.add_argument("ports", nargs="?", help="comma-separated TCP ports, e.g. 80,443")
|
||||||
|
ap.add_argument("--name", help="policy name (default is generated)")
|
||||||
|
ap.add_argument("--apply", action="store_true", help="actually create the policy")
|
||||||
|
ap.add_argument("--list-zones", action="store_true", help="print the console's zones and exit")
|
||||||
|
a = ap.parse_args()
|
||||||
|
|
||||||
|
c = Console(a.site)
|
||||||
|
sid = c.site_id()
|
||||||
|
zones = c.zones(sid)
|
||||||
|
if a.list_zones:
|
||||||
|
for name, zid in sorted(zones.items()):
|
||||||
|
print(f"{name:12} {zid}")
|
||||||
|
return 0
|
||||||
|
if not (a.source_zone and a.dest_ip and a.ports):
|
||||||
|
ap.error("source_zone, dest_ip and ports are required unless --list-zones")
|
||||||
|
if a.source_zone not in zones:
|
||||||
|
raise SystemExit(f"zone {a.source_zone!r} not on {a.site}; have: {', '.join(sorted(zones))}")
|
||||||
|
ports = [int(p) for p in a.ports.split(",")]
|
||||||
|
dest_zone, dest_net = c.zone_for_ip(sid, a.dest_ip)
|
||||||
|
dest_zone_name = next(n for n, z in zones.items() if z == dest_zone)
|
||||||
|
name = a.name or f"{a.source_zone} to {a.dest_ip} tcp {a.ports}"
|
||||||
|
|
||||||
|
body = {
|
||||||
|
"name": name,
|
||||||
|
"enabled": True,
|
||||||
|
"action": {"type": "ALLOW", "allowReturnTraffic": True},
|
||||||
|
"source": {"zoneId": zones[a.source_zone]},
|
||||||
|
"destination": {
|
||||||
|
"zoneId": dest_zone,
|
||||||
|
"trafficFilter": {
|
||||||
|
"type": "IP_ADDRESS",
|
||||||
|
"ipAddressFilter": {"type": "IP_ADDRESSES", "matchOpposite": False,
|
||||||
|
"items": [{"type": "IP_ADDRESS", "value": a.dest_ip}]},
|
||||||
|
"portFilter": {"type": "PORTS", "matchOpposite": False,
|
||||||
|
"items": [{"type": "PORT_NUMBER", "value": p} for p in ports]},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"ipProtocolScope": {"ipVersion": "IPV4",
|
||||||
|
"protocolFilter": {"type": "NAMED_PROTOCOL", "protocol": {"name": "TCP"},
|
||||||
|
"matchOpposite": False}},
|
||||||
|
"loggingEnabled": False,
|
||||||
|
}
|
||||||
|
|
||||||
|
print(f"{a.site} ({c.gateway}): {a.source_zone} -> {dest_zone_name} ({dest_net}) {a.dest_ip} tcp {a.ports}")
|
||||||
|
existing = [p for p in c.policies(sid) if p.get("name") == name]
|
||||||
|
if existing:
|
||||||
|
print(f"a policy named {name!r} already exists (id {existing[0]['id']}, index {existing[0].get('index')}); nothing to do")
|
||||||
|
return 0
|
||||||
|
if not a.apply:
|
||||||
|
print("DRY RUN, would POST /integration/v1/sites/<site>/firewall/policies:")
|
||||||
|
print(json.dumps(body, indent=2))
|
||||||
|
return 0
|
||||||
|
created = c.request("POST", f"/integration/v1/sites/{sid}/firewall/policies", body)
|
||||||
|
print(f"CREATED id {created.get('id')} index {created.get('index')} enabled {created.get('enabled')}")
|
||||||
|
blockers = [p for p in c.policies(sid) if p.get("name") == "Post-Authorization Restrictions"
|
||||||
|
and p["source"]["zoneId"] == zones[a.source_zone] and p["destination"]["zoneId"] == dest_zone]
|
||||||
|
for b in blockers:
|
||||||
|
order = "above" if created.get("index", 0) < b.get("index", 0) else "BELOW"
|
||||||
|
print(f"new policy sits {order} 'Post-Authorization Restrictions' (index {b.get('index')})")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
Reference in New Issue
Block a user