From b1a2e03fb95f5000280ebd69c08a1a181ffd6bee Mon Sep 17 00:00:00 2001 From: vringar Date: Sat, 20 Jun 2026 01:11:19 +0000 Subject: [PATCH] test: adversarial robustness suite for the crawl pipeline Generate adversarial circumstances for the crawl pipeline and assert it degrades gracefully: forward progress (every started visit reaches a terminal state), no silent data loss, no hang, proper incomplete-visit recording, and the crawl continues via watchdog/BrowserManager recovery. Browser-free tests drive the real StorageController subprocess in-process (via DataSocket) and the asyncio server via raw sockets, injecting provider faults, wire-protocol hostility, and malformed records. Browser-required tests exercise the full TaskManager -> BrowserManager -> Firefox recovery path for crashing/hanging custom commands and a mid-visit browser kill; they skip cleanly where no launchable Firefox is present and run for real in CI. The pipeline survives socket hostility (server stays up after every hostile frame incl. a 2 GiB length prefix; no hang), transient store failures (the controller recovers on a fresh connection), and malformed records (huge values + injection-y strings against the real SQLite path). The suite captures confirmed defects as xfail(strict) tests (the failing assertion is the finding; CI flags an XPASS once fixed): - a raising store_record task strands the visit: it is never enqueued to the completion queue, on finalize or shutdown -> data loss + a callback-bearing CommandSequence hangs forever - the same raise tears down the whole shared connection (BrokenPipe for subsequent records); the TaskManager's single long-lived socket means one bad record can break the crawl - a permanent write_table fault crashes the controller mid-drain -> completed visits lost These fixes are design calls (count-and-continue vs fail-loud, and whether a per-record failure should ever close the connection) and are left for a maintainer; this change delivers the adversarial tests + findings and a docs catalogue under docs/developers/Adversarial-Robustness.md. --- docs/developers/Adversarial-Robustness.md | 124 ++++++ test/storage/conftest.py | 32 ++ test/storage/test_adversarial_socket.py | 125 ++++++ .../test_adversarial_storage_controller.py | 419 ++++++++++++++++++ test/test_adversarial_pipeline.py | 253 +++++++++++ 5 files changed, 953 insertions(+) create mode 100644 docs/developers/Adversarial-Robustness.md create mode 100644 test/storage/conftest.py create mode 100644 test/storage/test_adversarial_socket.py create mode 100644 test/storage/test_adversarial_storage_controller.py create mode 100644 test/test_adversarial_pipeline.py diff --git a/docs/developers/Adversarial-Robustness.md b/docs/developers/Adversarial-Robustness.md new file mode 100644 index 000000000..6c8b76db1 --- /dev/null +++ b/docs/developers/Adversarial-Robustness.md @@ -0,0 +1,124 @@ +# Adversarial robustness of the crawl pipeline + +OpenWPM is built to crawl thousands-to-millions of untrusted, often hostile web +pages. The pipeline must therefore **degrade gracefully** under adversarial +conditions rather than hang, lose data silently, or stop making progress. + +This document describes the adversarial / chaos test suite and the robustness +gaps it surfaced. + +## The graceful-degradation property + +For every adversarial condition, the pipeline must guarantee: + +1. **Forward progress** — every visit that is *started* reaches a terminal + state, and the crawl continues to the next site. +2. **No silent data loss** — the offending visit is accounted for (completion + queue entry / `incomplete_visits` row), and prior visits' data survives. +3. **No hang** — recovery happens within the per-command timeout plus the + browser restart budget. +4. **Recovery continues the crawl** — the watchdog / `BrowserManager` restart + path brings the browser back and subsequent sites are visited. + +## Test suite + +| File | Tier | Browser? | +|---|---|---| +| `test/storage/test_adversarial_storage_controller.py` | StorageController + providers, driven in-process via a real subprocess + `DataSocket` | no (`pyonly`) | +| `test/storage/test_adversarial_socket.py` | Wire protocol / asyncio server | no (`pyonly`) | +| `test/test_adversarial_pipeline.py` | Full `TaskManager` → `BrowserManager` → Firefox recovery | **yes** | + +The browser-free tests follow the in-process driving pattern of +`test/storage/test_storage_controller.py`. The browser-required tests are +guarded by a `requires_browser` skip (resolved via the same logic as +`get_firefox_binary_path`) so they skip cleanly where no launchable Firefox is +present and run for real in CI (pinned unbranded Firefox + built xpi). They are +not faked. + +### Scenarios + +| # | Scenario | Where | Status | +|---|---|---|---| +| S1a | Custom command `execute()` raises | `test_adversarial_pipeline` | recovery asserted (CI) | +| S1b | Custom command hangs forever | `test_adversarial_pipeline` | timeout+kill asserted (CI) | +| S2 | Crashing extension modification | (design item, see below) | not implemented | +| S3 | Socket-level hostility (truncated / garbage / oversized / wrong-arity frames, mid-message disconnect) | `test_adversarial_socket` | **SURVIVES** | +| S4 | Provider write/flush faults (transient + permanent) | `test_adversarial_storage_controller` | controller recovers from transient; raising store task **strands visit (G1) + tears down shared connection (G1b)**; **permanent write fault = DEFECT (G2)** | +| S5 | Browser killed mid-visit | `test_adversarial_pipeline` | recovery asserted (CI) | +| S6 | Malformed / hostile records (huge values, injection-y strings, missing visit_id) | `test_adversarial_storage_controller` (SQLite) | **SURVIVES** | + +## Confirmed robustness gaps (defects) + +These are captured as `xfail(strict=True)` tests — the failing assertion *is* +the finding. When a gap is fixed the test XPASSes and CI flags it, prompting +removal of the marker. + +### G1 — A raising `store_record` task strands the visit + +`StorageController.store_record` fires each record off as an un-awaited +`asyncio` task and only surfaces exceptions when `finalize_visit_id` awaits +them. If a store task raises: + +- on the **finalize** path, the exception propagates out of `finalize_visit_id` + (after the tasks were already popped) before the completion token is + recorded, so the visit is **never enqueued to the completion queue**; +- on **shutdown**, the same raise aborts the shutdown finalize loop, so an + **unfinalized** visit (e.g. browser died mid-visit) is also never enqueued. + +Impact: a callback-bearing `CommandSequence` hangs forever, and the visit is +silently lost. Tests: `test_raising_store_record_visit_still_finalizes`, +`test_raising_store_record_unfinalized_visit_enqueued_on_shutdown`. + +### G1b — A raising `store_record` task tears down the whole connection + +The same raise propagates out of the per-connection handler (`_handler`), which +then closes that connection. Any client still using that **shared** connection +gets a `BrokenPipeError` on its next send. This matters because the +`TaskManager` keeps a single long-lived `DataSocket` (`self.sock`) for +`site_visits` / `crawl_history` / `finalize` records across **all** visits — so +one bad record can break the socket for the rest of the crawl, not just the +offending visit. The controller itself recovers (a good visit on a *fresh* +connection still completes — see `test_transient_store_failure_controller_recovers`), +but the shared connection does not. Test: +`test_raising_store_record_breaks_shared_connection`. + +### G2 — A permanent `write_table` fault loses completed visits + +A permanent `write_table` failure raises out of `flush_cache` during shutdown, +killing the controller before the completion queue is drained. The visit's +terminal state is lost and the structured-storage shutdown is skipped. Write +failures should be surfaced/counted, not silently drop completed visits. Test: +`test_permanent_write_table_fault_visit_still_completes`. + +### G3 — SQLite silently drops unknown-table / unknown-column records + +(Pre-existing, tracked in crosslink #28/#30.) `SQLiteStorageProvider.store_record` +catches `OperationalError` for an unknown table or column and only logs +"Unsupported record"; the whole record is dropped with no partial save and no +surfaced count. For a measurement framework this masks data loss as a log line. + +The fix for G1/G1b/G2 is a **design call** (where to record the terminal state +when the provider itself is the thing failing — count-and-continue vs. +fail-loud, and whether a per-record failure should ever close the connection) +and should be made by a maintainer; these tests pin the desired invariant. + +### Note: the in-memory test provider is not a faithful stand-in for huge values + +`MemoryStructuredProvider` round-trips records through a cross-process +`multiprocess.Queue`. A multi-MiB record value deadlocks that queue at shutdown +(the feeder thread blocks on a full pipe that no consumer drains). This is a +**test-harness artifact**, not a pipeline property: the real +`SQLiteStorageProvider` writes the same value to disk without issue +(`test_malformed_records_do_not_break_controller` therefore drives SQLite, not +the memory provider). Keep this in mind when extending the suite. + +## Not implemented: S2 (crashing extension modification) + +A faithful test of a WebExtension that throws on load / mid-instrumentation / +floods malformed records requires building a deliberately-broken xpi. Shipping +a broken extension build into the repo (or a parallel build pipeline for it) is +out of scope for the test suite and is left as a follow-up. Its **recovery +shape is already covered by S5**: when a broken extension takes the browser down +(or the `BrowserManager` subprocess dies), the BrowserManager restart path is +the same one S5 exercises. The Python side's tolerance of malformed records the +extension might flood is covered by S3/S6. diff --git a/test/storage/conftest.py b/test/storage/conftest.py new file mode 100644 index 000000000..f92802527 --- /dev/null +++ b/test/storage/conftest.py @@ -0,0 +1,32 @@ +"""Fixtures shared by the storage-tier tests. + +The project-wide ``mp_logger`` fixture (in ``test/conftest.py``) asserts that +no ``ERROR`` line was logged during the test. That is the right default for the +happy-path storage tests, but the *adversarial* storage tests in this directory +deliberately provoke error logging (a provider that raises, a hostile socket +frame, a malformed record). For those we need a logger that captures output the +same way but does not fail the test on the expected ERROR lines. +""" + +import logging +from pathlib import Path +from typing import Any, Generator + +import pytest + +from openwpm.mp_logger import MPLogger + + +@pytest.fixture() +def adversarial_mp_logger(tmp_path: Path) -> Generator[MPLogger, Any, None]: + """An ``MPLogger`` for tests that intentionally log ERRORs. + + Identical to ``mp_logger`` but without the post-test assertion that no + ERROR was logged - the adversarial tests *expect* ERROR lines and assert + on observable pipeline behaviour (forward progress / no data loss / + no hang) instead. + """ + log_path = tmp_path / "openwpm.log" + logger = MPLogger(log_path, log_level_console=logging.DEBUG) + yield logger + logger.close() diff --git a/test/storage/test_adversarial_socket.py b/test/storage/test_adversarial_socket.py new file mode 100644 index 000000000..e4e1bc75a --- /dev/null +++ b/test/storage/test_adversarial_socket.py @@ -0,0 +1,125 @@ +"""Adversarial wire-protocol tests against the real StorageController server. + +The StorageController exposes an ``asyncio`` TCP server speaking the +length-prefixed framing of ``openwpm/socket_interface.py`` (4-byte big-endian +length + 1-byte serialization tag + body). Untrusted/buggy clients (a crashing +extension, a half-open connection, a corrupted frame) can send arbitrary bytes +at it. + +PROPERTY: the server must degrade gracefully - a hostile or truncated frame may +kill *that one connection*, but the server must keep accepting new connections, +must not hang, and must not lose data for well-behaved clients. + +All of these are browser-free: we open raw sockets to the controller's listen +address and verify a subsequent good visit still completes. +""" + +import json +import socket +import struct +import time +from typing import List, Tuple + +import pytest + +from openwpm.storage.in_memory_storage import MemoryStructuredProvider +from openwpm.storage.storage_controller import DataSocket, StorageControllerHandle +from openwpm.storage.storage_providers import TableName +from openwpm.types import VisitId + +pytestmark = pytest.mark.pyonly + + +def _send_raw(addr: Tuple[str, int], payload: bytes) -> None: + """Open a raw socket, dump ``payload``, briefly wait, then close.""" + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3) + s.connect(addr) + try: + s.sendall(payload) + except OSError: + # The server may close on us for some payloads; that is acceptable. + pass + time.sleep(0.3) + s.close() + + +def _hostile_frames() -> List[Tuple[str, bytes]]: + one_tuple = json.dumps(["only_one_element"]).encode("utf-8") + return [ + # Unknown serialization tag 'X' with a 3-byte body. + ("unknown_serialization", struct.pack(">Lc", 3, b"X") + b"abc"), + # Length prefix claims 100 bytes but only 2 are sent, then EOF. + ("truncated_body", struct.pack(">Lc", 100, b"j") + b"ab"), + # Absurd length prefix (2 GiB) with no body - must not allocate/hang. + ("oversized_length_prefix", struct.pack(">Lc", 2**31, b"j")), + # Fewer than the 5 header bytes, then EOF. + ("short_header", b"\x01\x02"), + # Valid JSON frame but the record is a 1-element list (wrong arity); + # the controller logs "Query is not the correct length" and skips. + ("wrong_arity_record", struct.pack(">Lc", len(one_tuple), b"j") + one_tuple), + # Valid JSON frame whose body is not even a sequence. + ( + "non_sequence_record", + (lambda b: struct.pack(">Lc", len(b), b"j") + b)(b"42"), + ), + ] + + +@pytest.mark.usefixtures("adversarial_mp_logger") +@pytest.mark.parametrize( + "name,payload", _hostile_frames(), ids=[n for n, _ in _hostile_frames()] +) +def test_server_survives_hostile_frame(name: str, payload: bytes) -> None: + """After a single hostile frame on its own connection, the controller must + still accept a new connection and complete a good visit. + """ + handle = StorageControllerHandle(MemoryStructuredProvider(), None) + handle.launch() + assert handle.listener_address is not None + addr = handle.listener_address + + _send_raw(addr, payload) + + sock = DataSocket(addr, f"good-after-{name}") + good = VisitId(0xBEEF) + sock.store_record(TableName("site_visits"), good, {"site_url": "ok"}) + sock.finalize_visit_id(good, success=True) + sock.close() + + start = time.time() + handle.shutdown() + elapsed = time.time() - start + assert elapsed < 60, f"shutdown hung after hostile frame {name} ({elapsed:.1f}s)" + + seen = handle.get_new_completed_visits() + assert good in {vid for vid, _ in seen}, ( + f"server did not survive hostile frame {name}; good visit lost; " f"seen={seen}" + ) + + +@pytest.mark.usefixtures("adversarial_mp_logger") +def test_server_survives_abrupt_disconnect_mid_message() -> None: + """A client that connects, sends its name, sends a partial record header, + then drops the connection must not wedge the controller. + """ + handle = StorageControllerHandle(MemoryStructuredProvider(), None) + handle.launch() + assert handle.listener_address is not None + addr = handle.listener_address + + # Send a valid client-name frame, then half of a record header, then close. + name_body = json.dumps("evil-client").encode("utf-8") + payload = struct.pack(">Lc", len(name_body), b"j") + name_body + payload += b"\x00\x00" # 2 of the 5 header bytes of the next message + _send_raw(addr, payload) + + sock = DataSocket(addr, "good-after-disconnect") + good = VisitId(0xD00D) + sock.store_record(TableName("site_visits"), good, {"site_url": "ok"}) + sock.finalize_visit_id(good, success=True) + sock.close() + + handle.shutdown() + seen = handle.get_new_completed_visits() + assert good in {vid for vid, _ in seen}, f"controller wedged; seen={seen}" diff --git a/test/storage/test_adversarial_storage_controller.py b/test/storage/test_adversarial_storage_controller.py new file mode 100644 index 000000000..495c88e2a --- /dev/null +++ b/test/storage/test_adversarial_storage_controller.py @@ -0,0 +1,419 @@ +"""Adversarial / chaos tests driving the *real* ``StorageController`` in-process. + +These tests assert the central robustness property of the crawl pipeline's +storage tier: + + FORWARD PROGRESS + NO SILENT DATA LOSS + Every visit_id that is started (a record is stored for it and/or it is + finalized) MUST eventually reach the ``completion_queue`` exactly once, + even when the underlying ``StructuredStorageProvider`` misbehaves, and the + controller must never hang or silently swallow the visit. + +The controller is driven exactly like the existing storage-controller tests +(`test/storage/test_storage_controller.py`): a real ``StorageControllerHandle`` +spawns the real controller subprocess, and a real ``DataSocket`` feeds records +over a real socket. We then inject adversarial behaviour by subclassing the +in-memory providers to raise at well-defined points. + +Scenarios covered (browser-free): + * S4 - storage provider write/flush faults (transient + permanent) + * S1/S5 (storage side) - a store_record task that raises strands the visit + (and tears down the shared connection); this is the same failure + shape a custom command or browser crash produces once its records + reach the controller + * S6 - malformed / hostile records (huge values, injection-y strings, + missing visit_id), driven against the real SQLite provider + +Where a scenario exposes a *confirmed defect* the test is marked +``xfail(strict=True)`` so the failing assertion is captured as the finding +without breaking CI. If the defect is ever fixed the test will XPASS and CI +will flag it, prompting removal of the marker. See crosslink #46 (this work), +#28/#30 (SQLite silent drop). +""" + +import time +from pathlib import Path +from typing import Any, Dict, List, Set, Tuple + +import pytest +from pyarrow import Table + +from openwpm.storage.in_memory_storage import ( + MemoryArrowProvider, + MemoryStructuredProvider, +) +from openwpm.storage.sql_provider import SQLiteStorageProvider +from openwpm.storage.storage_controller import ( + DataSocket, + StorageControllerHandle, +) +from openwpm.storage.storage_providers import TableName +from openwpm.types import VisitId +from openwpm.utilities import db_utils + +pytestmark = pytest.mark.pyonly + + +# --------------------------------------------------------------------------- +# Adversarial providers +# --------------------------------------------------------------------------- + + +class RaisingStoreProvider(MemoryStructuredProvider): + """``store_record`` raises for every record. + + Models a structured-storage backend that throws while persisting a record + (e.g. a programming error, an unexpected type, or a transient backend + error that is not caught). Because ``StorageController.store_record`` + fires these off as un-awaited ``asyncio`` tasks and only surfaces the + exception when ``finalize_visit_id`` awaits them, a raising store task + aborts finalization for that visit. + """ + + async def store_record( + self, table: TableName, visit_id: VisitId, record: Dict[str, Any] + ) -> None: + raise RuntimeError(f"injected store_record failure for visit {visit_id}") + + +class FailingWriteArrowProvider(MemoryArrowProvider): + """``write_table`` raises permanently (e.g. disk full / S3 outage).""" + + async def write_table(self, table_name: TableName, table: Table) -> None: + raise IOError("injected permanent write_table failure") + + +class TransientFailingStoreProvider(MemoryStructuredProvider): + """``store_record`` fails the first ``n`` calls then recovers. + + Models a transient backend hiccup. A robust controller should still make + forward progress for visits stored after the backend recovers. + """ + + def __init__(self, fail_first: int) -> None: + super().__init__() + self._remaining_failures = fail_first + + async def store_record( + self, table: TableName, visit_id: VisitId, record: Dict[str, Any] + ) -> None: + if self._remaining_failures > 0: + self._remaining_failures -= 1 + raise RuntimeError("injected transient store_record failure") + await super().store_record(table, visit_id, record) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _drain_completions( + handle: StorageControllerHandle, timeout: float = 8.0 +) -> List[Tuple[int, bool]]: + """Poll the completion queue until ``timeout`` and return everything seen.""" + deadline = time.time() + timeout + seen: List[Tuple[int, bool]] = [] + while time.time() < deadline: + seen.extend(handle.get_new_completed_visits()) + time.sleep(0.2) + return seen + + +def _completed_ids(seen: List[Tuple[int, bool]]) -> Set[int]: + return {vid for vid, _ in seen} + + +# --------------------------------------------------------------------------- +# Control: the happy path completes (guards against false negatives below) +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("adversarial_mp_logger") +def test_control_good_visit_completes() -> None: + """A normal visit reaches the completion queue. Establishes the baseline + so that the stranding tests below are meaningful (not just slow polling). + """ + handle = StorageControllerHandle(MemoryStructuredProvider(), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "control") + vid = VisitId(0xC0FFEE) + sock.store_record(TableName("site_visits"), vid, {"site_url": "ok"}) + sock.finalize_visit_id(vid, success=True) + sock.close() + + # The MemoryStructuredProvider only resolves a visit's completion token on + # flush_cache, which happens on the batch timeout or at shutdown. So we + # check completions both during the run and after shutdown. + handle.shutdown() + seen = handle.get_new_completed_visits() + assert vid in _completed_ids(seen), f"good visit never completed; seen={seen}" + assert (vid, True) in seen + + +# --------------------------------------------------------------------------- +# S4 (transient): forward progress survives a transient store failure +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("adversarial_mp_logger") +def test_transient_store_failure_controller_recovers() -> None: + """The *controller* recovers from a transient store_record failure: a later + visit on a fresh connection still completes. + + Forward progress holds at the controller level. Note the recovery requires + a NEW connection - a raising store task tears down the connection that + carried the bad record (see + ``test_raising_store_record_breaks_shared_connection`` below), so the good + visit is sent on its own DataSocket, modelling a fresh per-visit/per-worker + connection. + """ + handle = StorageControllerHandle(TransientFailingStoreProvider(fail_first=1), None) + handle.launch() + assert handle.listener_address is not None + + bad = VisitId(1001) # its store_record will raise (consumes the 1 failure) + good = VisitId(1002) # stored after recovery, fresh conn -> must complete + + bad_conn = DataSocket(handle.listener_address, "transient-bad") + bad_conn.store_record(TableName("site_visits"), bad, {"site_url": "bad"}) + bad_conn.finalize_visit_id(bad, success=False) + time.sleep(1) # let the raising store task run + tear its connection down + bad_conn.close() + + good_conn = DataSocket(handle.listener_address, "transient-good") + good_conn.store_record(TableName("site_visits"), good, {"site_url": "good"}) + good_conn.finalize_visit_id(good, success=True) + good_conn.close() + + handle.shutdown() + seen = handle.get_new_completed_visits() + assert good in _completed_ids( + seen + ), f"controller did not recover: later visit did not complete; seen={seen}" + + +@pytest.mark.usefixtures("adversarial_mp_logger") +@pytest.mark.xfail( + strict=True, + reason="DEFECT (crosslink #46): a raising store_record task propagates out " + "of the connection handler, which then closes the connection. Any client " + "still using that *shared* connection gets a BrokenPipeError on its next " + "send. In the real pipeline the TaskManager keeps a single long-lived " + "DataSocket for site_visits/crawl_history/finalize across ALL visits, so " + "one bad record can break the socket for the rest of the crawl.", +) +def test_raising_store_record_breaks_shared_connection() -> None: + """INVARIANT (currently violated): a single failing record must not tear + down a shared connection and break unrelated later records on it. + """ + handle = StorageControllerHandle(TransientFailingStoreProvider(fail_first=1), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "shared-conn") + + bad = VisitId(1101) + good = VisitId(1102) + sock.store_record(TableName("site_visits"), bad, {"site_url": "bad"}) + sock.finalize_visit_id(bad, success=False) + time.sleep(1) # let the raising store task run and close the connection + # Reusing the SAME connection: this send raises BrokenPipeError today. + sock.store_record(TableName("site_visits"), good, {"site_url": "good"}) + sock.finalize_visit_id(good, success=True) + sock.close() + + handle.shutdown() + seen = handle.get_new_completed_visits() + assert good in _completed_ids( + seen + ), f"good record on shared connection lost; seen={seen}" + + +# --------------------------------------------------------------------------- +# S1/S5 (storage side) - DEFECT: a raising store task strands the visit +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("adversarial_mp_logger") +@pytest.mark.xfail( + strict=True, + reason="DEFECT (crosslink #46): when a store_record task raises, the visit " + "is never enqueued to the completion_queue - not on finalize and not on " + "shutdown - because finalize_visit_id pops the tasks then propagates the " + "exception before the completion token is recorded. A callback-bearing " + "CommandSequence would hang forever and the visit is silently lost.", +) +def test_raising_store_record_visit_still_finalizes() -> None: + """INVARIANT (currently violated): a finalized visit whose store_record + raised must still reach the completion_queue (marked unsuccessful), not + vanish. + """ + handle = StorageControllerHandle(RaisingStoreProvider(), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "raising") + vid = VisitId(2001) + sock.store_record(TableName("site_visits"), vid, {"site_url": "x"}) + sock.finalize_visit_id(vid, success=True) + sock.close() + + seen = _drain_completions(handle) + handle.shutdown() + seen.extend(handle.get_new_completed_visits()) + assert vid in _completed_ids( + seen + ), f"visit stranded - never reached completion queue; seen={seen}" + + +@pytest.mark.usefixtures("adversarial_mp_logger") +@pytest.mark.xfail( + strict=True, + reason="DEFECT (crosslink #46): an *unfinalized* visit (browser crashed " + "mid-visit) whose store_record raised is never enqueued by shutdown's " + "finalize loop, because the raising task propagates out of " + "finalize_visit_id and aborts shutdown before completion_queue.put.", +) +def test_raising_store_record_unfinalized_visit_enqueued_on_shutdown() -> None: + """INVARIANT (currently violated): on shutdown, every visit with pending + store tasks must be enqueued to the completion_queue even if those tasks + raise. Models a browser dying mid-visit (no Finalize sent). + """ + handle = StorageControllerHandle(RaisingStoreProvider(), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "raising-unfinalized") + vid = VisitId(2002) + sock.store_record(TableName("site_visits"), vid, {"site_url": "x"}) + # deliberately NO finalize_visit_id - the client "crashes" + sock.close() + time.sleep(1) + + start = time.time() + handle.shutdown() + assert time.time() - start < 60, "shutdown hung on the raising visit" + seen = handle.get_new_completed_visits() + assert vid in _completed_ids( + seen + ), f"unfinalized stranded visit never enqueued on shutdown; seen={seen}" + + +# --------------------------------------------------------------------------- +# S4 (permanent) - DEFECT: a permanent write_table fault strands the visit +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("adversarial_mp_logger") +@pytest.mark.xfail( + strict=True, + reason="DEFECT (crosslink #46): a permanent write_table failure raises out " + "of flush_cache during shutdown, killing the controller before the " + "completion_queue is populated, so the visit is lost. write failures " + "should be surfaced/counted, not silently drop completed visits.", +) +def test_permanent_write_table_fault_visit_still_completes() -> None: + """INVARIANT (currently violated): even if persistent storage rejects the + write, the visit's terminal state must reach the completion_queue and + shutdown must not crash before draining it. + """ + handle = StorageControllerHandle(FailingWriteArrowProvider(), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "write-fault") + vid = VisitId(3001) + sock.store_record( + TableName("site_visits"), + vid, + {"site_url": "x", "browser_id": 1, "site_rank": 1}, + ) + sock.finalize_visit_id(vid, success=True) + sock.close() + + start = time.time() + handle.shutdown() + assert time.time() - start < 60, "shutdown hung on the write fault" + seen = handle.get_new_completed_visits() + assert vid in _completed_ids( + seen + ), f"visit lost on permanent write fault; seen={seen}" + + +# --------------------------------------------------------------------------- +# S6 - malformed / hostile records sent through the real DataSocket +# --------------------------------------------------------------------------- + + +@pytest.mark.usefixtures("adversarial_mp_logger") +def test_malformed_records_do_not_break_controller(tmp_path: Path) -> None: + """Hostile/malformed records (huge strings, injection-y values, a record + with no visit_id) must not crash or hang the controller; a subsequent good + visit must still complete and be persisted. + + Driven against the *real* ``SQLiteStorageProvider`` so the huge value goes + through the production persistence path (writing to disk), not the + ``MemoryStructuredProvider`` test queue - a multi-MiB value deadlocks the + cross-process ``multiprocess.Queue`` the memory provider uses, which is a + test-harness artifact, not a pipeline property. + + SQLite-schema column/table drops (e.g. unknown columns) are a separate, + known issue - see crosslink #28/#30 - so this test stays within valid + ``site_visits`` columns. + """ + db_path = tmp_path / "malformed.sqlite" + handle = StorageControllerHandle(SQLiteStorageProvider(db_path), None) + handle.launch() + assert handle.listener_address is not None + sock = DataSocket(handle.listener_address, "malformed") + + huge = "A" * (2 * 1024 * 1024) # 2 MiB string in a valid TEXT column + # site_visits.visit_id is the PRIMARY KEY, so each visit gets one record. + huge_visit = VisitId(4001) + sock.store_record( + TableName("site_visits"), + huge_visit, + {"browser_id": 1, "site_url": huge, "site_rank": 1}, + ) + sock.finalize_visit_id(huge_visit, success=True) + + # injection-y value as a literal in a valid column on its own visit + injection_visit = VisitId(4002) + sock.store_record( + TableName("site_visits"), + injection_visit, + {"browser_id": 1, "site_url": "'; DROP TABLE site_visits;--", "site_rank": 2}, + ) + sock.finalize_visit_id(injection_visit, success=True) + + # A record with no visit_id at all (controller must skip it, not crash). + # Sent via the raw socket because DataSocket always injects visit_id. + sock.socket.send((TableName("site_visits"), {"site_url": "no visit id"})) + + # Now a normal visit must still complete -> controller survived. + good = VisitId(4003) + sock.store_record( + TableName("site_visits"), + good, + {"browser_id": 1, "site_url": "ok", "site_rank": 3}, + ) + sock.finalize_visit_id(good, success=True) + sock.close() + + start = time.time() + handle.shutdown() + assert time.time() - start < 60, "controller hung on a malformed record" + + seen = handle.get_new_completed_visits() + ids = _completed_ids(seen) + assert good in ids, f"controller did not survive malformed records; seen={seen}" + assert huge_visit in ids, f"huge-value visit lost; seen={seen}" + assert injection_visit in ids, f"injection-value visit lost; seen={seen}" + + # The injection string is stored as a literal value, table intact. + rows = db_utils.query_db( + db_path, "SELECT site_url FROM site_visits;", as_tuple=True + ) + stored = {r[0] for r in rows} + assert ( + "'; DROP TABLE site_visits;--" in stored + ), "injection value not stored literally" + assert huge in stored, "huge value not persisted" diff --git a/test/test_adversarial_pipeline.py b/test/test_adversarial_pipeline.py new file mode 100644 index 000000000..2b199b238 --- /dev/null +++ b/test/test_adversarial_pipeline.py @@ -0,0 +1,253 @@ +"""Browser-required adversarial tests for the full crawl pipeline. + +These exercise the end-to-end recovery path through ``TaskManager`` -> +``BrowserManager`` -> Firefox + WebExtension, which the browser-free +storage-tier tests (``test/storage/test_adversarial_*.py``) cannot reach. + +PROPERTY UNDER TEST (graceful degradation): + For each adversarial scenario the crawl must + 1. make forward progress - the offending visit reaches a terminal state + and the crawl continues to the *next* site, + 2. record the offending visit as incomplete (``incomplete_visits`` row), + 3. not hang (the whole sequence completes within the per-command timeout + plus restart budget), + 4. preserve prior data (records from earlier good visits survive). + +ENVIRONMENT BLOCK + These tests need a real Firefox + the built extension xpi. In CI that is + provided by the install script (pinned, unbranded Firefox 152 build) and the + ``xpi`` fixture. Locally the workspace frequently lacks ``firefox-bin`` and + ``FIREFOX_BINARY`` (system Firefox is a different version and not wired in), + so these tests are *skipped* rather than reported as failures. They are NOT + faked: the assertions below run for real in CI. + +Scenarios: + * S1a custom command raises -> incomplete + continue + * S1b custom command hangs forever -> timeout + kill + incomplete + continue + * S5 custom command kills the browser -> watchdog restart + incomplete + continue + +A crashing *extension modification* (S2) shares the recovery shape with S5 +(the BrowserManager subprocess / Firefox dies and must be restarted); a +faithful S2 test requires building a deliberately-broken xpi and is documented +as a design item in the PR / crosslink #46 rather than implemented here, to +avoid shipping a broken extension build into the repo. +""" + +import os +import time + +import pytest +from selenium.webdriver import Firefox + +from openwpm import command_sequence +from openwpm.commands.types import BaseCommand +from openwpm.config import BrowserParamsInternal, ManagerParamsInternal +from openwpm.socket_interface import ClientSocket +from openwpm.utilities import db_utils + +from .conftest import FullConfig, TaskManagerCreator +from .utilities import ServerUrls + + +def _firefox_available() -> bool: + """True iff a Firefox binary OpenWPM can launch is resolvable.""" + if os.environ.get("FIREFOX_BINARY"): + return os.path.isfile(os.environ["FIREFOX_BINARY"]) + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + return os.path.isfile(os.path.join(root, "firefox-bin", "firefox-bin")) + + +requires_browser = pytest.mark.skipif( + not _firefox_available(), + reason="No launchable Firefox (firefox-bin / FIREFOX_BINARY missing). " + "Browser-required adversarial test; runs in CI where the pinned Firefox " + "and built xpi are installed.", +) + + +# --------------------------------------------------------------------------- +# Adversarial custom commands +# --------------------------------------------------------------------------- + + +class RaisingCommand(BaseCommand): + """A user custom command whose ``execute`` raises immediately.""" + + def __repr__(self) -> str: + return "RaisingCommand" + + def execute( + self, + webdriver: Firefox, + browser_params: BrowserParamsInternal, + manager_params: ManagerParamsInternal, + extension_socket: ClientSocket, + ) -> None: + raise RuntimeError("intentional crash inside a user custom command") + + +class HangingCommand(BaseCommand): + """A user custom command that never returns. + + The BrowserManager must hit the per-command timeout, kill the browser, mark + the visit incomplete, and let the crawl proceed. + """ + + def __repr__(self) -> str: + return "HangingCommand" + + def execute( + self, + webdriver: Firefox, + browser_params: BrowserParamsInternal, + manager_params: ManagerParamsInternal, + extension_socket: ClientSocket, + ) -> None: + while True: + time.sleep(1) + + +class BrowserKillingCommand(BaseCommand): + """A custom command that hard-kills the browser process from inside the + BrowserManager subprocess, standing in for a mid-visit browser crash (and, + by recovery shape, a crashing extension that takes the browser down). + """ + + def __repr__(self) -> str: + return "BrowserKillingCommand" + + def execute( + self, + webdriver: Firefox, + browser_params: BrowserParamsInternal, + manager_params: ManagerParamsInternal, + extension_socket: ClientSocket, + ) -> None: + # Terminate the geckodriver-controlled Firefox out from under Selenium. + pid = webdriver.service.process.pid + os.kill(pid, 9) + # Touch the driver so the BrowserManager observes a WebDriverException. + webdriver.get("about:blank") + + +# --------------------------------------------------------------------------- +# Shared assertion: offending visit incomplete, later visit good, no hang +# --------------------------------------------------------------------------- + + +def _run_adversarial_then_good( + task_manager_creator: TaskManagerCreator, + http_params: FullConfig, + server: ServerUrls, + adversarial_command: BaseCommand, + per_command_timeout: int = 30, + overall_budget: float = 240.0, +) -> None: + manager_params, browser_params = http_params + # Single browser so the "next site" must reuse the (restarted) browser. + manager_params.num_browsers = 1 + browser_params = browser_params[:1] + # Keep the failure limit high enough that one failure does not abort. + manager_params.failure_limit = 5 + + manager, db = task_manager_creator((manager_params, browser_params)) + + bad_url = server.base + "/simple_a.html" + good_url = server.base + "/simple_d.html" + + start = time.time() + + # Visit 1: trip the adversarial command. + cs_bad = command_sequence.CommandSequence(bad_url) + cs_bad.get(sleep=0, timeout=per_command_timeout) + cs_bad.append_command(adversarial_command, timeout=per_command_timeout) + manager.execute_command_sequence(cs_bad) + + # Visit 2: a plain good visit must still run after recovery. + cs_good = command_sequence.CommandSequence(good_url) + cs_good.get(sleep=0, timeout=per_command_timeout) + manager.execute_command_sequence(cs_good) + + manager.close() + + elapsed = time.time() - start + assert elapsed < overall_budget, ( + f"crawl did not make forward progress in time ({elapsed:.0f}s) - " + "likely a hang in the recovery path" + ) + + # Forward progress: both visits were recorded as site_visits. + visits = db_utils.query_db( + db, "SELECT site_url FROM site_visits ORDER BY visit_id;", as_tuple=True + ) + visited_urls = {row[0] for row in visits} + assert bad_url in visited_urls, "offending visit was never started/recorded" + assert good_url in visited_urls, "crawl did not continue to the next site" + + # The offending visit must be marked incomplete. + incomplete = db_utils.query_db( + db, "SELECT visit_id FROM incomplete_visits;", as_tuple=True + ) + assert len(incomplete) >= 1, ( + "offending visit was not recorded as incomplete - data loss / " + "missing incomplete-visit accounting" + ) + + # The good visit's get() command must have succeeded (no global poisoning). + statuses = db_utils.query_db( + db, + "SELECT command, command_status FROM crawl_history " + "WHERE command = 'GetCommand';", + as_tuple=True, + ) + assert any( + status == "ok" for _, status in statuses + ), "no GetCommand ever succeeded - the crawl never recovered" + + +@requires_browser +@pytest.mark.usefixtures("xpi") +def test_crashing_custom_command_visit_incomplete_and_crawl_continues( + task_manager_creator: TaskManagerCreator, + http_params: FullConfig, + server: ServerUrls, +) -> None: + """S1a: a custom command that raises -> visit incomplete, crawl continues.""" + _run_adversarial_then_good( + task_manager_creator, http_params, server, RaisingCommand() + ) + + +@requires_browser +@pytest.mark.slow +@pytest.mark.usefixtures("xpi") +def test_hanging_custom_command_times_out_and_crawl_continues( + task_manager_creator: TaskManagerCreator, + http_params: FullConfig, + server: ServerUrls, +) -> None: + """S1b: a custom command that hangs forever -> timeout + kill, then the + crawl recovers and the next site is visited.""" + _run_adversarial_then_good( + task_manager_creator, + http_params, + server, + HangingCommand(), + per_command_timeout=20, + ) + + +@requires_browser +@pytest.mark.slow +@pytest.mark.usefixtures("xpi") +def test_browser_killed_mid_visit_recovers_and_continues( + task_manager_creator: TaskManagerCreator, + http_params: FullConfig, + server: ServerUrls, +) -> None: + """S5 (and S2 by recovery shape): the browser process dies mid-visit -> + watchdog/BrowserManager restart, offending visit incomplete, crawl + continues.""" + _run_adversarial_then_good( + task_manager_creator, http_params, server, BrowserKillingCommand() + )