From 7931ffc9986f7d9debdc635e305a63cd1989a690 Mon Sep 17 00:00:00 2001 From: s0kil Date: Mon, 17 Aug 2026 20:21:27 -0400 Subject: [PATCH 1/3] feat(drivers): add ARRIS SURFboard SB8200 (CBN firmware) support SB8200 units built by Compal Broadband Networks serve a CBN web UI rather than the HNAP1 interface the other SURFboard models expose, so /HNAP1/ returns 404 and the existing surfboard driver cannot drive them. Add a sb8200_cbn driver and a sb8200_cbn_xml format profile for that transport. Channel tables are read from /xml/getter.xml with numeric function codes (10 SC-QAM downstream, 11 upstream, 9 OFDM, 6 OFDMA, 19 codewords, 2 system info). Login reproduces the firmware's CBN_Encrypt envelope: username and password are AES-256-CBC encrypted with a key and IV derived from the rotating sessionToken cookie, then wrapped as HS::. Notes on the profile: - Codeword counts live in their own table keyed by dsid, which matches the SC-QAM chid, so the downstream lane joins the two. - The downstream table omits the symbol rate and the modem is an Annex B 6 MHz device, so Annex B rates are injected. Without them the analyzer falls back to the EuroDOCSIS 8 MHz default and every downstream capacity estimate reads ~30% high. - The OFDM codeword counters this firmware reports are not comparable to the SC-QAM ones. Measured on a locked 4096QAM channel at 33 dB MER they climb by ~1.3 million uncorrectables per minute while the whole SC-QAM cohort adds 0-1, uncorrectables exceed correctables, and the modem's own codeword table reports zero for the same dsid. Reported as measured codewords they pinned downstream health at critical on every poll, so the lane is left counter-unsupported instead. - OFDMA is reported as unsupported rather than guessed; no captured payload from this firmware has ever contained an active OFDMA row. - The serial number exposed by the system info table is deliberately not surfaced in device info. The firmware allows a single Web-UI session and rate-limits repeated login attempts, so the session is held across polls, re-authenticated at most once per expiry, and released on shutdown. --- ARCHITECTURE.md | 3 +- README.md | 5 +- app/drivers/__init__.py | 3 + app/drivers/formats/__init__.py | 1 + app/drivers/formats/xml_payloads.py | 286 ++++++++++++- app/drivers/sb8200_cbn.py | 272 +++++++++++++ .../test_driver_formats_contract_red.py | 3 + .../test_driver_formats_static.py | 5 +- tests/drivers/driver_format_cases.py | 29 ++ .../test_driver_format_characterization.py | 4 +- tests/test_sb8200_cbn_driver.py | 383 ++++++++++++++++++ 11 files changed, 984 insertions(+), 10 deletions(-) create mode 100644 app/drivers/sb8200_cbn.py create mode 100644 tests/test_sb8200_cbn_driver.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f21b80ee..5c338817 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -490,7 +490,7 @@ class ModemDriver(ABC): Every registered concrete class exposes a non-empty immutable `FORMAT_FAMILIES` tuple. Registry aliases appear together in the first column; -there are 21 keys, 20 concrete classes, and 22 explicit profiles. +there are 22 keys, 21 concrete classes, and 23 explicit profiles. | Registry key(s) | Concrete class | Format profile(s) | Cohesive module / entrypoint | |---|---|---|---| @@ -509,6 +509,7 @@ there are 21 keys, 20 concrete classes, and 22 explicit profiles. | `sb6141` | `SB6141Driver` | `sb6141_transposed_html` | `html_transposed.parse_sb6141_transposed_html` | | `sb6183` | `SB6183Driver` | `sb6183_html` | `html_rows.parse_sb6183_html` | | `sb6190` | `SB6190Driver` | `sb6190_html` | `html_rows.parse_sb6190_html` | +| `sb8200_cbn` | `SB8200CBNDriver` | `sb8200_cbn_xml` | `xml_payloads.parse_sb8200_cbn_xml` | | `sercom_dm1000` | `SercomDM1000Driver` | `sercom_dm1000_json` | `sercom.parse_sercom_dm1000_json` | | `surfboard` | `SurfboardDriver` | `arris_html`, `surfboard_hnap` | `html_rows.parse_arris_html`; `surfboard.parse_surfboard_hnap` | | `tc4400` | `TC4400Driver` | `tc4400_html` | `html_rows.parse_tc4400_html` | diff --git a/README.md b/README.md index 0d1af25f..b919e008 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@

