Recognize the recovery AP favicon redirect in browser checks

This commit is contained in:
mikecarper committed 2026-10-18 01:06:30 -07:00
1 parent 4b0b52e522
commit cfbdc13543
2 files changed
+159 -5

No files matched your search

+50 -4
View File
@@ -118,14 +118,17 @@ def served_html_hash(endpoint, timeout):
class PassiveCDP(CDP):
"""Observe the existing transport without Fetch interception or patches."""
def __init__(self, websocket, debug_port, profile, origin):
def __init__(self, websocket, debug_port, profile, origin, *, setup_mode=False):
self.origin = origin
self.setup_mode = setup_mode
self.operations = Counter()
self.unexpected_requests = 0
self.write_requests = 0
self.network_started = time.monotonic()
self.requests = []
self.request_ids = {}
self.approved_favicons = {}
self.ap_favicon_redirects = []
self.room_operations = {}
self.runtime_errors = Counter()
self.browser_errors = Counter()
@@ -139,6 +142,30 @@ class PassiveCDP(CDP):
self.socket.close()
raise
def is_ap_favicon_redirect(self, identifier, request, response):
previous = self.approved_favicons.get(identifier)
if not (self.setup_mode is True and isinstance(identifier, str) and identifier
and previous is not None and self.request_ids.get(identifier) is previous
and not previous["finished"] and not previous["failed"]
and isinstance(response, dict) and isinstance(response.get("headers"), dict)):
return False
locations = [value for key, value in response["headers"].items()
if key.lower() == "location"]
return (response.get("url") == self.origin + "/favicon.ico"
and response.get("status") in (301, 302, 303, 307, 308)
and locations == ["http://192.168.4.1/"]
and request.get("method") == "GET"
and request.get("url") == "http://192.168.4.1/"
and not request.get("postData"))
def favicon_redirect_evidence(self):
return [{"from_ordinal": source["ordinal"], "to_ordinal": target["ordinal"],
"redirect_status": source["status"],
"followup_state": ("failed" if target["failed"] else
"finished" if target["finished"] else "pending"),
"followup_finished": target["finished"], "followup_failed": target["failed"]}
for source, target in self.ap_favicon_redirects]
def network(self, method, event):
identifier = event.get("requestId")
elapsed = round(time.monotonic() - self.network_started, 3)
@@ -171,6 +198,15 @@ class PassiveCDP(CDP):
operation = "unexpected"
elif path == "/api/room/result":
operation = self.room_operations.get(sequence, "unmatched_result")
redirect_source = None
if "redirectResponse" in event:
# Only the setup portal's known favicon redirect may leave the
# forwarded origin. A redirect to another ordinarily allowed
# path must not regain permission through the normal allowlist.
read_only = self.is_ap_favicon_redirect(identifier, request, event["redirectResponse"])
if read_only:
redirect_source = self.approved_favicons.pop(identifier)
path = "captive_favicon_root"
entry = {"ordinal": len(self.requests) + 1, "path": path,
"method": verb if verb in ("GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS") else "other",
"operation": operation, "room_sequence": sequence,
@@ -178,11 +214,19 @@ class PassiveCDP(CDP):
"finished": False, "failed": False}
self.requests.append(entry)
self.request_ids[identifier] = entry
if redirect_source is not None:
redirect_source.update(status=int(event["redirectResponse"]["status"]),
response_seconds=elapsed, redirected=True,
redirect_to_ordinal=entry["ordinal"])
entry["redirect_from_ordinal"] = redirect_source["ordinal"]
self.ap_favicon_redirects.append((redirect_source, entry))
if not read_only:
self.unexpected_requests += 1
if verb not in ("GET", "HEAD", "OPTIONS"):
self.write_requests += 1
raise CheckFailure("unexpected_browser_request_observed_passively")
if path == "/favicon.ico" and isinstance(identifier, str) and identifier:
self.approved_favicons[identifier] = entry
return
entry = self.request_ids.get(identifier)
if entry is None:
@@ -237,12 +281,12 @@ class PassiveCDP(CDP):
raise CheckFailure("CDP_request_deadline")
def wait_for_room_browser(websocket, debug_port, profile, origin, browser, timeout=60):
def wait_for_room_browser(websocket, debug_port, profile, origin, browser, timeout=60, *, setup_mode=False):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
require(browser.poll() is None, "owned_Chromium_exited_before_ready")
try:
return PassiveCDP(websocket, debug_port, profile, origin)
return PassiveCDP(websocket, debug_port, profile, origin, setup_mode=setup_mode)
except (OSError, urllib.error.URLError, websocket.WebSocketTimeoutException,
websocket.WebSocketConnectionClosedException):
time.sleep(0.1)
@@ -383,7 +427,8 @@ def execute(args, manifest, private, report, record):
"--remote-debugging-address=127.0.0.1", "--remote-debugging-port=" + str(debug_port),
"--user-data-dir=" + str(profile), "about:blank"],
stdout=browser_log, stderr=browser_log, start_new_session=True)
page = wait_for_room_browser(websocket, debug_port, profile, origin, browser)
page = wait_for_room_browser(websocket, debug_port, profile, origin, browser,
setup_mode=manifest["status"]["mode"] == "setup")
report["browser_version"] = page.browser_version
report["phase"] = "initial_admin_UI_login"
page.call("Page.navigate", {"url": origin + "/room"})
@@ -489,6 +534,7 @@ def execute(args, manifest, private, report, record):
report["unexpected_browser_requests_observed"] = page.unexpected_requests
report["browser_write_requests_observed"] = page.write_requests
report["browser_request_evidence"] = page.requests
report["ap_favicon_redirect_evidence"] = page.favicon_redirect_evidence()
report["page_runtime_exception_classes"] = dict(page.runtime_errors)
report["browser_error_sources"] = dict(page.browser_errors)
page.socket.close()
+109 -1
View File
@@ -240,6 +240,113 @@ class ContractTests(unittest.TestCase):
resolve.assert_not_called()
class BrowserFaviconRedirectTests(unittest.TestCase):
origin = "http://127.0.0.1:19081"
def page(self, *, setup=True, approve=True):
with mock.patch.object(browser_hil.CDP, "__init__", return_value=None), \
mock.patch.object(browser_hil.PassiveCDP, "call"):
page = browser_hil.PassiveCDP(None, 1, Path("/inert"), self.origin, setup_mode=setup)
if approve:
page.network("Network.requestWillBeSent", {"requestId": "favicon-1", "request": {
"url": self.origin + "/favicon.ico", "method": "GET", "headers": {}}})
return page
def redirect(self):
return {"requestId": "favicon-1", "request": {
"url": "http://192.168.4.1/", "method": "GET", "headers": {}},
"redirectResponse": {"url": self.origin + "/favicon.ico", "status": 302,
"headers": {"Location": "http://192.168.4.1/"}}}
def test_setup_favicon_exact_chain_is_passive_and_pending_is_not_finished(self):
page = self.page()
page.network("Network.requestWillBeSent", self.redirect())
source, target = page.requests
self.assertEqual(source["status"], 302)
self.assertTrue(source["redirected"])
self.assertFalse(source["finished"])
self.assertEqual(target["path"], "captive_favicon_root")
self.assertEqual(target["redirect_from_ordinal"], source["ordinal"])
self.assertEqual(page.unexpected_requests, 0)
self.assertEqual(page.write_requests, 0)
self.assertEqual(page.favicon_redirect_evidence(), [{
"from_ordinal": 1, "to_ordinal": 2, "redirect_status": 302,
"followup_state": "pending", "followup_finished": False, "followup_failed": False}])
for private in (self.origin, "192.168.4.1", "Location"):
self.assertNotIn(private, json.dumps(page.requests + page.favicon_redirect_evidence()))
def test_unused_favicon_failure_is_recorded_as_failed_not_drained(self):
page = self.page()
page.network("Network.requestWillBeSent", self.redirect())
page.network("Network.loadingFailed", {"requestId": "favicon-1", "errorText": "net::ERR_ADDRESS_UNREACHABLE"})
item = page.favicon_redirect_evidence()[0]
self.assertEqual(item["followup_state"], "failed")
self.assertTrue(item["followup_failed"])
self.assertFalse(item["followup_finished"])
self.assertEqual(page.requests[1]["failure_class"], "net::ERR_ADDRESS_UNREACHABLE")
def test_completed_favicon_response_preserves_both_statuses(self):
page = self.page()
page.network("Network.requestWillBeSent", self.redirect())
page.network("Network.responseReceived", {"requestId": "favicon-1", "response": {"status": 200}})
page.network("Network.loadingFinished", {"requestId": "favicon-1", "encodedDataLength": 100})
self.assertEqual([row["status"] for row in page.requests], [302, 200])
self.assertEqual(page.favicon_redirect_evidence()[0]["followup_state"], "finished")
def test_redirect_requires_setup_mode_and_same_preapproved_request(self):
for setup, approve, identifier in ((False, True, "favicon-1"),
(True, False, "favicon-1"),
(True, True, "another-request")):
with self.subTest(setup=setup, approve=approve, identifier=identifier):
page = self.page(setup=setup, approve=approve)
event = self.redirect()
event["requestId"] = identifier
with self.assertRaisesRegex(hil.CheckFailure, "unexpected_browser_request"):
page.network("Network.requestWillBeSent", event)
self.assertEqual(page.favicon_redirect_evidence(), [])
def test_wrong_location_source_status_destination_or_write_is_rejected(self):
variations = [
("response", "url", self.origin + "/api/status"),
("response", "status", 200),
("response", "headers", {}),
("response", "headers", {"Location": "http://192.168.4.2/"}),
("response", "headers", {"Location": "http://192.168.4.1/?other"}),
("response", "headers", {"Location": "http://192.168.4.1/", "location": "http://192.168.4.1/"}),
("request", "url", "http://192.168.4.1/?other"),
("request", "url", "http://192.168.4.1/#other"),
("request", "url", "http://192.168.4.1/api/status"),
("request", "url", self.origin + "/room"),
("request", "url", self.origin + "/api/status"),
("request", "method", "POST"),
("request", "postData", "unexpected-body"),
]
for target, key, value in variations:
with self.subTest(target=target, key=key, value=value):
page = self.page()
event = self.redirect()
event["redirectResponse" if target == "response" else "request"][key] = value
with self.assertRaisesRegex(hil.CheckFailure, "unexpected_browser_request"):
page.network("Network.requestWillBeSent", event)
self.assertEqual(page.favicon_redirect_evidence(), [])
def test_missing_redirect_metadata_terminal_prior_or_second_hop_is_rejected(self):
for case in ("missing-response", "prior-failed", "prior-finished", "second-hop"):
with self.subTest(case=case):
page = self.page()
event = self.redirect()
if case == "missing-response":
del event["redirectResponse"]
elif case == "prior-failed":
page.network("Network.loadingFailed", {"requestId": "favicon-1"})
elif case == "prior-finished":
page.network("Network.loadingFinished", {"requestId": "favicon-1"})
else:
page.network("Network.requestWillBeSent", event)
with self.assertRaisesRegex(hil.CheckFailure, "unexpected_browser_request"):
page.network("Network.requestWillBeSent", event)
class BrowserStartupTests(unittest.TestCase):
def websocket(self):
return SimpleNamespace(
@@ -274,9 +381,10 @@ class BrowserStartupTests(unittest.TestCase):
websocket.WebSocketConnectionClosedException(), page]) as attach, \
mock.patch.object(browser_hil.time, "sleep") as pause:
actual = browser_hil.wait_for_room_browser(
websocket, 12345, Path("/owned"), "http://127.0.0.1:1234", process)
websocket, 12345, Path("/owned"), "http://127.0.0.1:1234", process, setup_mode=True)
self.assertIs(actual, page)
self.assertEqual(attach.call_count, 3)
self.assertTrue(all(call.kwargs == {"setup_mode": True} for call in attach.call_args_list))
self.assertEqual(pause.call_count, 2)
def test_ownership_refusal_is_not_retried(self):