""" Firmware download helper. Calls the UBox cloud `check_version` endpoint, extracts every URL it sees in the response, downloads each one to ~/dumps/javiscam_fw/, and reports sizes and sha256 hashes. Real code — no stubs, no placeholders. """ import hashlib import json import os import re import socket import urllib.request import urllib.error from datetime import datetime from utils.log import log, C_INFO, C_SUCCESS, C_ERROR, C_TRAFFIC, C_IMPORTANT from api import ubox_client FW_DIR = os.path.expanduser("~/dumps/javiscam_fw") URL_RE = re.compile(rb'https?://[^\s"\']+') def _walk_strings(obj): """Yield every string value in a nested dict/list.""" if isinstance(obj, str): yield obj elif isinstance(obj, dict): for v in obj.values(): yield from _walk_strings(v) elif isinstance(obj, list): for v in obj: yield from _walk_strings(v) def _extract_urls(obj): urls = set() for s in _walk_strings(obj): for m in URL_RE.findall(s.encode("utf-8")): urls.add(m.decode("utf-8", errors="replace")) return sorted(urls) def _safe_filename(url): name = url.split("?")[0].rsplit("/", 1)[-1] or "firmware.bin" name = re.sub(r"[^A-Za-z0-9._-]", "_", name) if not name: name = "firmware.bin" return name def _download(url, dest): req = urllib.request.Request(url, headers={ "User-Agent": "okhttp/4.9.1", "X-UbiaAPI-CallContext": "source=app&app=ubox&ver=1.1.360&osver=14", }) with urllib.request.urlopen(req, timeout=60) as resp: total = 0 h = hashlib.sha256() with open(dest, "wb") as f: while True: chunk = resp.read(64 * 1024) if not chunk: break f.write(chunk) h.update(chunk) total += len(chunk) return total, h.hexdigest(), resp.headers.get("Content-Type", "?") def check_and_download(cfg, host_version=None): """ Returns dict: { "ok": bool, "url_count": int, "downloads": [{"url","path","bytes","sha256","content_type"}], "raw": , "error": optional str, } """ if not cfg.get("api_token"): log("FW: not logged in", C_ERROR) return {"ok": False, "error": "not logged in"} if not cfg.get("device_uid"): log("FW: no device_uid — call /devices first", C_ERROR) return {"ok": False, "error": "no device_uid"} os.makedirs(FW_DIR, exist_ok=True) # Try several version formats — the cloud needs a string the device-side # comparator considers "older than current" to trigger a download URL. version_attempts = [host_version] if host_version else [ "2604.0.0.1", # same model prefix, ancient "1.0.0.0", "0.0.0.1", "2604.0.29.7", # one below known shipped version "0", "", ] result = None used_version = None attempts_log = [] # list of (version, status, summary) best_success = None # remember the best msg=success response even without URLs for v in version_attempts: log(f"FW: trying check_version with host_version={v!r}", C_INFO) candidate = ubox_client.api_post( cfg["api_base"], "user/qry/device/check_version/v3", { "device_uid": cfg["device_uid"], "host_version": v, "wifi_version": v, "is_lite": False, "zone_id": 2, }, cfg["api_token"], ) summary = json.dumps(candidate)[:200] if isinstance(candidate, dict) else repr(candidate)[:200] attempts_log.append({"version": v, "summary": summary}) if not isinstance(candidate, dict): continue urls_found = _extract_urls(candidate) if urls_found: log(f"FW: got {len(urls_found)} URL(s) using version={v!r}", C_SUCCESS) result = candidate used_version = v break if candidate.get("msg") == "success" and best_success is None: best_success = (v, candidate) log(f"FW: no URLs at version={v!r}, response={summary}", C_INFO) if result is None: if best_success is not None: used_version, result = best_success log(f"FW: keeping best success response from version={used_version!r}", C_INFO) else: result = candidate if isinstance(candidate, dict) else {"error": "all versions failed"} if not isinstance(result, dict): log(f"FW: bad response: {result!r}", C_ERROR) return {"ok": False, "error": "bad response", "raw": result} # Save the raw response next to firmware files for the record ts = datetime.now().strftime("%Y%m%d_%H%M%S") meta_path = os.path.join(FW_DIR, f"check_version_{ts}.json") try: with open(meta_path, "w") as f: json.dump(result, f, indent=2, ensure_ascii=False) except OSError as e: log(f"FW: cannot save meta: {e}", C_ERROR) urls = _extract_urls(result) log(f"FW: response saved -> {meta_path}", C_INFO) log(f"FW: found {len(urls)} URL(s) in response", C_INFO) for u in urls: log(f" • {u}", C_TRAFFIC) downloads = [] for url in urls: # Only attempt downloads of plausible binary blobs lower = url.lower() if not any(lower.endswith(ext) or ext in lower for ext in (".bin", ".rar", ".zip", ".tar", ".gz", ".img", ".pkg", "/firmware", "/ota", "ubiaota", "update")): log(f" skip (not firmware-looking): {url}", C_INFO) continue fname = _safe_filename(url) dest = os.path.join(FW_DIR, fname) try: log(f"FW: downloading {url}", C_INFO) n, sha, ct = _download(url, dest) log(f"FW: ✓ {fname} {n}B sha256={sha[:16]}… ct={ct}", C_IMPORTANT if n > 1024 else C_TRAFFIC) downloads.append({ "url": url, "path": dest, "bytes": n, "sha256": sha, "content_type": ct, }) except urllib.error.HTTPError as e: log(f"FW: HTTP {e.code} on {url}", C_ERROR) except urllib.error.URLError as e: log(f"FW: URL error: {e.reason} on {url}", C_ERROR) except (socket.timeout, OSError) as e: log(f"FW: download failed: {e}", C_ERROR) if not downloads: log("FW: nothing downloaded — check_version returned no firmware URLs", C_ERROR) return { "ok": bool(downloads), "used_version": used_version, "url_count": len(urls), "downloads": downloads, "attempts": attempts_log, "raw": result, }