Tyler King
6058e62348
Go-based network automation with YANG models, gRPC, Ansible, Terraform, and Kubernetes integration. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""
|
|
UniFi Controller API JSON generator.
|
|
|
|
Transforms YANG zone-policy data into UniFi Controller API payloads
|
|
for network and firewall rule configuration.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
|
|
def compile_zones(zones: list[dict[str, Any]]) -> dict[str, Any]:
|
|
"""
|
|
Compile zone definitions into UniFi Controller API payloads.
|
|
|
|
Returns a dict with 'networks' and 'firewall_rules' keys,
|
|
each containing UniFi API request bodies.
|
|
"""
|
|
result: dict[str, Any] = {
|
|
"target": "unifi",
|
|
"api_version": "v1",
|
|
"networks": [],
|
|
"firewall_rules": [],
|
|
}
|
|
|
|
for zone in zones:
|
|
import ipaddress
|
|
|
|
network = ipaddress.IPv4Network(zone["subnet"], strict=False)
|
|
|
|
# UniFi network (VLAN) configuration.
|
|
result["networks"].append({
|
|
"endpoint": "/api/s/default/rest/networkconf",
|
|
"method": "POST",
|
|
"body": {
|
|
"name": zone["name"],
|
|
"purpose": "corporate",
|
|
"vlan_enabled": True,
|
|
"vlan": zone["vlan_id"],
|
|
"ip_subnet": zone["subnet"],
|
|
"networkgroup": "LAN",
|
|
"dhcpd_enabled": False,
|
|
"domain_name": f"{zone['name']}.kedge.local",
|
|
},
|
|
})
|
|
|
|
# UniFi firewall rules for inter-zone traffic.
|
|
for i, policy in enumerate(zone.get("policies", [])):
|
|
rule: dict[str, Any] = {
|
|
"endpoint": "/api/s/default/rest/firewallrule",
|
|
"method": "POST",
|
|
"body": {
|
|
"name": f"{zone['name']}-to-{policy['dst_zone']}",
|
|
"enabled": True,
|
|
"ruleset": "LAN_IN",
|
|
"rule_index": 2000 + (zone["vlan_id"] * 10) + i,
|
|
"action": _map_action(policy["action"]),
|
|
"protocol": "all",
|
|
"src_networkconf_type": "NETv4",
|
|
"src_address": zone["subnet"],
|
|
},
|
|
}
|
|
|
|
if policy.get("services"):
|
|
# Map service names to port numbers for UniFi.
|
|
ports = [_service_to_port(svc) for svc in policy["services"]]
|
|
ports = [p for p in ports if p]
|
|
if ports:
|
|
rule["body"]["dst_port"] = ",".join(ports)
|
|
rule["body"]["protocol"] = "tcp_udp"
|
|
|
|
result["firewall_rules"].append(rule)
|
|
|
|
return result
|
|
|
|
|
|
def _map_action(yang_action: str) -> str:
|
|
"""Map YANG action enum to UniFi firewall action."""
|
|
mapping = {
|
|
"allow-stateful": "accept",
|
|
"allow-restricted": "accept",
|
|
"deny": "drop",
|
|
}
|
|
return mapping.get(yang_action, "drop")
|
|
|
|
|
|
def _service_to_port(service: str) -> str | None:
|
|
"""Map common service names to port numbers."""
|
|
port_map = {
|
|
"ssh": "22",
|
|
"http": "80",
|
|
"https": "443",
|
|
"dns": "53",
|
|
"ntp": "123",
|
|
"snmp": "161",
|
|
}
|
|
return port_map.get(service.lower())
|