#!/usr/bin/env python3 """LAN infrastructure (gateway / switch / AP) discovery + read-only login insight. Hard rules: - Credentials stay on the customer Protect box (0600 JSON). - READ-ONLY only: login to read info / tips — never change device config. - Any mutate/write requires fn.nology@gmail.com authorization (blocked until write tools ship). """ from __future__ import annotations import base64 import json import os import re import socket import subprocess import time import urllib.error import ssl import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any def _urlopen_any(req, timeout): """urlopen that tolerates self-signed TLS on http->https redirects. ReyeeOS EG routers force HTTPS on their LuCI web UI (http is answered with a JS/https redirect). urllib's default context rejects the self-signed certificate, so any Reyee login/probe/scan silently fails on those gateways. Retry with an unverified context (read-only RPCs).""" try: return urllib.request.urlopen(req, timeout=timeout) except urllib.error.URLError as exc: reason = str(getattr(exc, "reason", exc) or exc) if "CERTIFICATE_VERIFY_FAILED" not in reason and "self-signed" not in reason and "SSL" not in reason: raise opener = urllib.request.build_opener( urllib.request.HTTPSHandler(context=ssl._create_unverified_context()) ) return opener.open(req, timeout=timeout) DATA = Path("/var/lib/atlas-cyber-protect") STATE = DATA / "state.json" INFRA_CREDS = DATA / "infra-creds.json" INFRA_CACHE = DATA / "infra-inventory.json" # Legacy single-gateway file (still supported) LEGACY_CREDS = DATA / "gateway-creds.json" OWNER_EMAIL = "fn.nology@gmail.com" INFRA_ICONS = {"gateway", "ap", "switch", "protect"} INFRA_TYPES = { "gateway", "router", "router/ap", "ap", "switch", "access point", "wifi", } MGMT_PORTS = (22, 23, 80, 443, 8080, 8443, 8291, 161, 8728, 8729) def is_owner_email(email: str | None) -> bool: return (email or "").strip().casefold() == OWNER_EMAIL def _load_json(path: Path, default: Any) -> Any: try: if path.is_file(): return json.loads(path.read_text(encoding="utf-8")) except Exception: pass return default def _save_json(path: Path, payload: Any) -> None: DATA.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2), encoding="utf-8") try: os.chmod(path, 0o600) except Exception: pass def default_gateway_ip() -> str: try: out = subprocess.check_output( ["ip", "-j", "route", "show", "default"], text=True, stderr=subprocess.DEVNULL, timeout=5, ) rows = json.loads(out or "[]") if rows: return str(rows[0].get("gateway") or "").strip() except Exception: pass return "" def _mask(pw: str) -> str: if not pw: return "" return "*" * min(12, max(4, len(pw))) def _load_store() -> dict[str, Any]: raw = _load_json(INFRA_CREDS, {}) if not isinstance(raw, dict): raw = {} default = raw.get("default") if isinstance(raw.get("default"), dict) else {} devices = raw.get("devices") if isinstance(raw.get("devices"), dict) else {} # Migrate legacy main-gateway creds once if not (default.get("username") and default.get("password")): legacy = _load_json(LEGACY_CREDS, {}) if isinstance(legacy, dict) and legacy.get("username") and legacy.get("password"): default = { "username": legacy.get("username") or "", "password": legacy.get("password") or "", "notes": legacy.get("notes") or "Migrated main gateway login", } gip = str(legacy.get("gateway_ip") or "").strip() if gip and gip not in devices: devices[gip] = { "username": default["username"], "password": default["password"], "notes": "Main gateway", "role": "gateway", } raw = {"default": default, "devices": devices, "updated_at": time.time()} _save_json(INFRA_CREDS, raw) return { "default": { "username": str(default.get("username") or "").strip(), "password": str(default.get("password") or ""), "notes": str(default.get("notes") or "").strip(), }, "devices": devices, "updated_at": raw.get("updated_at") or "", } def public_status() -> dict[str, Any]: store = _load_store() d = store["default"] configured = bool(d.get("username") and d.get("password")) per = [] for ip, row in (store.get("devices") or {}).items(): if not isinstance(row, dict): continue per.append( { "ip": ip, "username": str(row.get("username") or ""), "password_set": bool(row.get("password")), "password_masked": _mask(str(row.get("password") or "")), "notes": str(row.get("notes") or ""), "role": str(row.get("role") or ""), } ) cache = _load_json(INFRA_CACHE, {}) items = cache.get("items") if isinstance(cache, dict) else [] return { "ok": True, "mode": "read_only", "mutate_requires": OWNER_EMAIL, "configured": configured, "username": d.get("username") or "", "password_set": bool(d.get("password")), "password_masked": _mask(str(d.get("password") or "")), "notes": d.get("notes") or "", "device_logins": per, "inventory_count": len(items) if isinstance(items, list) else 0, "last_scan_at": (cache.get("checked_at") if isinstance(cache, dict) else "") or "", "policy": ( "Logins for gateways, switches and APs stay on this Protect box. " f"Atlas only READs device info for tips. Never changes gear unless authorized by {OWNER_EMAIL}." ), } def save_default_login(*, username: str, password: str, notes: str = "", keep_password: bool = False) -> dict[str, Any]: store = _load_store() cur = store["default"] pw = password if password else (cur.get("password") if keep_password else "") if not (username or "").strip(): raise ValueError("username required") if not pw: raise ValueError("password required") store["default"] = { "username": username.strip()[:120], "password": pw, "notes": (notes or "").strip()[:400], } store["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) # Also mirror to legacy gateway-creds for older UI paths _save_json( LEGACY_CREDS, { "username": store["default"]["username"], "password": store["default"]["password"], "gateway_ip": default_gateway_ip(), "notes": store["default"]["notes"], "updated_at": store["updated_at"], "mode": "read_only", "mutate_requires": OWNER_EMAIL, }, ) _save_json(INFRA_CREDS, store) return public_status() def save_device_login( *, ip: str, username: str = "", password: str = "", notes: str = "", role: str = "", keep_password: bool = False, clear: bool = False, ) -> dict[str, Any]: ip = (ip or "").strip() if not ip: raise ValueError("device ip required") store = _load_store() devices = dict(store.get("devices") or {}) if clear: devices.pop(ip, None) else: cur = devices.get(ip) if isinstance(devices.get(ip), dict) else {} user = (username or cur.get("username") or store["default"].get("username") or "").strip() pw = password if password else (cur.get("password") if keep_password else "") if not pw: pw = store["default"].get("password") or "" if not user or not pw: raise ValueError("username and password required (or set default login first)") devices[ip] = { "username": user[:120], "password": pw, "notes": (notes or cur.get("notes") or "")[:400], "role": (role or cur.get("role") or "")[:40], "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } store["devices"] = devices store["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) _save_json(INFRA_CREDS, store) return public_status() def clear_all_logins() -> dict[str, Any]: try: if INFRA_CREDS.is_file(): INFRA_CREDS.unlink() except Exception: pass try: if LEGACY_CREDS.is_file(): LEGACY_CREDS.unlink() except Exception: pass return public_status() def refuse_mutate(email: str | None) -> dict[str, Any]: if is_owner_email(email): return { "ok": False, "allowed": False, "error": ( "Owner seen, but write/change tools for gateways/switches/APs are not enabled yet. " "Atlas stays read-only." ), "mutate_requires": OWNER_EMAIL, } return { "ok": False, "allowed": False, "error": f"Blocked. Only {OWNER_EMAIL} may authorize changes on gateways, switches or APs.", "mutate_requires": OWNER_EMAIL, } def _is_infra_device(d: dict) -> bool: icon = str(d.get("icon_key") or "").lower() typ = str(d.get("device_type") or "").lower() role = str(d.get("role_guess") or "").lower() vendor = str(d.get("vendor") or "").lower() product = str(d.get("product_hint") or "").lower() host = str(d.get("hostname") or "").lower() blob = f"{icon} {typ} {role} {vendor} {product} {host}" if icon in INFRA_ICONS and icon != "protect": return True if any(t in typ for t in ("gateway", "router", "ap", "switch")): return True if any(k in blob for k in ("mikrotik", "routerboard", "ubiquiti", "unifi", "tp-link", "cisco", "netgear", "asus", "openwrt")): return True # Management ports open on something that looks like network gear ports = _parse_ports(d.get("open_ports")) if ports.intersection({80, 443, 22, 23, 8291, 8728}) and any( k in blob for k in ("router", "gateway", "switch", "ap", "wifi", "unifi", "mikrotik") ): return True return False def _parse_ports(raw: Any) -> set[int]: out: set[int] = set() if isinstance(raw, list): for p in raw: try: out.add(int(p)) except Exception: pass return out for part in str(raw or "").replace(";", ",").split(","): part = part.strip() if part.isdigit(): out.add(int(part)) return out def _role_for(d: dict) -> str: icon = str(d.get("icon_key") or "").lower() typ = str(d.get("device_type") or "").lower() if icon == "switch" or "switch" in typ: return "switch" if icon == "ap" or "ap" in typ or "access point" in typ: return "ap" if icon == "gateway" or "gateway" in typ or "router" in typ: return "gateway" blob = f"{d.get('vendor')} {d.get('product_hint')} {d.get('hostname')}".lower() if "unifi" in blob or "ubiquiti" in blob: return "ap" if "mikrotik" in blob or "routerboard" in blob: return "gateway" return "network" def discover_infra_candidates(devices: list[dict] | None = None) -> list[dict[str, Any]]: if devices is None: st = _load_json(STATE, {}) devices = st.get("last_devices") if isinstance(st, dict) else [] if not isinstance(devices, list): devices = [] gw = default_gateway_ip() found: dict[str, dict[str, Any]] = {} for d in devices: if not isinstance(d, dict): continue ip = str(d.get("ip") or "").strip() if not ip or ip.endswith(".255") or ip.startswith("127."): continue # Always include default gateway is_gw = ip == gw if not is_gw and not _is_infra_device(d): continue role = "gateway" if is_gw else _role_for(d) found[ip] = { "ip": ip, "mac": str(d.get("mac") or ""), "hostname": str(d.get("hostname") or ""), "vendor": str(d.get("vendor") or ""), "product_hint": str(d.get("product_hint") or ""), "device_type": str(d.get("device_type") or ""), "icon_key": str(d.get("icon_key") or role), "role": role, "open_ports_scan": sorted(_parse_ports(d.get("open_ports"))), "is_default_gateway": is_gw, } # Ensure default gateway exists even if not in ARP list yet if gw and gw not in found: found[gw] = { "ip": gw, "mac": "", "hostname": "", "vendor": "", "product_hint": "Default gateway", "device_type": "gateway", "icon_key": "gateway", "role": "gateway", "open_ports_scan": [], "is_default_gateway": True, } # Prefer gateways first, then switches, APs order = {"gateway": 0, "switch": 1, "ap": 2, "network": 3} return sorted(found.values(), key=lambda x: (order.get(x.get("role"), 9), x.get("ip") or "")) def _tcp_open(ip: str, port: int, timeout: float = 0.9) -> bool: try: with socket.create_connection((ip, port), timeout=timeout): return True except Exception: return False def _page_title(html: str) -> str: m = re.search(r"]*>(.*?)", html or "", re.I | re.S) if not m: return "" return re.sub(r"\s+", " ", m.group(1)).strip()[:120] def _http_get(url: str, *, username: str = "", password: str = "", timeout: float = 5.0) -> dict[str, Any]: headers = {"User-Agent": "AtlasCyberProtect/infra-read-only", "Accept": "*/*"} if username: token = base64.b64encode(f"{username}:{password}".encode()).decode() headers["Authorization"] = f"Basic {token}" req = urllib.request.Request(url, headers=headers, method="GET") try: with _urlopen_any(req, timeout=timeout) as resp: raw = resp.read(80_000) text = raw.decode("utf-8", errors="replace") return { "ok": True, "status": getattr(resp, "status", 200), "url": url, "server": resp.headers.get("Server") or "", "title": _page_title(text), "snippet": text[:500], "json": _maybe_json(text), } except urllib.error.HTTPError as exc: body = "" try: body = (exc.read() or b"")[:500].decode("utf-8", errors="replace") except Exception: body = "" return { "ok": False, "status": int(exc.code), "url": url, "server": exc.headers.get("Server") if exc.headers else "", "title": _page_title(body), "error": f"HTTP {exc.code}", "snippet": body[:200], "json": _maybe_json(body), } except Exception as exc: return {"ok": False, "url": url, "error": str(exc)[:160]} def _maybe_json(text: str) -> Any: t = (text or "").strip() if not t.startswith("{") and not t.startswith("["): return None try: return json.loads(t) except Exception: return None def _fingerprint(title: str, server: str, body: str, vendor: str = "") -> str: blob = f"{title} {server} {body[:600]} {vendor}".lower() rules = [ ("mikrotik", ("mikrotik", "routeros", "webfig", "routerboard")), ("ubiquiti", ("unifi", "ubnt", "ubiquiti")), ("tp-link", ("tp-link", "tplink")), ("cisco", ("cisco",)), ("netgear", ("netgear",)), ("asus", ("asus", "asuswrt")), ("openwrt", ("openwrt", "luci")), ("pfSense", ("pfsense",)), ("ruijie", ("ruijie", "reyee", "eg105g", "eg210g", "eg205g", "eg_3.0", "reyeeos")), ] for name, keys in rules: if any(k in blob for k in keys): return name return "unknown" def _creds_for(ip: str, store: dict[str, Any]) -> tuple[str, str]: row = store.get("devices", {}).get(ip) if isinstance(store.get("devices"), dict) else None if isinstance(row, dict) and row.get("username") and row.get("password"): return str(row["username"]), str(row["password"]) d = store.get("default") or {} return str(d.get("username") or ""), str(d.get("password") or "") def _probe_device(item: dict[str, Any], store: dict[str, Any]) -> dict[str, Any]: ip = item["ip"] user, password = _creds_for(ip, store) ports = {p: _tcp_open(ip, p) for p in MGMT_PORTS} open_ports = sorted(p for p, ok in ports.items() if ok) brand = "unknown" login_ok = False info: dict[str, Any] = {} admin_title = "" admin_server = "" urls: list[str] = [] if ports.get(80): urls += [f"http://{ip}/", f"http://{ip}/rest/system/resource", f"http://{ip}/rest/system/identity"] if ports.get(443): urls += [f"https://{ip}/", f"https://{ip}/rest/system/resource", f"https://{ip}/rest/system/identity"] if ports.get(8080): urls.append(f"http://{ip}:8080/") if ports.get(8443): urls.append(f"https://{ip}:8443/") for url in urls: need_auth = bool(user) and ("/rest/" in url) r = _http_get(url, username=user if need_auth else "", password=password if need_auth else "") if not r.get("ok") and user and r.get("status") in (401, 403): r = _http_get(url, username=user, password=password) if r.get("title") or r.get("server"): admin_title = admin_title or str(r.get("title") or "") admin_server = admin_server or str(r.get("server") or "") brand = _fingerprint(admin_title, admin_server, str(r.get("snippet") or ""), item.get("vendor") or "") if r.get("ok") and user: login_ok = True if isinstance(r.get("json"), dict): # MikroTik REST resource/identity js = r["json"] if "version" in js or "board-name" in js or "architecture-name" in js: info["routeros"] = { "version": js.get("version"), "board": js.get("board-name") or js.get("board_name"), "arch": js.get("architecture-name") or js.get("architecture_name"), "uptime": js.get("uptime"), "cpu_load": js.get("cpu-load") or js.get("cpu_load"), "free_memory": js.get("free-memory") or js.get("free_memory"), "total_memory": js.get("total-memory") or js.get("total_memory"), } brand = "mikrotik" if "name" in js and "/identity" in url: info["identity"] = js.get("name") login_ok = True if login_ok and brand != "unknown" and ("/rest/" in url or r.get("ok")): break if brand == "unknown": brand = _fingerprint(admin_title, admin_server, "", item.get("vendor") or "") tips = _device_tips(item, brand, open_ports, login_ok, bool(user and password)) return { **item, "brand": brand, "open_ports": open_ports, "admin_title": admin_title, "admin_server": admin_server, "login_configured": bool(user and password), "login_ok": login_ok, "info": info, "tips": tips, "mode": "read_only", } def _device_tips( item: dict[str, Any], brand: str, open_ports: list[int], login_ok: bool, configured: bool, ) -> list[dict[str, str]]: tips: list[dict[str, str]] = [] role = item.get("role") or "network" ip = item.get("ip") if 23 in open_ports: tips.append({"level": "critical", "title": "Telnet open", "detail": f"Disable Telnet on {ip} ({role})."}) if 21 in open_ports: tips.append({"level": "warn", "title": "FTP open", "detail": f"FTP on {ip} is cleartext — disable if unused."}) if 80 in open_ports and 443 not in open_ports: tips.append({"level": "warn", "title": "HTTP admin only", "detail": f"Prefer HTTPS admin on {ip}."}) if configured and login_ok: tips.append({"level": "ok", "title": "Login OK (read)", "detail": f"Protect read {brand} info from {ip}."}) elif configured and not login_ok: tips.append( { "level": "warn", "title": "Login not confirmed", "detail": f"Saved login did not authenticate on {ip} yet (vendor form/API may differ).", } ) elif not configured: tips.append( { "level": "info", "title": "Add login to see more", "detail": f"Save default or per-device login to read deeper {role} info from {ip}.", } ) if brand == "mikrotik": tips.append( { "level": "info", "title": "MikroTik", "detail": "Keep RouterOS updated; do not expose Winbox/WebFig to WAN; Atlas reads REST only.", } ) if brand == "ubiquiti": tips.append( { "level": "info", "title": "Ubiquiti / UniFi", "detail": "Confirm remote access is intentional; review AP adoption and guest isolation.", } ) if item.get("is_default_gateway"): tips.append( { "level": "info", "title": "Default gateway", "detail": f"{ip} is this Protect box’s default route — primary router for Atlas Secure tips.", } ) tips.append( { "level": "info", "title": "Read-only", "detail": f"No config changes on {ip} unless authorized by {OWNER_EMAIL}.", } ) return tips def scan_infrastructure(*, force: bool = True) -> dict[str, Any]: """Discover gateways/switches/APs and read info when logins are available.""" cache = _load_json(INFRA_CACHE, {}) if ( not force and isinstance(cache, dict) and cache.get("items") and (time.time() - float(cache.get("checked_ts") or 0)) < 90 ): return cache store = _load_store() candidates = discover_infra_candidates() items: list[dict[str, Any]] = [] # Bound parallelism so we don't hammer the LAN with ThreadPoolExecutor(max_workers=6) as pool: futs = {pool.submit(_probe_device, c, store): c["ip"] for c in candidates} for fut in as_completed(futs): try: items.append(fut.result()) except Exception as exc: ip = futs[fut] items.append({"ip": ip, "ok": False, "error": str(exc)[:160], "role": "network", "tips": []}) order = {"gateway": 0, "switch": 1, "ap": 2, "network": 3} items.sort(key=lambda x: (order.get(str(x.get("role")), 9), str(x.get("ip") or ""))) summary = { "gateways": sum(1 for i in items if i.get("role") == "gateway"), "switches": sum(1 for i in items if i.get("role") == "switch"), "aps": sum(1 for i in items if i.get("role") == "ap"), "login_ok": sum(1 for i in items if i.get("login_ok")), "total": len(items), } out = { "ok": True, "mode": "read_only", "mutate_requires": OWNER_EMAIL, "checked_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "checked_ts": time.time(), "default_gateway": default_gateway_ip(), "credentials_configured": bool(store["default"].get("username") and store["default"].get("password")), "summary": summary, "items": items, "policy": ( "Atlas scanned this LAN for gateways, switches and APs. With logins, Protect reads device info only. " f"Never changes gear unless authorized by {OWNER_EMAIL}." ), } _save_json(INFRA_CACHE, out) return out def inventory_public() -> dict[str, Any]: cache = _load_json(INFRA_CACHE, {}) if isinstance(cache, dict) and cache.get("items"): # Strip heavy snippets if any return cache # Lightweight discover without full probe items = discover_infra_candidates() return { "ok": True, "mode": "read_only", "mutate_requires": OWNER_EMAIL, "checked_at": "", "default_gateway": default_gateway_ip(), "credentials_configured": public_status().get("configured"), "summary": { "gateways": sum(1 for i in items if i.get("role") == "gateway"), "switches": sum(1 for i in items if i.get("role") == "switch"), "aps": sum(1 for i in items if i.get("role") == "ap"), "login_ok": 0, "total": len(items), }, "items": items, "policy": public_status().get("policy"), "needs_scan": True, } # --------------------------------------------------------------------------- # Ruijie / ReyeeOS gateway poller (EG105G-V3 and friends) # # The customer's main gateway is a Ruijie Reyee EG-series router (e.g. # RG-EG105G-V3) running ReyeeOS. Its web UI is a JSON-RPC API under # /cgi-bin/luci/api/*. We log in with the saved credentials (read-only) and # pull live WAN status, real-time throughput, link info, and connected WiFi / # LAN clients, then hand the snapshot to the agent heartbeat so Atlas shows # the internet line + live traffic + client states. # # Hard rules: READ-ONLY. We never change router config. Credentials stay in # the box's 0600 store (infra-creds.json). If login or any call fails we # return a degraded snapshot with error notes - the box + Atlas keep working. # --------------------------------------------------------------------------- # ReyeeOS JSON-RPC methods used by the EG-series web UI. Some firmware # versions expose slightly different names; we try several per concern. REYEE_LOGIN_PATH = "/cgi-bin/luci/api/auth" REYEE_API_BASE = "/cgi-bin/luci/api" REYEE_TIMEOUT = 8.0 # ReyeeOS web-login encryption (CVE-2021-43164 research / open-source # GibberishAES): the browser encrypts the password with AES-256-CBC using this # static key before POSTing, with encry=true. We replicate it in pure Python. REYEE_AES_KEY = "RjYkhwzx$2018!" def _reyee_encrypt_pw(password: str) -> str: """Mirror the ReyeeOS web login: openssl enc -aes-256-cbc -a -k key -md md5. This is exactly what the router's own web UI does (GibberishAES) and what the CVE-2021-43164 PoC uses. openssl is present on the Protect box image; if it is ever missing we fall back to plaintext below. """ import subprocess p = subprocess.run( [ "openssl", "enc", "-aes-256-cbc", "-a", "-k", REYEE_AES_KEY, "-md", "md5", ], input=password.encode(), capture_output=True, timeout=8, ) if p.returncode != 0: raise RuntimeError(f"openssl encrypt failed: {p.stderr.decode(errors='replace')[:120]}") return p.stdout.decode("ascii", errors="ignore").replace("\n", "").replace("\r", "") REYEE_METHODS = { "status": ["getSystemStatus", "getStatus", "getSysInfo", "getSystemInfo"], "wan": ["getWanStatus", "getWanInfo", "getWanList", "getInternetStatus", "getWan"], "traffic": ["getTraffic", "getRealTimeTraffic", "getFlow", "getNetTraffic"], "clients": ["getClientList", "getDeviceList", "getHostList", "getLanClients", "getOnlineClient"], "wifi": ["getWifiStatus", "getWirelessStatus", "getWifiInfo", "getWirelessInfo"], "link": ["getLinkStatus", "getLinkInfo", "getUplinkStatus"], } REYEE_CREDS_FILE = DATA / "ruijie-creds.json" REYEE_CACHE_FILE = DATA / "ruijie-cache.json" def _reyee_load_creds() -> dict[str, Any]: """Return Ruijie gateway credentials (default store or dedicated file).""" store = _load_store() d = store.get("default") or {} if d.get("username") and d.get("password"): return {"username": str(d["username"]), "password": str(d["password"]), "source": "default"} raw = _load_json(REYEE_CREDS_FILE, {}) if isinstance(raw, dict) and raw.get("username") and raw.get("password"): return {"username": str(raw["username"]), "password": str(raw["password"]), "source": "ruijie-file"} return {} def _reyee_rpc(host: str, sid: str, method: str, params: dict[str, Any] | None = None, api_url: str = "", timeout: float = REYEE_TIMEOUT) -> dict[str, Any]: """Call a ReyeeOS JSON-RPC method with the session id.""" # api_url is a path (with auth token); host is always prepended. url = f"http://{host}{api_url}" if api_url else f"http://{host}{REYEE_API_BASE}?auth={sid}" payload = {"method": method, "params": params or {}, "id": f"atlas-{int(time.time()*1000)}"} try: req = urllib.request.Request( url, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", "User-Agent": "AtlasCyberProtect/infra-read-only"}, method="POST", ) with _urlopen_any(req, timeout=timeout) as resp: raw = resp.read(200_000) text = raw.decode("utf-8", errors="replace") try: data = json.loads(text) except Exception: # Non-JSON reply (e.g. HTML login redirect) — surface a hint. head = " ".join(text.split())[:160] return {"code": -1, "error": "non-JSON reply", "head": head} if isinstance(data, dict) and data.get("code") == 0: return data return {"code": -1, "error": str(data.get("error") or data)[:160]} except Exception as exc: return {"code": -1, "error": str(exc)[:160]} def _reyee_login(host: str) -> tuple[str, str]: """Log into ReyeeOS. Returns (token_or_sid, error). Newer ReyeeOS 2.4xx returns both `token` and `sid`; the API accepts the `token` (used as the stok). Older builds only have `sid`. """ creds = _reyee_load_creds() if not creds: return "", "no credentials configured" # Trial order: (encrypted, encry=true) first — newer ReyeeOS (and the EG # series) require it; plaintext only helps ancient firmware. pw_enc = _reyee_encrypt_pw(creds["password"]) trials = [ (pw_enc, True), (creds["password"], False), ] last_err = "login failed" for pw, encry in trials: payload = { "method": "login", "params": { "username": creds["username"], "password": pw, "time": str(int(time.time())), "encry": encry, "limit": False, }, "id": None, } try: req = urllib.request.Request( f"http://{host}{REYEE_LOGIN_PATH}", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", "User-Agent": "AtlasCyberProtect/infra-read-only"}, method="POST", ) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: raw = resp.read(200_000) body = raw.decode("utf-8", errors="replace") data = json.loads(body) if body.strip() else None except Exception as exc: last_err = str(exc)[:160] continue # Tolerate every known ReyeeOS response shape — some firmware puts the # session id at top level, some under data, under .data.session, etc. if isinstance(data, dict): inner = data.get("data") sid = token = None if isinstance(inner, dict): sid = inner.get("sid") or inner.get("session") or inner.get("sessionid") or inner.get("token") or inner.get("auth") or inner.get("cookie") token = inner.get("token") or inner.get("stok") or inner.get("sessionid") if not sid and isinstance(inner.get("data"), dict): sid = inner["data"].get("sid") or inner["data"].get("session") or inner["data"].get("token") token = inner["data"].get("token") or inner["data"].get("stok") if not sid: sid = data.get("sid") or data.get("session") or data.get("sessionid") or data.get("token") or data.get("auth") token = data.get("token") or data.get("stok") if sid or token: # Prefer token — the newer firmware requires it as the stok. return str(token or sid)[:128], "" last_err = str(data.get("error") or data.get("msg") or data.get("message") or "")[:160] if not last_err: last_err = body[:160] continue last_err = f"login non-JSON response: {body[:160]}" return "", last_err def _reyee_parse_wan(wan: Any) -> dict[str, Any]: """Normalise a WAN/internet status blob into a flat dict.""" out: dict[str, Any] = {} if isinstance(wan, dict): for k, v in wan.items(): if k in ("ip", "ipv4", "ipv4addr", "addr", "address", "ip_address", "wan_ip", "gateway", "dns1", "dns2", "dns"): out[k] = str(v)[:80] elif k in ("status", "link", "connected", "online", "state", "link_status", "connection"): out[k] = str(v)[:40] elif k in ("rx_bytes", "tx_bytes", "rx", "tx", "down", "up", "download", "upload", "speed_down", "speed_up", "rx_rate", "tx_rate", "rx_speed", "tx_speed", "down_rate", "up_rate", "bandwidth", "rate"): try: out[k] = float(v) except Exception: out[k] = v elif k in ("uptime", "online_time", "duration"): try: out[k] = int(v) except Exception: out[k] = v elif k in ("name", "type", "mode", "proto", "protocol", "mac", "isp", "operator", "model", "version", "firmware", "wan_type"): out[k] = str(v)[:80] elif k in ("ipv6", "ipv6addr", "ipv6_address"): out[k] = str(v)[:80] return out def _reyee_parse_client(c: Any) -> dict[str, Any]: """Normalise a client/device row into a flat dict.""" out: dict[str, Any] = {} if isinstance(c, dict): for k, v in c.items(): kl = k.lower() if kl in ("mac", "macaddr", "mac_address", "hwaddr", "host_mac"): out["mac"] = str(v).lower()[:24] elif kl in ("ip", "ipaddr", "ip_address", "lan_ip", "ipv4"): out["ip"] = str(v)[:48] elif kl in ("hostname", "name", "devicename", "host_name", "host"): out["hostname"] = str(v)[:80] elif kl in ("online", "online_status", "status", "connected", "active", "is_online"): out["online"] = bool(v) if isinstance(v, bool) else (str(v).lower() in ("1", "true", "yes", "online", "up")) elif kl in ("type", "device_type", "kind", "devtype"): out["type"] = str(v)[:40] elif kl in ("signal", "signal_strength", "rssi", "dbm", "signal_db"): try: out["signal"] = int(v) except Exception: out["signal"] = v elif kl in ("rx_bytes", "tx_bytes", "rx", "tx", "down", "up", "download", "upload", "rx_rate", "tx_rate", "rx_speed", "tx_speed"): try: out[kl] = float(v) except Exception: out[kl] = v elif kl in ("vendor", "manufacturer", "oui", "brand"): out["vendor"] = str(v)[:60] elif kl in ("wired", "is_wired", "link_type", "connection_type", "access"): out["wired"] = bool(v) if isinstance(v, bool) else (str(v).lower() in ("1", "true", "wired", "lan", "ethernet")) return out def _reyee_scan(host: str = "") -> dict[str, Any]: """One full read-only scan of the Reyee gateway.""" host = host or default_gateway_ip() if not host: return {"ok": False, "error": "no gateway ip"} sid, err = _reyee_login(host) if not sid: _save_json( REYEE_CACHE_FILE, {"ok": False, "error": err or "login failed", "gateway_ip": host, "fetched_at": time.time()}, ) return {"ok": False, "error": err or "login failed", "gateway_ip": host, "fetched_at": time.time()} state = {"ok": True, "gateway_ip": host, "connected": True, "method": "reyeeos", "clients": [], "wifi": [], "wan": {}, "traffic": {}, "link": {}, "status": {}, "fetched_at": time.time()} for concern, methods in REYEE_METHODS.items(): for m in methods: r = _reyee_rpc(host, sid, m) if r.get("code") == 0: data = r.get("data") or {} if concern in ("status", "wan", "traffic", "link"): state[concern] = _reyee_parse_wan(data) elif concern in ("clients", "wifi"): items = data if isinstance(data, list) else data.get("list") or data.get("items") or data.get("clients") or data.get("devices") or [] if isinstance(items, list): for c in items: parsed = _reyee_parse_client(c) if parsed.get("mac") or parsed.get("ip") or parsed.get("hostname"): state[concern].append(parsed) if concern == "clients" and state["clients"]: break break return state def _reyee_summary(state: dict[str, Any]) -> dict[str, Any]: """Compress a gateway scan into the small summary the heartbeat carries.""" wan = state.get("wan") or {} traffic = state.get("traffic") or {} link = state.get("link") or {} status = state.get("status") or {} clients = state.get("clients") or [] online = sum(1 for c in clients if c.get("online") is not False) wifi = state.get("wifi") or [] return { "ok": bool(state.get("ok")), "gateway_ip": state.get("gateway_ip") or "", "connected": bool(state.get("connected")), "method": state.get("method") or "", "wan": { "status": wan.get("status") or wan.get("link") or wan.get("state") or "", "ip": wan.get("ip") or wan.get("wan_ip") or wan.get("addr") or "", "gateway": wan.get("gateway") or "", "dns": wan.get("dns") or wan.get("dns1") or "", "mac": wan.get("mac") or "", "isp": wan.get("isp") or wan.get("operator") or "", "download": wan.get("download") or wan.get("down") or wan.get("rx") or wan.get("rx_rate") or wan.get("speed_down") or 0, "upload": wan.get("upload") or wan.get("up") or wan.get("tx") or wan.get("tx_rate") or wan.get("speed_up") or 0, }, "traffic": { "down": traffic.get("down") or traffic.get("rx") or traffic.get("download") or traffic.get("rx_rate") or traffic.get("down_rate") or 0, "up": traffic.get("up") or traffic.get("tx") or traffic.get("upload") or traffic.get("tx_rate") or traffic.get("up_rate") or 0, "down_total": traffic.get("down_total") or traffic.get("rx_bytes") or 0, "up_total": traffic.get("up_total") or traffic.get("tx_bytes") or 0, }, "link": { "status": link.get("status") or link.get("link") or link.get("state") or "", "mode": link.get("mode") or link.get("proto") or link.get("protocol") or "", "uptime": link.get("uptime") or link.get("inet_time") or 0, }, "status": { "model": status.get("model") or status.get("product") or "", "version": status.get("version") or status.get("firmware") or status.get("software") or "", "uptime": status.get("uptime") or 0, "cpu": status.get("cpu") or status.get("cpu_usage") or status.get("cpu_load") or 0, "mem": status.get("mem") or status.get("memory") or status.get("mem_usage") or 0, }, "clients": clients[:60], "wifi": wifi[:60], "clients_online": online, "fetched_at": state.get("fetched_at") or time.time(), } def reyee_scan(*, force: bool = False) -> dict[str, Any]: """Scan the Ruijie gateway. Cached 90s (success OR failure), unless force.""" cache = _load_json(REYEE_CACHE_FILE, {}) if not isinstance(cache, dict): cache = {} if ( not force and (cache.get("ok") or cache.get("error")) and (time.time() - float(cache.get("fetched_at") or 0)) < 90 ): return cache state = _reyee_scan() state["summary"] = _reyee_summary(state) _save_json(REYEE_CACHE_FILE, state) return state def reyee_probe(host: str = "", limit: int = 3_000_000) -> dict[str, Any]: """Pull the router's own web assets to discover its real RPC method names. The EG-series ReyeeOS UI is a JS app; its API method names appear verbatim in the bundles. We fetch the root HTML + referenced JS and extract candidate names. Read-only, size-capped. """ import re as _re host = host or default_gateway_ip() if not host: return {"ok": False, "error": "no gateway ip"} found: dict[str, Any] = {"scripts": [], "methods": [], "urls": [], "html_head": ""} try: req = urllib.request.Request( f"http://{host}/", headers={"User-Agent": "AtlasCyberProtect/probe"} ) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: html = resp.read(200_000).decode("utf-8", errors="replace") if "fromindex" in html or "cgi-bin/luci/?stamp" in html: # Landing page is just an IE-check stub that JS-redirects to # /cgi-bin/luci/ — fetch the actual UI to find its assets. for path in ( "/cgi-bin/luci/?stamp=1", "/cgi-bin/luci/", "/web/index.html", "/static/index.html", "/index.html", ): try: req = urllib.request.Request( f"http://{host}{path}", headers={"User-Agent": "AtlasCyberProtect/probe", "Accept": "text/html"}, ) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: html = resp.read(200_000).decode("utf-8", errors="replace") found["urls"].append(path) if len(html) >= 400 and "fromindex" not in html: break except Exception: continue found["html_head"] = _re.sub(r"\s+", " ", html)[:9000] # any location/redirect targets inside inline JS found["redirects"] = _re.findall(r"location\.(?:href|replace)\s*=\s*['\"]([^'\"]+)", html)[:10] found["title"] = _re.findall(r"(.*?)", html, _re.S | _re.I)[:1] srcs = _re.findall(r"(?:src|href)=[\"']([^\"']+)[\"']", html, _re.I) srcs = [s for s in srcs if (".js" in s.lower() or ".css" in s.lower())] found["scripts"] = srcs[:40] # Login page only carries libs; the main console loads after auth, so # also probe known static dirs for app bundles. blobs_extra: list[str] = [] for extra in ( "/luci-static/web/static/app.js", "/luci-static/web/static/main.js", "/luci-static/web/js/main.js", "/luci-static/web/static/index.js", ): try: req = urllib.request.Request( f"http://{host}{extra}", headers={"User-Agent": "AtlasCyberProtect/probe"} ) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: data = resp.read(120_000) if len(data) > 2000 and not data[:40].lstrip().startswith(b"<"): srcs.append("auto:" + extra) blobs_extra.append(data.decode("utf-8", errors="replace")) except Exception: continue blobs = [html] for src in srcs[:40]: if src.startswith("auto:"): continue url = src if src.startswith("http") else f"http://{host}/{src.lstrip('/')}" try: req = urllib.request.Request(url, headers={"User-Agent": "AtlasCyberProtect/probe"}) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: blobs.append(resp.read(120_000).decode("utf-8", errors="replace")) found["urls"].append(src) except Exception: continue blobs += blobs_extra blob = "\n".join(blobs) # method-like identifiers: quoted strings ending in common RPC verbs cands = _re.findall(r"[\"'`](get|set|add|del|edit|save|query|load|login|logout)[A-Za-z0-9_]*[\"'`]", blob) cands += _re.findall(r"method\s*[:=]\s*[\"']([^\"']{2,64})[\"']", blob) cands += _re.findall(r"[\"'`](get[A-Z][A-Za-z0-9]{2,40}|get[A-Za-z0-9]{4,40})[\"'`]", blob) seen: list[str] = [] for c in cands: c = c.strip('"\'`') if c not in seen: seen.append(c) found["methods"] = seen[:120] found["blob_len"] = len(blob) found["ok"] = True except Exception as exc: return {"ok": False, "error": str(exc)[:200]} return found def _reyee_login_raw(host: str) -> tuple[dict[str, Any], str, str]: """Login and return the RAW JSON response body + sid + token, so diagnostics can see exactly which token field this firmware returns (newer ReyeeOS 2.4xx returns a `token` used as the stok for API calls; older builds only had `sid`).""" creds = _reyee_load_creds() if not creds: return {}, "no credentials configured" pw_enc = _reyee_encrypt_pw(creds["password"]) trials = [(pw_enc, True), (creds["password"], False)] last_err = "login failed" for pw, encry in trials: payload = { "method": "login", "params": { "username": creds["username"], "password": pw, "time": str(int(time.time())), "encry": encry, "limit": False, }, "id": None, } try: req = urllib.request.Request( f"http://{host}{REYEE_LOGIN_PATH}", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", "User-Agent": "AtlasCyberProtect/infra-read-only"}, method="POST", ) with _urlopen_any(req, timeout=REYEE_TIMEOUT) as resp: raw = resp.read(200_000) body = raw.decode("utf-8", errors="replace") data = json.loads(body) if body.strip() else None except Exception as exc: last_err = str(exc)[:160] continue if isinstance(data, dict): inner = data.get("data") sid = token = None if isinstance(inner, dict): sid = ( inner.get("sid") or inner.get("stok") or inner.get("session") or inner.get("sessionid") or inner.get("token") or inner.get("auth") or inner.get("cookie") ) token = inner.get("token") or inner.get("stok") or inner.get("sessionid") if not sid and isinstance(inner.get("data"), dict): sid = inner["data"].get("sid") or inner["data"].get("stok") or inner["data"].get("token") token = inner["data"].get("token") or inner["data"].get("stok") if not sid: sid = data.get("sid") or data.get("stok") or data.get("session") or data.get("sessionid") or data.get("token") or data.get("auth") token = data.get("token") or data.get("stok") return {"body": body[:4000], "data": data}, (str(sid)[:128] if sid else ""), (str(token)[:128] if token else "") last_err = f"login non-JSON response: {body[:160]}" return {"error": last_err}, "", "" def reyee_diag(host: str = "") -> dict[str, Any]: """Diagnose the ReyeeOS RPC surface: log in, then dump the RAW reply for every candidate method so Atlas can map the real names on this firmware. Tries both API shapes: the legacy `?auth=` query form and the newer `/cgi-bin/luci/;stok=/api/...` path form. Read-only - never mutates router config. Response bodies are truncated to keep the payload small. """ host = host or default_gateway_ip() if not host: return {"ok": False, "error": "no gateway ip"} raw_login, sid, token = _reyee_login_raw(host) out: dict[str, Any] = { "ok": True, "gateway_ip": host, "login": raw_login, "methods": {}, } if not sid and not token: out["ok"] = False out["error"] = raw_login.get("error") or "login failed (no sid/token in response)" return out out["connected"] = True # Newer ReyeeOS uses `token` as the API stok; older builds use `sid`. auth = token or sid out["sid_prefix"] = str(auth)[:8] out["auth_kind"] = "token" if token else "sid" # Candidate methods per concern (kept short so the diagnostic runs fast; # the box heartbeat loop executes commands synchronously). Each method is # tried via the shapes this firmware family uses: the ;stok= path form # first (ReyeeOS 2.4xx / LuCI-style), then the legacy ?auth= query form. probed: dict[str, list[str]] = { "status": ["getSystemStatus", "getStatus", "getSysInfo"], "wan": ["getWanStatus", "getWanInfo", "getWanList"], "traffic": ["getTraffic", "getRealTimeTraffic", "getFlow"], "clients": ["getClientList", "getDeviceList", "getHostList"], "wifi": ["getWifiStatus", "getWirelessStatus", "getWifiInfo"], "link": ["getLinkStatus", "getLinkInfo", "getUplinkStatus"], "sys": ["getSysInfo", "getSystem", "getRouterInfo"], "wan2": ["getWanInfoV2", "getWanStatusV2", "getInterfaceStatus"], "clients2": ["getLanClientList", "getUserList", "getOnlineUsers"], } for concern, methods in probed.items(): for m in methods: variants = [ ("stok_p", f"/cgi-bin/luci/;stok={auth}/api"), ("auth_q", f"{REYEE_API_BASE}?auth={auth}"), ] for vname, api_url in variants: r = _reyee_rpc(host, sid, m, api_url=api_url, timeout=4.0) raw = json.dumps(r, default=str)[:220] key = f"{concern}:{m}:{vname}" out["methods"][key] = {"code": r.get("code"), "raw": raw} if r.get("code") == 0: out["methods"][key]["works"] = True out.setdefault("working", {})[concern] = {"method": m, "shape": vname} break if "working" in out and concern in out["working"]: break return out def reyee_public() -> dict[str, Any]: """Public (unauthenticated) read-only snapshot for the portal/dashboard.""" cache = _load_json(REYEE_CACHE_FILE, {}) if isinstance(cache, dict) and cache.get("ok") and cache.get("summary"): return {"ok": True, **cache.get("summary", {})} if isinstance(cache, dict) and cache.get("error"): # Surface the real failure so the dashboards can show *why* the # gateway link is unavailable instead of a blank card. return { "ok": False, "error": str(cache.get("error"))[:240], "gateway_ip": str(cache.get("gateway_ip") or ""), "fetched_at": cache.get("fetched_at") or 0, } return {"ok": False, "error": "gateway not scanned yet", "reason": "no_cache"} def reyee_wan_counters() -> dict[str, int]: """Cumulative WAN byte counters from the last gateway scan (read-only). Returns {"down_bytes": int, "up_bytes": int} when the router reported lifetime rx/tx counters, else {} (caller falls back to box-observed internet traffic). """ try: cache = _load_json(REYEE_CACHE_FILE, {}) if not isinstance(cache, dict) or not cache.get("summary"): return {} tr = cache.get("summary", {}).get("traffic") or {} down = int(tr.get("down_total") or 0) up = int(tr.get("up_total") or 0) if down or up: return {"down_bytes": down, "up_bytes": up} except Exception: pass return {} def save_reyee_login(*, username: str, password: str, notes: str = "") -> dict[str, Any]: """Save the Ruijie gateway login (read-only use).""" if not (username or "").strip() or not password: raise ValueError("username and password required") _save_json( REYEE_CREDS_FILE, { "username": username.strip()[:120], "password": password, "notes": (notes or "").strip()[:400], "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), }, ) return reyee_public_status() PUBLIC_IP_FILE = DATA / "public-ip.json" PUBLIC_IP_TTL = 5 * 86400 # re-check every 5 days unless forced def public_ip_lookup(*, force: bool = False) -> dict[str, Any]: """Look up this box's public IP + ISP from the web (read-only, cached). The router RPC may not expose WAN details on every firmware, so we ask a public service (ip-api.com) what the internet line actually is. Cached for 5 days; `force` re-checks immediately (the dashboard's manual refresh button sends a refresh_public_ip command). """ cache = _load_json(PUBLIC_IP_FILE, {}) now = time.time() if ( not force and isinstance(cache, dict) and cache.get("ip") and (now - float(cache.get("checked_at") or 0)) < PUBLIC_IP_TTL ): return cache out: dict[str, Any] = {"ok": False, "error": "lookup failed"} for url in ( "http://ip-api.com/json/?fields=status,query,isp,org,as,country,city,regionName,timezone", "https://ipapi.co/json/", ): try: req = urllib.request.Request(url, headers={"User-Agent": "AtlasCyberProtect/box"}) with _urlopen_any(req, timeout=8) as resp: body = json.loads(resp.read(20_000).decode("utf-8", errors="replace")) if url.startswith("http://ip-api.com"): if body.get("status") != "success" or not body.get("query"): continue out = { "ok": True, "ip": str(body["query"]), "isp": str(body.get("isp") or ""), "org": str(body.get("org") or ""), "as": str(body.get("as") or ""), "country": str(body.get("country") or ""), "city": str(body.get("city") or ""), "region": str(body.get("regionName") or ""), "checked_at": now, } else: if not body.get("ip"): continue out = { "ok": True, "ip": str(body["ip"]), "isp": str(body.get("org") or ""), "org": str(body.get("org") or ""), "as": "", "country": str(body.get("country_name") or ""), "city": str(body.get("city") or ""), "region": str(body.get("region") or ""), "checked_at": now, } break except Exception as exc: out["error"] = str(exc)[:160] continue _save_json(PUBLIC_IP_FILE, out) return out def public_ip_cached() -> dict[str, Any]: """Return the cached public-IP info (never does a web call).""" cache = _load_json(PUBLIC_IP_FILE, {}) if isinstance(cache, dict) and cache.get("ip"): return cache return {} def reyee_public_status() -> dict[str, Any]: creds = _reyee_load_creds() return { "ok": True, "mode": "read_only", "mutate_requires": OWNER_EMAIL, "configured": bool(creds), "gateway_ip": default_gateway_ip(), "policy": "Ruijie gateway login stays on this Protect box. Atlas only READS WAN status, traffic and clients.", } # —— Backward-compatible aliases used by older gateway_* portal routes —— def load_creds() -> dict[str, str]: st = _load_store()["default"] return { "username": st.get("username") or "", "password": st.get("password") or "", "gateway_ip": default_gateway_ip(), "notes": st.get("notes") or "", "updated_at": "", } def save_creds(**kwargs: Any) -> dict[str, Any]: return save_default_login( username=str(kwargs.get("username") or ""), password=str(kwargs.get("password") or ""), notes=str(kwargs.get("notes") or ""), keep_password=bool(kwargs.get("keep_password")), ) def clear_creds() -> dict[str, Any]: return clear_all_logins() def read_gateway(*, force: bool = False) -> dict[str, Any]: inv = scan_infrastructure(force=force) gw_ip = default_gateway_ip() match = next((i for i in inv.get("items") or [] if i.get("ip") == gw_ip), None) if not match: match = next((i for i in inv.get("items") or [] if i.get("role") == "gateway"), None) tips = list((match or {}).get("tips") or []) tips.insert( 0, { "level": "info", "title": "Full LAN gear scan", "detail": ( f"Found {inv.get('summary', {}).get('total', 0)} network devices " f"(GW {inv.get('summary', {}).get('gateways', 0)} · " f"SW {inv.get('summary', {}).get('switches', 0)} · " f"AP {inv.get('summary', {}).get('aps', 0)})." ), }, ) return { "ok": True, "mode": "read_only", "mutate_requires": OWNER_EMAIL, "gateway_ip": gw_ip, "brand": (match or {}).get("brand") or "unknown", "open_ports": (match or {}).get("open_ports") or [], "login_ok": bool((match or {}).get("login_ok")), "credentials_configured": inv.get("credentials_configured"), "tips": tips, "info": (match or {}).get("info") or {}, "inventory": inv, "checked_at": inv.get("checked_at"), "policy": inv.get("policy"), }