Skip to content

Commit f431707

Browse files
itsDNNSitsDNNS
committed
fix: add CM3000 form login fallback
1 parent 5018834 commit f431707

File tree

2 files changed

+131
-1
lines changed

2 files changed

+131
-1
lines changed

‎app/drivers/cm3000.py‎

Lines changed: 84 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
import logging
1818
import re
19+
from urllib.parse import urljoin
1920

2021
import requests
2122

@@ -41,6 +42,12 @@
4142
)
4243
_RE_TITLE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
4344
_RE_FORM_ACTION = re.compile(r"<form[^>]+action=['\"]([^'\"]+)['\"]", re.IGNORECASE)
45+
_RE_FORM_BLOCK = re.compile(
46+
r"<form[^>]*action=['\"](?P<action>[^'\"]+)['\"][^>]*>(?P<body>.*?)</form>",
47+
re.IGNORECASE | re.DOTALL,
48+
)
49+
_RE_INPUT = re.compile(r"<input\b(?P<attrs>[^>]*)>", re.IGNORECASE | re.DOTALL)
50+
_RE_ATTR = re.compile(r"(?P<name>[A-Za-z_:][-A-Za-z0-9_:.]*)\s*=\s*['\"](?P<value>[^'\"]*)['\"]")
4451

4552
# Fields per channel for each section (after the leading count value).
4653
_DS_QAM_FIELDS = 9 # num|lock|mod|chID|freq|power|snr|corrErr|uncorrErr
@@ -73,7 +80,18 @@ def login(self) -> None:
7380
try:
7481
r = self._session.get(f"{self._url}{_STATUS_PATH}", timeout=30)
7582
r.raise_for_status()
76-
self._ensure_status_page(r.text)
83+
try:
84+
self._ensure_status_page(r.text)
85+
except RuntimeError:
86+
if not self._looks_like_login_page(r.text):
87+
self._log_status_page_diagnostics(r.text, "login")
88+
raise
89+
if not self._login_via_form():
90+
self._log_status_page_diagnostics(r.text, "login")
91+
raise
92+
r = self._session.get(f"{self._url}{_STATUS_PATH}", timeout=30)
93+
r.raise_for_status()
94+
self._ensure_status_page(r.text)
7795
self._status_html = r.text
7896
log.info("CM3000 auth OK")
7997
return
@@ -178,6 +196,71 @@ def _fetch_status_page(self) -> str:
178196
raise
179197
return r.text
180198

199+
def _login_via_form(self) -> bool:
200+
"""Try the newer Netgear web login flow used before DocsisStatus access."""
201+
login_url = urljoin(f"{self._url}/", "Login.htm")
202+
try:
203+
r = self._session.get(login_url, timeout=30)
204+
r.raise_for_status()
205+
action, payload = self._extract_login_form(r.text)
206+
if not action:
207+
action = "/goform/Login"
208+
payload = payload or {}
209+
self._apply_login_credentials(payload)
210+
if not any(v for k, v in payload.items() if "pass" in k.lower()):
211+
payload["loginPassword"] = self._password
212+
if not any(v for k, v in payload.items() if "user" in k.lower() or "name" in k.lower()):
213+
payload["loginName"] = self._user
214+
post_url = urljoin(f"{self._url}/", action.lstrip("/"))
215+
r = self._session.post(post_url, data=payload, timeout=30)
216+
r.raise_for_status()
217+
return True
218+
except requests.RequestException as exc:
219+
log.debug("CM3000 form login failed: %s", exc)
220+
return False
221+
222+
def _apply_login_credentials(self, payload: dict) -> None:
223+
"""Populate parsed login form fields with configured credentials."""
224+
lowered = {k.lower(): k for k in payload}
225+
user_keys = [k for k in payload if any(token in k.lower() for token in ("loginname", "username", "user", "name"))]
226+
pass_keys = [k for k in payload if "pass" in k.lower()]
227+
for key in user_keys:
228+
payload[key] = self._user
229+
for key in pass_keys:
230+
payload[key] = self._password
231+
if not user_keys and "loginname" not in lowered:
232+
payload["loginName"] = self._user
233+
if not pass_keys and "loginpassword" not in lowered:
234+
payload["loginPassword"] = self._password
235+
236+
@staticmethod
237+
def _extract_login_form(html: str) -> tuple[str | None, dict]:
238+
"""Extract login form action and input values from Login.htm."""
239+
match = _RE_FORM_BLOCK.search(html or "")
240+
if not match:
241+
return None, {}
242+
action = match.group("action").strip()
243+
body = match.group("body")
244+
payload = {}
245+
for input_match in _RE_INPUT.finditer(body):
246+
attrs = {
247+
attr_match.group("name").lower(): attr_match.group("value")
248+
for attr_match in _RE_ATTR.finditer(input_match.group("attrs"))
249+
}
250+
name = attrs.get("name")
251+
if not name:
252+
continue
253+
input_type = attrs.get("type", "").lower()
254+
if input_type in {"submit", "button", "image"}:
255+
continue
256+
payload[name] = attrs.get("value", "")
257+
return action, payload
258+
259+
@staticmethod
260+
def _looks_like_login_page(html: str) -> bool:
261+
lower_html = (html or "").lower()
262+
return any(marker in lower_html for marker in _LOGIN_MARKERS)
263+
181264
@staticmethod
182265
def _ensure_status_page(html: str) -> None:
183266
"""Reject login/placeholder pages that would otherwise parse as zero channels."""

‎tests/test_cm3000_driver.py‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -288,6 +288,53 @@ def test_login_accepts_status_page_even_with_login_markers(self, driver):
288288
driver.login()
289289
assert driver._status_html == mock_response.text
290290

291+
def test_login_falls_back_to_form_login(self, driver):
292+
login_wrapper = MagicMock()
293+
login_wrapper.raise_for_status = MagicMock()
294+
login_wrapper.text = """
295+
<html><body>
296+
<script>
297+
if (sessionStorage.getItem('PrivateKey') === null) {
298+
window.location.replace('../Login.htm');
299+
}
300+
</script>
301+
</body></html>
302+
"""
303+
login_page = MagicMock()
304+
login_page.raise_for_status = MagicMock()
305+
login_page.text = """
306+
<html><body>
307+
<form action="/goform/Login" method="post">
308+
<input type="hidden" name="foo" value="bar">
309+
<input type="text" name="loginName" value="">
310+
<input type="password" name="loginPassword" value="">
311+
</form>
312+
</body></html>
313+
"""
314+
status_page = MagicMock()
315+
status_page.raise_for_status = MagicMock()
316+
status_page.text = STATUS_HTML
317+
login_submit = MagicMock()
318+
login_submit.raise_for_status = MagicMock()
319+
320+
with (
321+
patch.object(
322+
driver._session,
323+
"get",
324+
side_effect=[login_wrapper, login_page, status_page],
325+
) as mock_get,
326+
patch.object(driver._session, "post", return_value=login_submit) as mock_post,
327+
):
328+
driver.login()
329+
330+
assert driver._status_html == STATUS_HTML
331+
assert mock_get.call_count == 3
332+
mock_post.assert_called_once_with(
333+
"http://192.168.100.1/goform/Login",
334+
data={"foo": "bar", "loginName": "admin", "loginPassword": "password"},
335+
timeout=30,
336+
)
337+
291338
def test_login_accepts_double_quoted_status_page_with_login_markers(self, driver):
292339
mock_response = MagicMock()
293340
mock_response.raise_for_status = MagicMock()

0 commit comments

Comments
 (0)