""" Network Policy Engine Enforces security, naming, and design best practices """ from typing import List, Dict, Any, Tuple from ipaddress import ip_network, ip_address import re class PolicyViolation: """Represents a policy violation""" def __init__(self, severity: str, category: str, message: str, location: str = ""): self.severity = severity # ERROR, WARNING, INFO self.category = category # security, naming, design, addressing self.message = message self.location = location def __str__(self): loc = f" [{self.location}]" if self.location else "" return f"{self.severity} ({self.category}){loc}: {self.message}" class NetworkPolicy: """ Enforces network design policies Checks against industry best practices and organizational standards """ def __init__(self, config: Dict[str, Any] = None): """ Initialize policy engine with configuration Default policies: - RFC1918 private addressing required - VLAN 1 not allowed for production - Management VLAN required - Secure naming conventions - Gateway is first usable IP in subnet """ self.config = config or {} self.violations: List[PolicyViolation] = [] def check_network_model(self, model: Dict[str, Any]) -> List[PolicyViolation]: """ Run all policy checks on network model Returns list of violations """ self.violations = [] # Naming policies self._check_naming_conventions(model) # Addressing policies self._check_addressing_policies(model) # VLAN policies self._check_vlan_policies(model) # Security policies self._check_security_policies(model) # Design best practices self._check_design_practices(model) return self.violations def _check_naming_conventions(self, model: Dict[str, Any]): """Check naming follows standards""" # Device naming for device in model.get('devices', []): name = device.get('name', '') role = device.get('role', '') # Should include role in name if role and role not in name.lower(): self.violations.append(PolicyViolation( severity="WARNING", category="naming", message=f"Device name should indicate role '{role}'", location=f"device:{name}" )) # Should use hyphens not underscores if '_' in name: self.violations.append(PolicyViolation( severity="INFO", category="naming", message="Consider using hyphens instead of underscores in device names", location=f"device:{name}" )) # Check for sequential numbering if not re.search(r'\d{2}$', name): self.violations.append(PolicyViolation( severity="INFO", category="naming", message="Device name should end with 2-digit number for scalability", location=f"device:{name}" )) # VLAN naming for vlan in model.get('vlans', []): name = vlan.get('name', '') # No spaces in VLAN names if ' ' in name: self.violations.append(PolicyViolation( severity="WARNING", category="naming", message="VLAN names should not contain spaces", location=f"vlan:{vlan.get('id')}" )) # Should be descriptive if len(name) < 3: self.violations.append(PolicyViolation( severity="INFO", category="naming", message="VLAN name should be descriptive (3+ characters)", location=f"vlan:{vlan.get('id')}" )) def _check_addressing_policies(self, model: Dict[str, Any]): """Check IP addressing follows best practices""" # Check for RFC1918 private addressing require_private = self.config.get('require_rfc1918', True) for subnet in model.get('subnets', []): network = subnet.get('network', '') gateway = subnet.get('gateway', '') try: net = ip_network(network, strict=False) # Check if using private addressing if require_private and not net.is_private: self.violations.append(PolicyViolation( severity="ERROR", category="addressing", message=f"Non-private address space detected: {network} (use RFC1918: 10/8, 172.16/12, 192.168/16)", location=f"subnet:{network}" )) # Check gateway is first usable IP if gateway: gw = ip_address(gateway) first_usable = list(net.hosts())[0] if net.num_addresses > 2 else None if first_usable and gw != first_usable: self.violations.append(PolicyViolation( severity="INFO", category="addressing", message=f"Gateway {gateway} is not first usable IP ({first_usable})", location=f"subnet:{network}" )) # Warn on wasteful subnets (e.g., /24 for 2 devices) if net.num_addresses > 256: self.violations.append(PolicyViolation( severity="INFO", category="addressing", message=f"Large subnet ({net.num_addresses} IPs) - consider smaller subnets for better security segmentation", location=f"subnet:{network}" )) except Exception as e: self.violations.append(PolicyViolation( severity="ERROR", category="addressing", message=f"Invalid subnet: {e}", location=f"subnet:{network}" )) # Check for overlapping subnets subnets_list = [s.get('network') for s in model.get('subnets', [])] for i, subnet1 in enumerate(subnets_list): for subnet2 in subnets_list[i+1:]: try: net1 = ip_network(subnet1, strict=False) net2 = ip_network(subnet2, strict=False) if net1.overlaps(net2): self.violations.append(PolicyViolation( severity="ERROR", category="addressing", message=f"Overlapping subnets: {subnet1} and {subnet2}", location="subnets" )) except Exception: pass def _check_vlan_policies(self, model: Dict[str, Any]): """Check VLAN configuration policies""" vlan_ids = [v.get('id') for v in model.get('vlans', [])] vlan_names = {v.get('id'): v.get('name', '') for v in model.get('vlans', [])} # VLAN 1 should not be used if 1 in vlan_ids: self.violations.append(PolicyViolation( severity="WARNING", category="security", message="VLAN 1 detected - default VLAN should not be used for production", location="vlan:1" )) # Check for management VLAN mgmt_vlans = [v for v in model.get('vlans', []) if 'mgmt' in v.get('name', '').lower() or 'management' in v.get('name', '').lower()] if not mgmt_vlans and len(model.get('devices', [])) > 0: self.violations.append(PolicyViolation( severity="WARNING", category="design", message="No dedicated management VLAN found - consider creating one for device management", location="vlans" )) # Check VLAN ID ranges (common practice: 10-99 infrastructure, 100+ user) for vlan in model.get('vlans', []): vid = vlan.get('id') name = vlan.get('name', '') if vid and vid >= 1006 and vid <= 1024: self.violations.append(PolicyViolation( severity="WARNING", category="design", message=f"VLAN {vid} is in extended range reserved range - may not be supported on all devices", location=f"vlan:{vid}" )) def _check_security_policies(self, model: Dict[str, Any]): """Check security best practices""" # Check for guest network isolation guest_vlans = [v for v in model.get('vlans', []) if 'guest' in v.get('name', '').lower() or 'visitor' in v.get('name', '').lower()] if guest_vlans: # Guest networks should be isolated (different subnet range) guest_subnets = [s for s in model.get('subnets', []) if any(s.get('vlan') == gv.get('id') for gv in guest_vlans)] if not guest_subnets: self.violations.append(PolicyViolation( severity="WARNING", category="security", message="Guest VLAN exists but no dedicated subnet configured", location="guest network" )) # Check for DHCP snooping (implied by DHCP configuration) for subnet in model.get('subnets', []): if subnet.get('dhcp_enabled'): # Should have DHCP range defined if not subnet.get('dhcp_range_start') or not subnet.get('dhcp_range_end'): self.violations.append(PolicyViolation( severity="WARNING", category="design", message="DHCP enabled but no pool range defined", location=f"subnet:{subnet.get('network')}" )) # Warn if no redundancy in design core_devices = [d for d in model.get('devices', []) if d.get('role') == 'core'] if len(core_devices) < 2 and len(model.get('devices', [])) > 3: self.violations.append(PolicyViolation( severity="WARNING", category="design", message="No redundant core devices - consider adding redundancy for HA", location="devices" )) def _check_design_practices(self, model: Dict[str, Any]): """Check general design best practices""" # Every subnet should have a VLAN vlans_with_subnets = {s.get('vlan') for s in model.get('subnets', []) if s.get('vlan')} all_vlans = {v.get('id') for v in model.get('vlans', [])} vlans_without_subnets = all_vlans - vlans_with_subnets if vlans_without_subnets: self.violations.append(PolicyViolation( severity="INFO", category="design", message=f"VLANs without subnets: {sorted(vlans_without_subnets)}", location="vlans/subnets" )) # Check routing protocol makes sense for network size routing = model.get('routing') device_count = len(model.get('devices', [])) if routing: protocol = routing.get('protocol') if protocol == 'ospf' and device_count < 3: self.violations.append(PolicyViolation( severity="INFO", category="design", message="OSPF may be overkill for small network (<3 devices) - consider static routing", location="routing" )) if protocol == 'static' and device_count > 10: self.violations.append(PolicyViolation( severity="WARNING", category="design", message="Static routing challenging for large networks - consider dynamic protocol", location="routing" )) # Check for services configuration services = model.get('services', []) recommended_services = {'DHCP', 'DNS', 'NTP'} missing_services = recommended_services - set(services) if missing_services: self.violations.append(PolicyViolation( severity="INFO", category="design", message=f"Consider adding services: {', '.join(missing_services)}", location="services" )) def get_violations_by_severity(self) -> Dict[str, List[PolicyViolation]]: """Group violations by severity""" result = {'ERROR': [], 'WARNING': [], 'INFO': []} for v in self.violations: result[v.severity].append(v) return result def has_errors(self) -> bool: """Check if there are any ERROR-level violations""" return any(v.severity == 'ERROR' for v in self.violations) def format_violations(self) -> str: """Format violations as readable text""" if not self.violations: return "✓ No policy violations detected" lines = [] by_severity = self.get_violations_by_severity() for severity in ['ERROR', 'WARNING', 'INFO']: viols = by_severity[severity] if viols: lines.append(f"\n{severity}S ({len(viols)}):") for v in viols: lines.append(f" • {v}") return "\n".join(lines)