226 lines
7.6 KiB
Python
226 lines
7.6 KiB
Python
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
|