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..df60fad5 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, parse_optional_finite_float + +_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,311 @@ def parse_ch7465_xml( {"docsis": "3.0", "downstream": downstream, "upstream": upstream}, tuple(diagnostics), ) + + +def _optional_int(value: object) -> int | None: + """Parse an optional counter, keeping a missing value distinct from zero.""" + try: + return int(str(value).strip()) + except (TypeError, ValueError): + 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_root(payload: str | None, expected_tag: str) -> ET.Element | None: + """Return the named table root, or None when absent, malformed, or foreign. + + The modem answers an unauthenticated request with a login page rather than + an error, so a document that parses but is not the expected table must not + be reported as a table holding no channels. + """ + if not payload: + return None + try: + root = ET.fromstring(payload) + except ET.ParseError: + return None + return root if root.tag == expected_tag else None + + +def _sb8200_optional_root( + payload: str | None, + expected_tag: str, + diagnostics: list[ParseDiagnostic], + *, + direction: str, + field: str, +) -> ET.Element | None: + """Resolve an enrichment table, recording why it could not be used.""" + root = _sb8200_root(payload, expected_tag) + if root is None and payload: + diagnostics.append(_sb8200_issue("invalid_xml", direction=direction, field=field)) + return root + + +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 + if dsid in counters: + diagnostics.append(_sb8200_issue( + "duplicate_row", direction="downstream", index=index, field="dsid", + )) + continue + corrected = _optional_int(_text(entry.find("correctable"))) + uncorrected = _optional_int(_text(entry.find("uncorrectable"))) + if corrected is None or uncorrected is None: + diagnostics.append(_sb8200_issue( + "invalid_row", direction="downstream", index=index, field="codewords", + )) + counters[dsid] = (corrected, uncorrected) + return counters, diagnostics + + +def _sb8200_locked( + channel: ET.Element, + field: str, + diagnostics: list[ParseDiagnostic], + *, + direction: str, + index: int, +) -> bool: + """Report lock state, treating only a known marker as authoritative. + + A missing marker counts as locked; ``0`` counts as unlocked; anything else + is a firmware spelling this profile has never seen, so the row is dropped + with a diagnostic rather than silently disappearing. + """ + state = _text(channel.find(field)).strip() + if state in ("", "1"): + return True + if state != "0": + diagnostics.append(_sb8200_issue( + "unknown_lock_state", direction=direction, index=index, field=field, + )) + return False + + +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", diagnostics, direction="downstream", index=index + ): + continue + channel_id = _optional_int(_text(entry.find("chid"))) + power = parse_optional_finite_float(_text(entry.find("pow"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="downstream", index=index, field="sc_qam", + )) + continue + channel: RawChannel = { + "channelID": channel_id, + "frequency": hz_to_mhz(_text(entry.find("freq"))), + "powerLevel": power, + } + snr = parse_optional_finite_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 = parse_optional_finite_float(_text(entry.find("PLCPower"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="downstream", index=index, field="ofdm", + )) + continue + channel: RawChannel = { + "channelID": channel_id, + "type": "OFDM", + "frequency": hz_to_mhz(_text(entry.find("Subcarr0Frequency"))), + "powerLevel": power, + "mse": None, + } + mer = parse_optional_finite_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", diagnostics, direction="upstream", index=index + ): + continue + channel_id = _optional_int(_text(entry.find("usid"))) + power = parse_optional_finite_float(_text(entry.find("power"))) + if channel_id is None or power is None: + diagnostics.append(_sb8200_issue( + "invalid_channel", direction="upstream", index=index, field="sc_qam", + )) + 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["multiplex"] = multiplex + # The table reports the symbol rate in Msym/s; channels carry ksym/s. + symbol_rate = parse_optional_finite_float(_text(entry.find("srate"))) + if symbol_rate is not None: + channel["symbolRate"] = 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, "downstream_table") + upstream_root = _sb8200_root(upstream_xml, "upstream_table") + 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_optional_root( + signal_xml, "signal_table", diagnostics, + direction="downstream", field="signal_table", + ) + if signal_root is not None: + 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_optional_root( + downstream_ofdm_xml, "downstreamOFDM_table", diagnostics, + direction="downstream", field="ofdm_table", + ) + if ofdm_root is not None: + 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_optional_root( + upstream_ofdma_xml, "upstreamOFDMA_table", diagnostics, + direction="upstream", field="ofdma_table", + ) + if ofdma_root is not None and _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..826059f9 --- /dev/null +++ b/app/drivers/sb8200_cbn.py @@ -0,0 +1,416 @@ +"""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 `sessionToken` +cookie, then wrapped in a `HS::` envelope. That cookie rotates on +every response, so the credentials must be keyed with the token the login +request itself carries. +""" + +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 broken or +# hostile endpoint cannot stream an unbounded document into memory. +MAX_RESPONSE_BYTES = 1_048_576 + +_UPTIME_RE = re.compile(r"(\d+)\s*day\(s\)\s*(\d+)h:(\d+)m:(\d+)s") +_SID_RE = re.compile(r"[A-Za-z0-9]{1,64}") +_UNREADABLE = (requests.RequestException, RuntimeError, ValueError, ET.ParseError) +_FALLBACK_DEVICE: DeviceInfo = { + "manufacturer": "ARRIS", "model": "SB8200", "sw_version": "", +} + + +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, rotates its CSRF token on + every response, and rate-limits repeated logins, so the session is held + 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 bound to the client address and User-Agent, so this + # must stay stable for the lifetime of the login. + session.headers["User-Agent"] = "docsight/2" + # The size bound counts the bytes that arrive on the socket, so the + # transport must not hand back a stream that inflates while it is read. + session.headers["Accept-Encoding"] = "identity" + self._session: requests.Session = session + self._logged_in = False + self._reauthenticated = 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 or swaps drivers. + self._finalizer = weakref.finalize(self, SB8200CBNDriver._cleanup, url, session) + self._finalizer.atexit = True + + @staticmethod + def _clear_sid(session: requests.Session) -> None: + """Drop every `SID` cookie the jar holds. + + `clear(name=...)` requires a domain and a path, and a duplicated SID + makes `__delitem__` raise, so each match is dropped explicitly. + """ + for cookie in list(session.cookies): + if cookie.name == "SID": + session.cookies.clear(cookie.domain, cookie.path, cookie.name) + + @staticmethod + def _held_sid(session: requests.Session) -> str: + """Read the session cookie without tripping over a duplicate.""" + for cookie in session.cookies: + if cookie.name == "SID" and cookie.value: + return cookie.value + return "" + + @staticmethod + def _cleanup(url: str, session: requests.Session) -> None: + """Close the Web-UI session so another client can connect.""" + try: + if SB8200CBNDriver._held_sid(session): + session.post( + f"{url}/xml/setter.xml", + data={ + "token": session.cookies.get("sessionToken", ""), + "fun": str(Action.LOGOUT.value), + }, + timeout=10, + stream=True, + ).close() + SB8200CBNDriver._clear_sid(session) + except Exception: # noqa: BLE001 - must never raise from a finalizer + pass + finally: + session.close() + + def login(self) -> None: + """Authenticate with the modem, reusing an already-open session.""" + if self._logged_in: + return + + # Release a session this driver still holds before discarding the + # cookies that identify it, so a failed login cannot strand an open + # session on a modem that permits only one. + self._release_session() + + # Only the rotating session cookie is needed, so the login page body + # is never downloaded. + response = self._session.get( + f"{self._url}/common_page/login.html", timeout=10, stream=True + ) + response.close() + response.raise_for_status() + + # Resolve the hardware model first: that request rotates the session + # token, and the credentials must be keyed with the token the login + # request will carry. + hw_model = self._hardware_model() + token = self._session_token() + if not token: + raise RuntimeError("Modem did not issue a session token") + + body = self._set_data(Action.LOGIN, { + "Username": self._encrypt(self._user or "admin", token, hw_model), + "Password": self._encrypt(self._password, token, hw_model), + }) + sid = body.split("SID=", 1)[1].strip() if "SID=" in body else "" + if not body.lstrip().startswith("successful") or not _SID_RE.fullmatch(sid): + # The response body can echo account state, so it is not logged. + raise RuntimeError("Modem authentication failed: check username and password") + + # The modem may also answer the login with its own `Set-Cookie: SID`. + # `cookies.set()` only replaces a cookie carrying the same domain and + # path, so both would survive, be sent together, and make every later + # read of the cookie raise `CookieConflictError`. + self._clear_sid(self._session) + self._session.cookies.set("SID", sid) + 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. + + Only the two SC-QAM tables are required. The OFDM, OFDMA, and codeword + tables enrich the result, so an endpoint this firmware does not serve + degrades to `None` instead of discarding the channels that were read. + """ + self._begin_call() + downstream_xml = self._get_data(Query.DOWNSTREAM_TABLE) + upstream_xml = self._get_data(Query.UPSTREAM_TABLE) + parsed = parse_sb8200_cbn_xml( + downstream_xml=downstream_xml, + upstream_xml=upstream_xml, + downstream_ofdm_xml=self._get_optional_data(Query.DOWNSTREAM_OFDM_TABLE), + upstream_ofdma_xml=self._get_optional_data(Query.UPSTREAM_OFDMA_TABLE), + signal_xml=self._get_optional_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. + """ + self._begin_call() + try: + root = self._xml(self._get_data(Query.SYSTEM_INFO)) + except _UNREADABLE as e: + log.warning("Failed to get device info: %s", e) + return dict(_FALLBACK_DEVICE) + if root is None: + log.warning("Modem returned unreadable system info") + return dict(_FALLBACK_DEVICE) + + 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 + + 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. + """ + self._begin_call() + try: + root = self._xml(self._get_data(Query.SYSTEM_INFO)) + except _UNREADABLE as e: + log.warning("Failed to get connection info: %s", e) + return {} + mode = _node_text(root.find("cm_docsis_mode")) if root is not None else "" + return {"connection_type": mode} if mode else {} + + @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 _begin_call(self) -> None: + """Open one driver call with a fresh single-re-authentication budget.""" + self._reauthenticated = False + + def _release_session(self) -> None: + """Close a session this driver still holds, then drop its cookies.""" + if self._held_sid(self._session): + try: + self._request(Action.LOGOUT) + except _UNREADABLE as e: + log.debug("Logout before re-authentication failed: %s", e) + self._session.cookies.clear() + self._logged_in = False + + 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(Query.GLOBAL_SETTINGS)[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, function: Query | Action, data: dict[str, str] | None = None + ) -> tuple[str, int]: + """Post one function call, returning its body and HTTP status.""" + endpoint = "getter.xml" if isinstance(function, Query) else "setter.xml" + payload = {"token": self._session_token(), "fun": str(function.value)} + if data: + payload |= data + response = self._session.post( + f"{self._url}/xml/{endpoint}", + data=payload, + timeout=10, + allow_redirects=False, + stream=True, + ) + try: + response.raise_for_status() + # Reading with `decode_content=True` bounds the compressed bytes + # taken from the socket, not the bytes they expand to, so a + # kilobyte of gzip could still inflate past the limit before it is + # measured. The request asks for `identity`; anything else is + # refused before a single byte is buffered. + encoding = response.headers.get("Content-Encoding", "").strip().lower() + if encoding not in ("", "identity"): + raise RuntimeError( + f"Modem response for fun={function.value} is {encoding}-encoded" + ) + body = response.raw.read(MAX_RESPONSE_BYTES + 1, decode_content=False) + finally: + response.close() + if len(body) > MAX_RESPONSE_BYTES: + raise RuntimeError(f"Modem response for fun={function.value} exceeds the size limit") + return body.decode("utf-8", errors="replace"), response.status_code + + @staticmethod + def _session_lost(body: str, status: int) -> bool: + """Report whether an answer means the session is no longer valid. + + Measured against the device: once the session lapses, every table code + answers `302` with an empty body while `fun=1` still returns 200. Any + request carrying a stale CSRF token drops the session, so this is a + routine condition rather than an error. + """ + return status == 302 or not body.strip() + + def _reauthenticate(self) -> bool: + """Re-establish a lapsed session, at most once per call. + + Only a session this driver believed it held is re-established, which + keeps a failing login off the modem's rate limiter when a table is + unreadable for any other reason. The budget is spent once per call, so + a poll reading five tables still costs the limiter a single login. + """ + if self._reauthenticated or not self._logged_in: + return False + self._reauthenticated = True + log.info("Session lost, re-authenticating") + self._logged_in = False + self.login() + return True + + def _get_data(self, query: Query) -> str: + """Query a required table, re-authenticating at most once per call.""" + body, status = self._request(query) + if not self._session_lost(body, status): + return body + if not self._reauthenticate(): + raise RuntimeError(f"Modem returned no data for fun={query.value}") + + body, status = self._request(query) + if self._session_lost(body, status): + raise RuntimeError( + f"Modem returned no data for fun={query.value} after re-authentication" + ) + return body + + def _get_optional_data(self, query: Query) -> str | None: + """Query an enrichment table, degrading to `None` when it is absent. + + A `302` is session loss and is worth the call's single + re-authentication. A `200` carrying an empty body is how this firmware + answers for a table it does not serve, so it must not spend a login on + the modem's rate limiter. A transport or HTTP failure here must not + discard the SC-QAM channels the required tables already returned. + """ + try: + body, status = self._request(query) + if status == 302 and self._reauthenticate(): + body, status = self._request(query) + except _UNREADABLE as e: + # The response body can echo session state, so only the failure + # class is logged. + log.warning( + "Optional table fun=%s is unavailable (%s)", query.value, type(e).__name__ + ) + return None + if status == 302 or not body.strip(): + log.debug("Optional table fun=%s reported no data", query.value) + return None + 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(action, data)[0] diff --git a/docs/index.html b/docs/index.html index c446ff4d..ac914061 100644 --- a/docs/index.html +++ b/docs/index.html @@ -219,7 +219,7 @@

Your ISP says everything is fine. DOCSight shows the timeline.

  • Local data
  • Demo mode
  • Reports
    • -
  • 20 modem families
    • +
  • 21 modem families
  • MIT licensed
    • 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..8f53cdfa 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,25 @@ 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, + } + + # The OFDM, OFDMA, and codeword tables travel the optional path, which + # degrades to None rather than raising, so both fetchers must be served. + with ( + patch.object(driver, "_get_data", side_effect=lambda query: payloads[query]), + patch.object(driver, "_get_optional_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"""