import os import re import logging from typing import Optional import yaml logger = logging.getLogger(__name__) _MAC_RE = re.compile(r"^([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$") _REQUIRED_CONFIG_KEYS = ["network", "server", "database", "rate_limiting", "profile"] def validate_mac(mac: str) -> bool: """Return True if mac is a valid colon-separated MAC address.""" return bool(mac and _MAC_RE.match(mac)) def normalize_mac(mac: str) -> str: """Return a lowercase, colon-separated MAC address. Accepts input with dashes, dots, or no separators. """ clean = mac.lower().replace("-", ":").replace(".", ":") # Handle 12-char no-separator format if ":" not in clean and len(clean) == 12: clean = ":".join(clean[i : i + 2] for i in range(0, 12, 2)) return clean def validate_imsi(imsi: str) -> bool: """Return True if imsi is exactly 15 decimal digits.""" return bool(imsi) and imsi.isdigit() and len(imsi) == 15 def validate_op_key(op_key: str) -> bool: """Return True if op_key is a 16-byte (32 hex digit) hex string.""" if not isinstance(op_key, str) or len(op_key) != 32: return False try: bytes.fromhex(op_key) except ValueError: return False return True def load_config(path: str) -> dict: """Load and minimally validate a YAML config file.""" if not os.path.exists(path): raise FileNotFoundError(f"Config file not found: {path}") with open(path) as f: cfg = yaml.safe_load(f) if not isinstance(cfg, dict): raise ValueError(f"Config file is empty or not a mapping: {path}") for key in _REQUIRED_CONFIG_KEYS: if key not in cfg: raise ValueError(f"Missing required config section: '{key}'") op_key = cfg["network"].get("op_key") if not validate_op_key(op_key): raise ValueError( "network.op_key must be a 16-byte hex string (32 hex digits)" ) return cfg def ensure_parent_dir(path: str) -> None: """Create the parent directory of ``path`` if it has one. Guards against ``os.makedirs("")`` raising FileNotFoundError when ``path`` is a bare filename with no directory component. """ parent = os.path.dirname(path) if parent: os.makedirs(parent, exist_ok=True) def hex_to_bytes(h: str) -> bytes: return bytes.fromhex(h) def bytes_to_hex(b: bytes) -> str: return b.hex() def get_client_ip(request) -> str: """Return the real client IP, respecting X-Forwarded-For from nginx.""" forwarded = request.headers.get("X-Forwarded-For") if forwarded: return forwarded.split(",")[0].strip() return request.remote_addr or "" def get_mac_from_arp(ip: str, arp_path: str = "/proc/net/arp") -> Optional[str]: """Attempt to resolve a MAC address from the ARP table for the given IP.""" try: with open(arp_path) as f: for line in f: parts = line.split() if len(parts) >= 4 and parts[0] == ip: mac = parts[3] if validate_mac(mac): return normalize_mac(mac) except Exception as exc: logger.debug(f"ARP lookup failed for {ip}: {exc}") return None