import hashlib from dataclasses import dataclass, field import os import re import time from typing import Any import requests import urllib3 from dns_sync.adapters import Endpoint from dns_sync.normalizer import normalize_hostname, normalize_ipv4 LUCI_USERNAME = ( "edbd881f1ee2f76ba0bd70fd184f87711be991a0401fd07ccd4b199665f00761" "afc91731d8d8ba6cbb188b2ed5bfb465b9f3d30231eb0430b9f90fe91d136648" ) CFG_RE = re.compile(r'addCfg\("dns_(ip|name)(\d+)",\s*"([^"]*)",\s*"([^"]*)"\);') @dataclass(frozen=True) class VerizonRecord: hostname: str ip: str record_type: str = "A" metadata: dict[str, Any] = field(default_factory=dict) class _Slot: def __init__(self, slot): self.slot = slot self.hostname = "" self.ip = "" self.name_token = "" self.ip_token = "" class VerizonClient: def __init__( self, endpoint: Endpoint, ): self.endpoint = endpoint if endpoint.plugin != "verizon": raise RuntimeError( "VerizonClient requires a verizon endpoint" ) self.router_url = f"https://{endpoint.address}" self.password = os.getenv("ROUTER_PASSWORD") if not self.password: raise RuntimeError("ROUTER_PASSWORD is required by the Verizon adapter") self.timeout = int(os.getenv("ROUTER_TIMEOUT", "10")) self.session = requests.Session() self.session.verify = False urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.auth_token = None self._last_slot_count = 0 def _url(self, path): return f"{self.router_url}{path}" def connect(self): response = self.session.get(self._url("/loginStatus.cgi"), timeout=self.timeout) response.raise_for_status() login_token = response.json().get("loginToken") if not login_token: raise RuntimeError("Router did not return loginToken") md5_password = hashlib.md5(self.password.encode("utf-8")).hexdigest() sha512_md5 = hashlib.sha512(md5_password.encode("ascii")).hexdigest() luci_password = hashlib.sha512((login_token + sha512_md5).encode("ascii")).hexdigest() response = self.session.post( self._url("/login.cgi"), data={ "luci_username": LUCI_USERNAME, "luci_password": luci_password, "luci_view": "Desktop", "luci_token": login_token, "luci_keep_login": "0", }, allow_redirects=False, timeout=self.timeout, ) if response.status_code not in (200, 302): raise RuntimeError(f"Router login failed: HTTP {response.status_code}") response = self.session.get(self._url("/loginStatus.cgi"), timeout=self.timeout) response.raise_for_status() self.auth_token = response.json().get("token") if not self.auth_token: raise RuntimeError("Router login succeeded but authenticated token was missing") if "sysauth" not in self.session.cookies: raise RuntimeError("Router login succeeded but sysauth cookie was missing") def _fetch_slots(self): if not self.auth_token: self.connect() response = self.session.get( self._url("/cgi/cgi_dns_server.js"), timeout=self.timeout, headers={"Cache-Control": "no-cache"}, ) response.raise_for_status() slots = {} for kind, slot_s, token, value in CFG_RE.findall(response.text): slot_num = int(slot_s) slot = slots.setdefault(slot_num, _Slot(slot_num)) if kind == "ip": slot.ip = value slot.ip_token = token else: slot.hostname = value slot.name_token = token if not slots: raise RuntimeError( "No Verizon dns_ip/dns_name slots were found; authentication may have expired or the router format changed" ) self._last_slot_count = len(slots) return slots @staticmethod def _find(slots, hostname): target = hostname.rstrip(".").lower() for slot in slots.values(): if slot.hostname.rstrip(".").lower() == target: return slot return None @staticmethod def _first_empty(slots): for slot_num in sorted(slots): slot = slots[slot_num] if not slot.hostname and not slot.ip: return slot return None def list_records(self): slots = self._fetch_slots() result = {} for slot in slots.values(): if not slot.hostname: continue hostname, wildcard = normalize_hostname(slot.hostname, allow_wildcard=True) if wildcard: # The firmware UI has no wildcard support, but normalize defensively. pass ip = normalize_ipv4(slot.ip, hostname) result[hostname.lower()] = VerizonRecord( hostname=hostname, ip=ip, record_type="A", metadata={"slot": slot.slot}, ) return result def capacity(self): slots = self._fetch_slots() used = sum(1 for slot in slots.values() if slot.hostname or slot.ip) return used, len(slots) - used def _write_slot(self, slot, hostname, ip): if not self.auth_token: raise RuntimeError("Verizon client is not authenticated") if not slot.ip_token or not slot.name_token: raise RuntimeError(f"Slot {slot.slot} is missing dynamic field tokens") response = self.session.post( self._url("/apply_abstract.cgi"), data={ "token": self.auth_token, "action": "dnsmasq_reload", slot.ip_token: ip, slot.name_token: hostname, }, headers={"Origin": self.router_url, "Referer": f"{self.router_url}/"}, timeout=self.timeout, ) response.raise_for_status() def add_record(self, record): slots = self._fetch_slots() # fresh opaque tokens before every mutation existing = self._find(slots, record.hostname) if existing: raise RuntimeError(f"{record.hostname} already exists in Verizon slot {existing.slot}") slot = self._first_empty(slots) if slot is None: raise RuntimeError("No empty Verizon IPv4 DNS slots are available") self._write_slot(slot, record.hostname, record.ip) return f"slot {slot.slot}" def update_record(self, record): slots = self._fetch_slots() # fresh opaque tokens before every mutation slot = self._find(slots, record.hostname) if slot is None: raise RuntimeError(f"{record.hostname} does not exist in Verizon DNS") self._write_slot(slot, record.hostname, record.ip) return f"slot {slot.slot}" def delete_record(self, hostname): slots = self._fetch_slots() slot = self._find(slots, hostname) if slot is None: return None self._write_slot(slot, "", "") return f"slot {slot.slot}" def verify_record(self, record): attempts = int(os.getenv("VERIFY_ATTEMPTS", "5")) delay = float(os.getenv("VERIFY_DELAY", "1")) for _ in range(attempts): slots = self._fetch_slots() slot = self._find(slots, record.hostname) if slot is not None and slot.ip == record.ip: return True time.sleep(delay) return False