Development

routlin/core.py

@@ -781,7 +781,7 @@ def cmd_apply(data, dry_run=False):
         timers.install_maint_timer(data)
     else:
         timers.remove_timers([timers.MAINT_TIMER_NAME], [timers.MAINT_TIMER_FILE], [timers.MAINT_TIMER_SVC_FILE])
-        print("No enabled DDNS providers — timer not installed.")
+        print("No enabled DDNS providers - timer not installed.")
     print()
 
     print("Boot service ========================================================")

routlin/mod_nftables.py

@@ -404,19 +404,41 @@ def build_nft_config(data, dry_run=False):
         "",
     ]
 
+    L += ["        # Anti-spoofing: drop packets arriving on a VLAN interface with a source IP outside that VLAN's subnet", ""]
+    for vlan in vlans:
+        if validation.is_wg(vlan):
+            continue
+        iface  = validation.derive_interface(vlan, data)
+        subnet = vlan.get('subnet', '')
+        mask   = vlan.get('subnet_mask', 24)
+        if subnet:
+            L.append(f"        iif \"{iface}\" ip saddr != {subnet}/{mask} drop  # {vlan['name']} anti-spoof")
+    L.append("")
+
     L.append("        # Allow each VLAN -> WAN (outbound internet)")
     for vlan in vlans:
+        if vlan.get('restricted_vlan'):
+            continue
         L.append(f"        iif \"{validation.derive_interface(vlan, data)}\" oif \"{wan}\" accept  # {vlan['name']} -> WAN")
     L.append("")
 
     if container_bridges:
         L.append("        # Allow VLAN -> Docker bridge forwarding")
         for vlan in vlans:
+            if vlan.get('restricted_vlan'):
+                continue
             for bridge in container_bridges:
                 L.append(f"        iif \"{validation.derive_interface(vlan, data)}\" oif \"{bridge}\" ct state new accept"
                          f"  # {vlan['name']} -> {bridge}")
         L.append("")
 
+    restricted = [v for v in vlans if v.get('restricted_vlan')]
+    if restricted:
+        L.append("        # Block restricted VLANs -> WAN")
+        for vlan in restricted:
+            L.append(f"        iif \"{validation.derive_interface(vlan, data)}\" oif \"{wan}\" drop  # {vlan['name']} -> WAN (restricted)")
+        L.append("")
+
     L += [
         "        # Allow Docker containers -> WAN (outbound internet access)",
         f"        iif != \"{wan}\" oif \"{wan}\" ct state new accept",

routlin/mod_validation.py

@@ -845,6 +845,26 @@ def validate_config(data):
         nat_check_port(f"{label} dest_port", r.get("dest_port"))
         nat_check_port(f"{label} nat_port",  r.get("nat_port"))
         nat_check_ip(f"{label} nat_ip", r.get("nat_ip", ""))
+        if r.get("enabled"):
+            nat_ip_str = r.get("nat_ip", "")
+            try:
+                nat_addr = ipaddress.IPv4Address(nat_ip_str)
+                for v in _all_vlans:
+                    if not v.get("restricted_vlan"):
+                        continue
+                    try:
+                        vnet = ipaddress.IPv4Network(f"{v['subnet']}/{v['subnet_mask']}", strict=False)
+                    except Exception:
+                        continue
+                    if nat_addr in vnet:
+                        errors.append(
+                            f"Port forwarding rule '{desc}' is enabled but its destination "
+                            f"({nat_ip_str}) is on restricted VLAN '{v['name']}'. "
+                            f"Disable the rule or remove the restricted_vlan flag."
+                        )
+                        break
+            except Exception:
+                pass
 
     for r in data.get("inter_vlan_exceptions", []):
         desc  = r.get("description", "?")
@@ -930,3 +950,32 @@ def validate_config(data):
             errors.append(f"{lbl}: '{ip_val}' is not a valid IP, CIDR, or wildcard pattern.")
 
     return errors
+
+
+def disable_portfwd_on_restricted_vlans(data):
+    """Auto-disable enabled port forwarding rules whose nat_ip falls within a restricted VLAN's subnet.
+    Mutates data in place. Returns list of descriptions of rules that were disabled."""
+    restricted_nets = []
+    for v in data.get('vlans', []):
+        if v.get('restricted_vlan'):
+            try:
+                restricted_nets.append(ipaddress.IPv4Network(f"{v['subnet']}/{v['subnet_mask']}", strict=False))
+            except Exception:
+                pass
+
+    if not restricted_nets:
+        return []
+
+    disabled = []
+    for rule in data.get('port_forwarding', []):
+        if not rule.get('enabled'):
+            continue
+        try:
+            addr = ipaddress.IPv4Address(rule.get('nat_ip', ''))
+        except Exception:
+            continue
+        if any(addr in net for net in restricted_nets):
+            rule['enabled'] = False
+            disabled.append(rule.get('description') or rule.get('nat_ip', '?'))
+
+    return disabled