- Self-hosted • Local data • Demo • Reports • 20 modem families • MIT + Self-hosted • Local data • Demo • Reports • 21 modem families • MIT

@@ -281,7 +281,7 @@ More views from the product: ## Supported Hardware -DOCSight supports **20 modem families** out of the box and also offers **Generic Router mode** for fiber, DSL, and satellite connections. +DOCSight supports **21 modem families** out of the box and also offers **Generic Router mode** for fiber, DSL, and satellite connections. ### Common setups @@ -293,6 +293,7 @@ DOCSight supports **20 modem families** out of the box and also offers **Generic - **Sagemcom F3896LG** (Hub 5 / Liberty Global REST firmware): unauthenticated API, works in modem mode - **Technicolor TC4400** - **Arris SURFboard** (S33, S34, SB8200): HNAP1 API +- **Arris SURFboard SB8200** (CBN firmware, `SB8200v3`): XML API, for units that serve the CBN web UI instead of HNAP1 - **Arris SB6183:** HTTP status pages, no authentication required - **Hitron CODA-56 and CODA-4680** - **Netgear CM3000** diff --git a/app/drivers/__init__.py b/app/drivers/__init__.py index 89ac4be1..c9979d18 100644 --- a/app/drivers/__init__.py +++ b/app/drivers/__init__.py @@ -36,6 +36,9 @@ driver_registry.register_builtin("sb6190", "app.drivers.sb6190.SB6190Driver", "Arris SB6190", hints={"default_url": "https://192.168.100.1", "default_user": "admin"}) +driver_registry.register_builtin("sb8200_cbn", "app.drivers.sb8200_cbn.SB8200CBNDriver", + "Arris SURFboard SB8200 (CBN firmware)", + hints={"default_url": "https://192.168.100.1", "default_user": "admin"}) driver_registry.register_builtin("cm8200", "app.drivers.cm8200.CM8200Driver", "Arris Touchstone CM8200A", hints={"default_url": "https://192.168.100.1", "default_user": "admin"}) diff --git a/app/drivers/formats/__init__.py b/app/drivers/formats/__init__.py index 9e8cda79..00b4385e 100644 --- a/app/drivers/formats/__init__.py +++ b/app/drivers/formats/__init__.py @@ -24,6 +24,7 @@ "sb6141_transposed_html": "app.drivers.formats.html_transposed", "sb6183_html": "app.drivers.formats.html_rows", "sb6190_html": "app.drivers.formats.html_rows", + "sb8200_cbn_xml": "app.drivers.formats.xml_payloads", "sercom_dm1000_json": "app.drivers.formats.sercom", "surfboard_hnap": "app.drivers.formats.surfboard", "tc4400_html": "app.drivers.formats.html_rows", diff --git a/app/drivers/formats/xml_payloads.py b/app/drivers/formats/xml_payloads.py index bb8c36bc..7a738117 100644 --- a/app/drivers/formats/xml_payloads.py +++ b/app/drivers/formats/xml_payloads.py @@ -1,12 +1,21 @@ -"""Pure parser for the Compal CH7465 downstream/upstream XML profile.""" +"""Pure parsers for the XML channel payloads served by CBN-built modems.""" from __future__ import annotations import xml.etree.ElementTree as ET +from types import MappingProxyType -from ...types import RawChannel -from .contract import ParseDiagnostic, ParseResult, diagnostic -from .primitives import normalize_modulation +from ...types import DocsisDataFritz, RawChannel +from .contract import ParseDiagnostic, ParseResult, diagnostic, docsis_split +from .primitives import hz_to_mhz, normalize_modulation + +_XML_FAMILY = "xml_payloads" +_SB8200_PROFILE = "sb8200_cbn_xml" + +# The SB8200 is an Annex B (6 MHz) device and its downstream table omits the +# symbol rate. Without these the analyzer falls back to the EuroDOCSIS 8 MHz +# default and every downstream capacity estimate reads ~30% high. +_SB8200_ANNEX_B_DS_SYMBOL_RATES = MappingProxyType({"64QAM": 5057, "256QAM": 5361}) def _text(node: ET.Element | None, default: str = "") -> str: @@ -86,3 +95,272 @@ def parse_ch7465_xml( {"docsis": "3.0", "downstream": downstream, "upstream": upstream}, tuple(diagnostics), ) + + +def _optional_int(value: str) -> int | None: + """Parse an optional counter, keeping a missing value distinct from zero.""" + try: + return int(value.strip()) + except (AttributeError, ValueError): + return None + + +def _optional_float(value: str) -> float | None: + """Parse an optional measurement, keeping a missing value distinct from zero.""" + try: + return float(value.strip()) + except (AttributeError, ValueError): + return None + + +def _sb8200_root(payload: str | None) -> ET.Element | None: + """Return the parsed table root, or None for an absent or malformed table.""" + if not payload: + return None + try: + return ET.fromstring(payload) + except ET.ParseError: + return None + + +def _sb8200_issue( + code: str, + *, + direction: str | None = None, + index: int | None = None, + field: str | None = None, +) -> ParseDiagnostic: + return diagnostic( + _SB8200_PROFILE, code, family=_XML_FAMILY, + direction=direction, index=index, field=field, + ) + + +def _sb8200_error_counters( + root: ET.Element, +) -> tuple[dict[int, tuple[int | None, int | None]], list[ParseDiagnostic]]: + """Index the separate codeword table that both downstream lanes join on. + + The SB8200 reports codeword counts in their own table keyed by ``dsid``, + which matches the SC-QAM ``chid`` and the OFDM ``dsid``. + """ + counters: dict[int, tuple[int | None, int | None]] = {} + diagnostics: list[ParseDiagnostic] = [] + for index, entry in enumerate(root.findall("signal")): + dsid = _optional_int(_text(entry.find("dsid"))) + if dsid is None: + diagnostics.append(_sb8200_issue( + "invalid_row", direction="downstream", index=index, field="dsid", + )) + continue + counters[dsid] = ( + _optional_int(_text(entry.find("correctable"))), + _optional_int(_text(entry.find("uncorrectable"))), + ) + return counters, diagnostics + + +def _sb8200_locked(channel: ET.Element, field: str) -> bool: + """Treat only an explicit unlocked marker as unlocked.""" + return _text(channel.find(field)).strip() in ("", "1") + + +def _sb8200_ofdm_locked(channel: ET.Element) -> bool: + """Report OFDM lock from the explicit flags, falling back to the PLC state.""" + active = _text(channel.find("ofdmIsActive")).strip() + if active and active != "1": + return False + locked = _text(channel.find("ofdmIsLocked")).strip() + if locked: + return locked == "1" + return _text(channel.find("PLCLocked")).strip().upper() == "YES" + + +def _sb8200_downstream_scqam( + root: ET.Element, + counters: dict[int, tuple[int | None, int | None]], +) -> tuple[list[RawChannel], list[ParseDiagnostic]]: + channels: list[RawChannel] = [] + diagnostics: list[ParseDiagnostic] = [] + for index, entry in enumerate(root.findall("downstream")): + if not _sb8200_locked(entry, "IsLocked"): + continue + channel_id = _optional_int(_text(entry.find("chid"))) + power = _optional_float(_text(entry.find("pow"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="downstream", index=index, + )) + continue + channel: RawChannel = { + "channelID": channel_id, + "frequency": hz_to_mhz(_text(entry.find("freq"))), + "powerLevel": power, + } + snr = _optional_float(_text(entry.find("snr"))) + if snr is not None: + channel["mer"] = snr + channel["mse"] = -snr + modulation = normalize_modulation(_text(entry.find("mod"))) + if modulation: + channel["modulation"] = modulation + symbol_rate = _SB8200_ANNEX_B_DS_SYMBOL_RATES.get(modulation) + if symbol_rate is not None: + channel["symbolRate"] = symbol_rate + corrected, uncorrected = counters.get(channel_id, (None, None)) + if corrected is not None: + channel["corrErrors"] = corrected + if uncorrected is not None: + channel["nonCorrErrors"] = uncorrected + channels.append(channel) + return channels, diagnostics + + +def _sb8200_downstream_ofdm( + root: ET.Element, +) -> tuple[list[RawChannel], list[ParseDiagnostic]]: + channels: list[RawChannel] = [] + diagnostics: list[ParseDiagnostic] = [] + for index, entry in enumerate(root.findall("downstream")): + if not _sb8200_ofdm_locked(entry): + continue + channel_id = _optional_int(_text(entry.find("dsid"))) + power = _optional_float(_text(entry.find("PLCPower"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="downstream", index=index, + )) + continue + channel: RawChannel = { + "channelID": channel_id, + "type": "OFDM", + "frequency": hz_to_mhz(_text(entry.find("Subcarr0Frequency"))), + "powerLevel": power, + "mse": None, + } + mer = _optional_float(_text(entry.find("DataScAvgMer"))) + if mer is not None: + channel["mer"] = mer + modulation = normalize_modulation(_text(entry.find("ofdmModulation"))) + if modulation: + channel["modulation"] = modulation + # The OFDM codeword counters this firmware reports are not comparable + # to the SC-QAM ones. Measured on a locked 4096QAM channel at 33 dB + # MER they climb by roughly 1.3 million uncorrectables per minute + # while the entire SC-QAM cohort adds 0-1, uncorrectables exceed + # correctables, and the modem's own codeword table reports zero for + # the same dsid. Reporting them as measured codewords pins downstream + # health at critical, so the lane is left counter-unsupported instead. + if _text(entry.find("ofdmCorrected")) or _text(entry.find("ofdmUncorrectable")): + diagnostics.append(_sb8200_issue( + "unsupported_counters", direction="downstream", index=index, + field="ofdm_codewords", + )) + channels.append(channel) + return channels, diagnostics + + +def _sb8200_upstream_scqam( + root: ET.Element, +) -> tuple[list[RawChannel], list[ParseDiagnostic]]: + channels: list[RawChannel] = [] + diagnostics: list[ParseDiagnostic] = [] + for index, entry in enumerate(root.findall("upstream")): + if not _sb8200_locked(entry, "usLocked"): + continue + channel_id = _optional_int(_text(entry.find("usid"))) + power = _optional_float(_text(entry.find("power"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="upstream", index=index, + )) + continue + channel: RawChannel = { + "channelID": channel_id, + "frequency": hz_to_mhz(_text(entry.find("freq"))), + "powerLevel": power, + } + modulation = normalize_modulation(_text(entry.find("mod"))) + if modulation: + channel["modulation"] = modulation + multiplex = _text(entry.find("channeltype")).strip().upper() + if multiplex: + channel["type"] = multiplex + channel["multiplex"] = multiplex + symbol_rate = _optional_float(_text(entry.find("srate"))) + if symbol_rate is not None: + channel["symbolRate"] = int(round(symbol_rate * 1000)) + channels.append(channel) + return channels, diagnostics + + +def _sb8200_ofdma_present(root: ET.Element) -> bool: + """Report whether the modem claims an active OFDMA upstream lane.""" + if root.findall("upstream"): + return True + return bool(_optional_int(_text(root.find("us_num")))) + + +def parse_sb8200_cbn_xml( + downstream_xml: str | None, + upstream_xml: str | None, + downstream_ofdm_xml: str | None, + upstream_ofdma_xml: str | None, + signal_xml: str | None, +) -> ParseResult[DocsisDataFritz | None]: + """Normalize the five CBN XML tables an SB8200 serves for DOCSIS status. + + The downstream and upstream SC-QAM tables are required. The OFDM, OFDMA, + and codeword tables enrich the result and degrade to a diagnostic so a + partially reachable modem still reports the channels it did return. + """ + downstream_root = _sb8200_root(downstream_xml) + upstream_root = _sb8200_root(upstream_xml) + if downstream_root is None or upstream_root is None: + return ParseResult(None, (_sb8200_issue("invalid_xml"),)) + + diagnostics: list[ParseDiagnostic] = [] + + counters: dict[int, tuple[int | None, int | None]] = {} + signal_root = _sb8200_root(signal_xml) + if signal_root is None: + if signal_xml: + diagnostics.append(_sb8200_issue( + "invalid_xml", direction="downstream", field="signal_table", + )) + else: + counters, counter_issues = _sb8200_error_counters(signal_root) + diagnostics.extend(counter_issues) + + ds30, ds30_issues = _sb8200_downstream_scqam(downstream_root, counters) + diagnostics.extend(ds30_issues) + + ds31: list[RawChannel] = [] + ofdm_root = _sb8200_root(downstream_ofdm_xml) + if ofdm_root is None: + if downstream_ofdm_xml: + diagnostics.append(_sb8200_issue( + "invalid_xml", direction="downstream", field="ofdm_table", + )) + else: + ds31, ds31_issues = _sb8200_downstream_ofdm(ofdm_root) + diagnostics.extend(ds31_issues) + + us30, us30_issues = _sb8200_upstream_scqam(upstream_root) + diagnostics.extend(us30_issues) + + # The observed firmware reports us_num=0 and no OFDMA rows. The lane is + # reported as unsupported rather than guessed from field names that no + # captured payload has ever shown. + ofdma_root = _sb8200_root(upstream_ofdma_xml) + if ofdma_root is None: + if upstream_ofdma_xml: + diagnostics.append(_sb8200_issue( + "invalid_xml", direction="upstream", field="ofdma_table", + )) + elif _sb8200_ofdma_present(ofdma_root): + diagnostics.append(_sb8200_issue( + "unsupported_lane", direction="upstream", field="ofdma_table", + )) + + return ParseResult(docsis_split(ds30, ds31, us30, []), tuple(diagnostics)) diff --git a/app/drivers/sb8200_cbn.py b/app/drivers/sb8200_cbn.py new file mode 100644 index 00000000..8b2172b2 --- /dev/null +++ b/app/drivers/sb8200_cbn.py @@ -0,0 +1,272 @@ +"""ARRIS SURFboard SB8200 (CBN firmware) driver for DOCSight. + +The SB8200 units built by Compal Broadband Networks serve a CBN web UI instead +of the HNAP1 interface used by the other SURFboard models, so `/HNAP1/` returns +404 and the `surfboard` driver cannot drive them. Status tables are fetched +from `/xml/getter.xml` with numeric function codes and normalized by the +`sb8200_cbn_xml` profile. + +Login mirrors the firmware's `CBN_Encrypt()` helper: username and password are +each AES-256-CBC encrypted with a key and IV derived from the rotating +`sessionToken` cookie, then wrapped in a `HS::` envelope. +""" + +from __future__ import annotations + +import base64 +import hashlib +import logging +import re +import weakref +import xml.etree.ElementTree as ET +from enum import Enum + +import requests +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +from .base import ModemDriver +from .formats.xml_payloads import parse_sb8200_cbn_xml +from ..types import ConnectionInfo, DeviceInfo, DocsisData + +log = logging.getLogger("docsis.driver.sb8200_cbn") + +# The status tables top out around 9 KB. Bound the response so a hostile or +# broken endpoint cannot feed an unbounded document to the XML parser. +MAX_RESPONSE_BYTES = 1_048_576 + +_UPTIME_RE = re.compile(r"(\d+)\s*day\(s\)\s*(\d+)h:(\d+)m:(\d+)s") + + +class Query(Enum): + """Getter function codes usable with `_get_data()`.""" + + GLOBAL_SETTINGS = 1 + SYSTEM_INFO = 2 + UPSTREAM_OFDMA_TABLE = 6 + DOWNSTREAM_OFDM_TABLE = 9 + DOWNSTREAM_TABLE = 10 + UPSTREAM_TABLE = 11 + SIGNAL_TABLE = 19 + + +class Action(Enum): + """Setter function codes usable with `_set_data()`.""" + + LOGIN = 15 + LOGOUT = 16 + + +def _node_text(node: ET.Element | None, default: str = "") -> str: + """Extract one element's text, falling back to `default` when absent.""" + if node is not None and node.text is not None: + return node.text.strip() + return default + + +class SB8200CBNDriver(ModemDriver): + """Driver for ARRIS SURFboard SB8200 modems running CBN firmware. + + The firmware allows one Web-UI session at a time and rotates its CSRF + token on every response, so the session is held open across polls and + released on shutdown. + """ + + FORMAT_FAMILIES = ("sb8200_cbn_xml",) + + def __init__(self, url: str, user: str, password: str): + if url.startswith("http://"): + url = "https://" + url[len("http://"):] + log.info("SB8200 requires HTTPS, upgraded URL to %s", url) + super().__init__(url, user, password) + session = requests.Session() + # The modem presents a self-signed per-device ARRIS certificate. + session.verify = False + session.headers["Referer"] = url.rstrip("/") + "/" + session.headers["X-Requested-With"] = "XMLHttpRequest" + # The session is locked to one IP plus User-Agent, so this must stay + # stable for the lifetime of the login. + session.headers["User-Agent"] = "docsight/2" + self._session: requests.Session = session + self._logged_in = False + self._hw_model: str | None = None + # Release the single Web-UI session so the operator is not locked out + # of their own modem when DOCSight exits. + self._finalizer = weakref.finalize(self, SB8200CBNDriver._cleanup, url, session) + self._finalizer.atexit = True + + @staticmethod + def _cleanup(url: str, session: requests.Session) -> None: + """Close the Web-UI session so another client can connect.""" + if "SID" not in session.cookies: + return + try: + session.post( + f"{url}/xml/setter.xml", + data={ + "token": session.cookies.get("sessionToken", ""), + "fun": str(Action.LOGOUT.value), + }, + timeout=10, + ) + except requests.RequestException: + pass + session.cookies.pop("SID", None) + + def login(self) -> None: + """Authenticate with the modem, reusing an already-open session. + + The firmware rate-limits repeated login attempts, so an established + session is kept rather than re-authenticated on every poll. + """ + if self._logged_in: + return + + self._session.cookies.clear() + response = self._session.get(f"{self._url}/common_page/login.html", timeout=10) + response.raise_for_status() + response.close() + + token = self._session_token() + if not token: + raise RuntimeError("Modem did not issue a session token") + + hw_model = self._hardware_model() + body = self._set_data(Action.LOGIN, { + "Username": self._encrypt(self._user or "admin", token, hw_model), + "Password": self._encrypt(self._password, token, hw_model), + }) + + if "successful" not in body or "SID=" not in body: + # The response body can echo account state, so it is not logged. + raise RuntimeError("Modem authentication failed: check username and password") + + self._session.cookies.set("SID", body.split("SID=", 1)[1].strip()) + self._logged_in = True + log.info("Auth OK (%s)", hw_model) + + def get_docsis_data(self) -> DocsisData: + """Query the SC-QAM, OFDM, OFDMA, and codeword tables.""" + parsed = parse_sb8200_cbn_xml( + self._get_data(Query.DOWNSTREAM_TABLE), + self._get_data(Query.UPSTREAM_TABLE), + self._get_data(Query.DOWNSTREAM_OFDM_TABLE), + self._get_data(Query.UPSTREAM_OFDMA_TABLE), + self._get_data(Query.SIGNAL_TABLE), + ) + if parsed.value is None: + raise ValueError("invalid SB8200 channel XML") + return parsed.value + + def get_device_info(self) -> DeviceInfo: + """Read model, firmware, and uptime metadata. + + The serial number reported by this table is deliberately not exposed. + """ + try: + root = self._xml(self._get_data(Query.SYSTEM_INFO)) + if root is None: + raise ValueError("missing system info") + info: DeviceInfo = { + "manufacturer": "ARRIS", + "model": _node_text(root.find("HwModel"), "SB8200"), + "hw_version": _node_text(root.find("cm_hardware_version")), + "sw_version": _node_text(root.find("SwVersion")), + "docsis_status": _node_text(root.find("cm_status")), + } + uptime = self._uptime_seconds(_node_text(root.find("cm_system_uptime"))) + if uptime is not None: + info["uptime_seconds"] = uptime + return info + except Exception: + return {"manufacturer": "ARRIS", "model": "SB8200", "sw_version": ""} + + def get_connection_info(self) -> ConnectionInfo: + """Report the DOCSIS mode. + + This firmware exposes no service-flow table, so provisioned rates are + left unset rather than guessed. + """ + try: + root = self._xml(self._get_data(Query.SYSTEM_INFO)) + mode = _node_text(root.find("cm_docsis_mode")) if root is not None else "" + return {"connection_type": mode} if mode else {} + except Exception as e: + log.warning("Failed to get connection info: %s", e) + return {} + + @staticmethod + def _encrypt(value: str, token: str, hw_model: str) -> str: + """Reproduce the firmware's `CBN_Encrypt()` field envelope.""" + key = hashlib.sha256(token.encode()).digest() + iv = hashlib.md5(token.encode(), usedforsecurity=False).digest() + padder = padding.PKCS7(128).padder() + padded = padder.update(value.encode()) + padder.finalize() + encryptor = Cipher(algorithms.AES(key), modes.CBC(iv)).encryptor() + ciphertext = encryptor.update(padded) + encryptor.finalize() + return base64.b64encode(f"HS:{hw_model}:{ciphertext.hex()}".encode()).decode() + + @staticmethod + def _uptime_seconds(uptime: str) -> int | None: + """Convert the `14day(s)20h:50m:40s` uptime string to seconds.""" + match = _UPTIME_RE.search(uptime) + if not match: + return None + days, hours, minutes, seconds = (int(part) for part in match.groups()) + return days * 86400 + hours * 3600 + minutes * 60 + seconds + + @staticmethod + def _xml(payload: str) -> ET.Element | None: + try: + return ET.fromstring(payload) + except ET.ParseError: + return None + + def _session_token(self) -> str: + """Read the CSRF token the modem rotates on every response.""" + return self._session.cookies.get("sessionToken", "") + + def _hardware_model(self) -> str: + """Read the hardware model that keys the login envelope.""" + if self._hw_model: + return self._hw_model + root = self._xml(self._request("getter.xml", Query.GLOBAL_SETTINGS.value)[0]) + model = _node_text(root.find("HwModel")) if root is not None else "" + if not model: + raise RuntimeError("Modem did not report a hardware model") + self._hw_model = model + return model + + def _request(self, endpoint: str, function: int, data: dict[str, str] | None = None) -> tuple[str, int]: + """Post one function call, returning its body and HTTP status.""" + payload = {"token": self._session_token(), "fun": str(function)} + if data: + payload |= data + response = self._session.post( + f"{self._url}/xml/{endpoint}", + data=payload, + timeout=10, + allow_redirects=False, + ) + response.raise_for_status() + if len(response.content) > MAX_RESPONSE_BYTES: + raise RuntimeError(f"Modem response for fun={function} exceeds the size limit") + return response.text, response.status_code + + def _get_data(self, query: Query) -> str: + """Query one table, re-authenticating once if the session expired.""" + body, status = self._request("getter.xml", query.value) + if status == 302: + log.info("Session expired, re-authenticating") + self._logged_in = False + self.login() + body, status = self._request("getter.xml", query.value) + if status == 302: + raise RuntimeError(f"Modem rejected fun={query.value} after re-authentication") + return body + + def _set_data(self, action: Action, data: dict[str, str]) -> str: + """Execute one action on the modem.""" + if "fun" in data or "token" in data: + raise ValueError("invalid data key in SB8200 command") + return self._request("setter.xml", action.value, data)[0] diff --git a/tests/architecture/test_driver_formats_contract_red.py b/tests/architecture/test_driver_formats_contract_red.py index 53713298..620d8d12 100644 --- a/tests/architecture/test_driver_formats_contract_red.py +++ b/tests/architecture/test_driver_formats_contract_red.py @@ -25,6 +25,7 @@ "sb6141", "sb6183", "sb6190", + "sb8200_cbn", "sercom_dm1000", "surfboard", "tc4400", @@ -48,6 +49,7 @@ "app.drivers.sb6141.SB6141Driver": ("sb6141_transposed_html",), "app.drivers.sb6183.SB6183Driver": ("sb6183_html",), "app.drivers.sb6190.SB6190Driver": ("sb6190_html",), + "app.drivers.sb8200_cbn.SB8200CBNDriver": ("sb8200_cbn_xml",), "app.drivers.sercom_dm1000.SercomDM1000Driver": ("sercom_dm1000_json",), "app.drivers.surfboard.SurfboardDriver": ("arris_html", "surfboard_hnap"), "app.drivers.tc4400.TC4400Driver": ("tc4400_html",), @@ -77,6 +79,7 @@ "sb6141_transposed_html": "app.drivers.formats.html_transposed", "sb6183_html": "app.drivers.formats.html_rows", "sb6190_html": "app.drivers.formats.html_rows", + "sb8200_cbn_xml": "app.drivers.formats.xml_payloads", "sercom_dm1000_json": "app.drivers.formats.sercom", "surfboard_hnap": "app.drivers.formats.surfboard", "tc4400_html": "app.drivers.formats.html_rows", diff --git a/tests/architecture/test_driver_formats_static.py b/tests/architecture/test_driver_formats_static.py index 302df486..78772023 100644 --- a/tests/architecture/test_driver_formats_static.py +++ b/tests/architecture/test_driver_formats_static.py @@ -34,6 +34,7 @@ "sb6141_transposed_html": "parse_sb6141_transposed_html", "sb6183_html": "parse_sb6183_html", "sb6190_html": "parse_sb6190_html", + "sb8200_cbn_xml": "parse_sb8200_cbn_xml", "sercom_dm1000_json": "parse_sercom_dm1000_json", "surfboard_hnap": "parse_surfboard_hnap", "tc4400_html": "parse_tc4400_html", @@ -130,8 +131,8 @@ def test_registry_matrix_is_complete_and_alias_safe_in_both_directions(): for profile in profiles } assert matrix_profiles == set(FORMAT_PROFILE_MODULES) - assert len(EXPECTED_CLASS_FAMILIES) == 20 - assert len(PROFILE_ENTRYPOINTS) == 22 + assert len(EXPECTED_CLASS_FAMILIES) == 21 + assert len(PROFILE_ENTRYPOINTS) == 23 def test_migrated_private_methods_are_finite_one_statement_delegations(): diff --git a/tests/drivers/driver_format_cases.py b/tests/drivers/driver_format_cases.py index fd4c689e..619a9b67 100644 --- a/tests/drivers/driver_format_cases.py +++ b/tests/drivers/driver_format_cases.py @@ -33,6 +33,7 @@ from app.drivers.sb6141 import SB6141Driver from app.drivers.sb6183 import SB6183Driver from app.drivers.sb6190 import SB6190Driver +from app.drivers.sb8200_cbn import Query as SB8200Query, SB8200CBNDriver from app.drivers.sercom_dm1000 import SercomDM1000Driver from app.drivers.surfboard import SurfboardDriver from app.drivers.tc4400 import TC4400Driver @@ -173,6 +174,20 @@ def get_data(query: Query) -> str: return driver.get_docsis_data() +def _sb8200_cbn(ds: str, us: str, ofdm: str, ofdma: str, signal: str) -> Any: + driver = SB8200CBNDriver("https://modem.invalid", "admin", "password") + payloads = { + SB8200Query.DOWNSTREAM_TABLE: ds, + SB8200Query.UPSTREAM_TABLE: us, + SB8200Query.DOWNSTREAM_OFDM_TABLE: ofdm, + SB8200Query.UPSTREAM_OFDMA_TABLE: ofdma, + SB8200Query.SIGNAL_TABLE: signal, + } + + with patch.object(driver, "_get_data", side_effect=lambda query: payloads[query]): + return driver.get_docsis_data() + + def _cm3000_html(ds: str, us: str, ds_ofdm: str, us_ofdma: str) -> str: return f"""