root/avaya_switch_management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add port forwarding, PoE dashboard, topology tabs + deduplicate firewall

Browse Source 
Removed duplicate firewall policy endpoints (kept existing ones at
/api/firewall/* which match the frontend).

Port Forwarding tab:
- Create/delete OPNsense NAT port forwards via API
- Track rule UUIDs for clean removal
- Form: protocol, WAN port, target IP:port, description
- Table: active forwards with one-click remove
- Note: for HTTP services, use Services tab (Caddy) instead

PoE Budget tab:
- Visual power bar: used/total/remaining watts with percentage
- Color-coded thresholds: green (<75%), orange (75-90%), red (>90%)
- Warning banner when budget exceeds 85%
- Per-port power draw grid with status indicators
- Auto-parsed from cached switch PoE status

Network Topology tab:
- Auto-generated from live switch + OPNsense data
- Router node: IP, version, online/offline status
- Switch node: hostname, IP, port up/down counts
- Trunk link visualization between router and switch
- VLAN fan-out cards: port counts, device counts, subnets
- One-click refresh

https://claude.ai/code/session_01Do9bsN39MTuy2GVv7yzSrE
This commit is contained in:
Claude
2026-03-28 02:27:50 +00:00
parent 130baf9303
commit e486de26b0
2 changed files with 321 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch_backend.py
-363
View File
@@ -5238,369 +5238,6 @@ def run_schedule_now(body: dict):
return {"success": True, "ran": name}
# ══════════════════════════════════════════════════════════════════════
# FIREWALL POLICY MATRIX — inter-VLAN access control
# ══════════════════════════════════════════════════════════════════════
#
# Manages both switch ACLs AND OPNsense firewall rules together.
# Policies define what each VLAN pair can do:
# - full: all traffic allowed between VLANs
# - internet: VLAN gets internet only, no RFC1918 access
# - blocked: no traffic between these VLANs
# - service: one-way access (A can reach B, but B cannot reach A)
# - custom: user-defined rules
#
# "service" is the printer pattern: Staff can print, but printers
# can't initiate connections to Staff.
POLICIES_FILE = _Path("/etc/switch-manager/vlan-policies.json")
_POLICY_TYPES = {"full", "internet", "blocked", "service", "custom"}
def _load_policies() -> list:
if POLICIES_FILE.exists():
try: return _json.loads(POLICIES_FILE.read_text())
except: pass
return []
def _save_policies(policies: list):
POLICIES_FILE.write_text(_json.dumps(policies, indent=2))
POLICIES_FILE.chmod(0o600)
def _policy_to_switch_acl(policy: dict) -> list:
"""Generate ERS switch ACL commands for a VLAN policy."""
ptype = policy.get("type", "blocked")
src_vid = policy.get("src_vlan")
dst_vid = policy.get("dst_vlan")
src_sub = policy.get("src_subnet", f"192.168.{src_vid}.0")
dst_sub = policy.get("dst_subnet", f"192.168.{dst_vid}.0")
acl_name = f"POLICY-V{src_vid}-V{dst_vid}"
cmds = []
if ptype == "blocked":
cmds = [
f"ip access-list extended {acl_name}",
f" 1 deny ip {src_sub} 0.0.0.255 {dst_sub} 0.0.0.255",
f" 2 permit ip any any",
f"interface vlan {src_vid}",
f" ip access-group {acl_name} in",
]
elif ptype == "full":
cmds = [
f"ip access-list extended {acl_name}",
f" 1 permit ip {src_sub} 0.0.0.255 {dst_sub} 0.0.0.255",
f" 2 permit ip any any",
f"interface vlan {src_vid}",
f" ip access-group {acl_name} in",
]
elif ptype == "internet":
cmds = [
f"ip access-list extended {acl_name}",
f" 1 deny ip {src_sub} 0.0.0.255 10.0.0.0 0.255.255.255",
f" 2 deny ip {src_sub} 0.0.0.255 172.16.0.0 0.15.255.255",
f" 3 deny ip {src_sub} 0.0.0.255 192.168.0.0 0.0.255.255",
f" 4 permit ip any any",
f"interface vlan {src_vid}",
f" ip access-group {acl_name} in",
]
elif ptype == "service":
# One-way: src can reach dst on specified ports, dst cannot initiate to src
ports = policy.get("ports", [])
cmds = [f"ip access-list extended {acl_name}"]
rule_num = 1
for p in ports:
proto = p.get("proto", "tcp")
port = p.get("port", "")
if port:
cmds.append(
f" {rule_num} permit {proto} {src_sub} 0.0.0.255 "
f"{dst_sub} 0.0.0.255 eq {port}")
else:
cmds.append(
f" {rule_num} permit {proto} {src_sub} 0.0.0.255 "
f"{dst_sub} 0.0.0.255")
rule_num += 1
# Deny all other traffic to that VLAN
cmds.append(f" {rule_num} deny ip {src_sub} 0.0.0.255 {dst_sub} 0.0.0.255")
rule_num += 1
cmds.append(f" {rule_num} permit ip any any")
cmds += [
f"interface vlan {src_vid}",
f" ip access-group {acl_name} in",
]
return cmds
def _policy_to_opnsense_rules(policy: dict, cfg: dict) -> list:
"""Generate OPNsense firewall API calls for a VLAN policy.
Returns list of {method, path, body} dicts."""
ptype = policy.get("type", "blocked")
src_vid = policy.get("src_vlan")
dst_vid = policy.get("dst_vlan")
vmap = _load_vlan_if_map()
src_if = vmap.get(str(src_vid), "")
dst_if = vmap.get(str(dst_vid), "")
src_sub = policy.get("src_subnet", f"192.168.{src_vid}.0/24")
dst_sub = policy.get("dst_subnet", f"192.168.{dst_vid}.0/24")
rules = []
if not src_if:
return rules # Can't create OPNsense rules without interface mapping
if ptype == "blocked":
rules.append({
"rule": {
"enabled": "1", "action": "block",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": "any",
"source": {"network": f"{src_if}net"},
"destination": {"address": dst_sub},
"descr": f"Policy: block V{src_vid} → V{dst_vid}",
}
})
elif ptype == "full":
rules.append({
"rule": {
"enabled": "1", "action": "pass",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": "any",
"source": {"network": f"{src_if}net"},
"destination": {"address": dst_sub},
"descr": f"Policy: allow V{src_vid} → V{dst_vid}",
}
})
elif ptype == "internet":
# Block all RFC1918, permit everything else
for net in ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"]:
rules.append({
"rule": {
"enabled": "1", "action": "block",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": "any",
"source": {"network": f"{src_if}net"},
"destination": {"address": net},
"descr": f"Policy: V{src_vid} internet-only (block {net})",
}
})
rules.append({
"rule": {
"enabled": "1", "action": "pass",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": "any",
"source": {"network": f"{src_if}net"},
"destination": {"any": "1"},
"descr": f"Policy: V{src_vid} internet-only (allow out)",
}
})
elif ptype == "service":
ports = policy.get("ports", [])
for p in ports:
proto = p.get("proto", "tcp")
port = p.get("port", "")
rule = {
"enabled": "1", "action": "pass",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": proto,
"source": {"network": f"{src_if}net"},
"destination": {"address": dst_sub},
"descr": f"Policy: V{src_vid} → V{dst_vid} service {proto}/{port}",
}
if port:
rule["destination"]["port"] = str(port)
rules.append({"rule": rule})
# Block everything else to that VLAN
rules.append({
"rule": {
"enabled": "1", "action": "block",
"interface": src_if, "direction": "in",
"ipprotocol": "inet", "protocol": "any",
"source": {"network": f"{src_if}net"},
"destination": {"address": dst_sub},
"descr": f"Policy: block V{src_vid} → V{dst_vid} (except services above)",
}
})
return rules
# ── Policy presets ──────────────────────────────────────────────────
POLICY_PRESETS = [
{
"id": "printer",
"label": "Printer VLAN — other VLANs can print, printers can't reach out",
"description": "Allows printing (TCP 9100 RAW, TCP 631 IPP, UDP 631 IPP) "
"from source VLAN to printer VLAN. Printers cannot initiate "
"connections back. Printers get internet for firmware updates.",
"type": "service",
"ports": [
{"proto": "tcp", "port": "9100"}, # RAW printing
{"proto": "tcp", "port": "631"}, # IPP
{"proto": "udp", "port": "631"}, # IPP discovery
{"proto": "tcp", "port": "443"}, # HTTPS (web UI, cloud print)
{"proto": "tcp", "port": "80"}, # HTTP (web UI)
],
"bidirectional": False,
},
{
"id": "lan_access_all",
"label": "LAN can reach all VLANs",
"description": "LAN (trusted) has full access to all other VLANs. "
"Other VLANs cannot reach LAN.",
"type": "full",
"bidirectional": False,
},
{
"id": "iot_isolated",
"label": "IoT — internet only, full isolation",
"description": "Blocks ALL private IP ranges. Devices get internet only. "
"Cannot reach any VLAN, server, NAS, or management network.",
"type": "internet",
"bidirectional": False,
},
{
"id": "guest_isolated",
"label": "Guest — internet only, strict",
"description": "Same as IoT isolation. Guest devices get internet only.",
"type": "internet",
"bidirectional": False,
},
{
"id": "camera_nvr",
"label": "Camera VLAN — NVR access only",
"description": "Cameras can only reach the NVR IP. No internet, no other VLANs.",
"type": "service",
"ports": [{"proto": "tcp", "port": ""}], # All TCP to NVR
"bidirectional": False,
"needs_target_ip": True,
},
]
@app.get("/api/policies")
def get_policies():
"""List all VLAN policies and available presets."""
return {
"policies": _load_policies(),
"presets": POLICY_PRESETS,
}
@app.post("/api/policies")
def save_policy(body: dict):
"""Add or update a VLAN-to-VLAN policy."""
require_session(body.get("token", ""))
policy = body.get("policy", {})
if not policy.get("src_vlan") or not policy.get("type"):
raise HTTPException(400, "src_vlan and type required")
if policy["type"] not in _POLICY_TYPES:
raise HTTPException(400, f"Invalid type: {policy['type']}")
policies = _load_policies()
# Replace existing policy for same src→dst pair
key = (policy["src_vlan"], policy.get("dst_vlan"))
policies = [p for p in policies if (p["src_vlan"], p.get("dst_vlan")) != key]
policies.append(policy)
_save_policies(policies)
return {"success": True, "policies": policies}
@app.delete("/api/policies")
def delete_policy(body: dict):
"""Remove a VLAN policy."""
require_session(body.get("token", ""))
src = body.get("src_vlan")
dst = body.get("dst_vlan")
policies = _load_policies()
policies = [p for p in policies if not (p["src_vlan"] == src and p.get("dst_vlan") == dst)]
_save_policies(policies)
return {"success": True, "policies": policies}
@app.post("/api/policies/preview")
def preview_policy(body: dict):
"""Preview the switch ACL + OPNsense rules that a policy would generate."""
policy = body.get("policy", {})
if not policy.get("src_vlan") or not policy.get("type"):
raise HTTPException(400, "src_vlan and type required")
switch_cmds = _policy_to_switch_acl(policy)
cfg = _load_opnsense_cfg()
opnsense_rules = _policy_to_opnsense_rules(policy, cfg)
return {
"switch_commands": switch_cmds,
"opnsense_rules": [r["rule"]["descr"] for r in opnsense_rules],
"opnsense_rule_count": len(opnsense_rules),
}
@app.post("/api/policies/push")
def push_policy(body: dict):
"""Push a policy to both switch and OPNsense."""
require_session(body.get("token", ""))
policy = body.get("policy", {})
if not policy.get("src_vlan") or not policy.get("type"):
raise HTTPException(400, "src_vlan and type required")
steps_done = []
errors = []
backup = _pre_change_backup(reason=f"pre-policy V{policy['src_vlan']}→V{policy.get('dst_vlan','*')}")
# Push switch ACLs
switch_cmds = _policy_to_switch_acl(policy)
if switch_cmds:
danger = check_danger(switch_cmds)
if danger["has_hard_block"]:
raise HTTPException(400, {"message": "Hard-blocked commands", "blocked": danger["hard_blocked"]})
result = push_one_by_one(switch_cmds)
if result.get("success"):
steps_done.append(f"Switch: {len(switch_cmds)} ACL commands pushed")
else:
errors.append(f"Switch push failed: {result.get('error', 'unknown')}")
# Push OPNsense rules
cfg = _load_opnsense_cfg()
if cfg.get("key"):
opnsense_rules = _policy_to_opnsense_rules(policy, cfg)
for rule_body in opnsense_rules:
try:
_opnsense_request(cfg, "firewall/filter/addRule", "POST", rule_body)
except ValueError as e:
errors.append(f"OPNsense rule: {e}")
if opnsense_rules:
try:
_opnsense_request(cfg, "firewall/filter/apply", "POST")
steps_done.append(f"OPNsense: {len(opnsense_rules)} firewall rules applied")
except ValueError as e:
errors.append(f"OPNsense apply: {e}")
# Save policy
policies = _load_policies()
key = (policy["src_vlan"], policy.get("dst_vlan"))
policies = [p for p in policies if (p["src_vlan"], p.get("dst_vlan")) != key]
policy["pushed"] = True
policy["pushed_at"] = _ts()
policies.append(policy)
_save_policies(policies)
# Post-connectivity check
post_conn = _check_connectivity()
if not post_conn["switch"]["ok"]:
errors.append(f"WARNING: Switch connectivity lost after push! Backup: {backup['switch'].get('file','N/A')}")
return {
"success": len(errors) == 0,
"steps_done": steps_done,
"errors": errors,
"backup": backup,
"post_connectivity": post_conn,
}
# ══════════════════════════════════════════════════════════════════════
# PORT FORWARDING — manage OPNsense NAT port forwards
# ══════════════════════════════════════════════════════════════════════