test: adversarial robustness suite for the crawl pipeline

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.

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()

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}"