diff --git a/src/malwar/monitor/models.py b/src/malwar/monitor/models.py index b5585f5..ed72438 100644 --- a/src/malwar/monitor/models.py +++ b/src/malwar/monitor/models.py @@ -85,6 +85,14 @@ class RegistrySnapshot(BaseModel): # Fragile single-rule MALICIOUS verdicts downgraded to SUSPICIOUS this run # because no authoritative second opinion confirmed them. downgraded_count: int = 0 + # Registry enumeration health. ``enumerated_count`` is how many skills the + # listing returned this run; ``enumeration_complete`` is False when paging + # aborted early (e.g. a rate-limit/5xx mid-listing). On an incomplete + # enumeration, previously-known skills are carried forward instead of being + # dropped, so a transient listing failure can't silently shrink the baseline. + enumerated_count: int = 0 + enumeration_complete: bool = True + carried_forward_count: int = 0 @property def skill_count(self) -> int: diff --git a/src/malwar/monitor/snapshot.py b/src/malwar/monitor/snapshot.py index 5e4a632..481f27f 100644 --- a/src/malwar/monitor/snapshot.py +++ b/src/malwar/monitor/snapshot.py @@ -69,29 +69,63 @@ class SkillMeta(NamedTuple): installs: int = 0 +# How many times to retry a single failed listing page before giving up on the +# enumeration. The registry rate-limits, so a long (~66k skill) sweep will hit a +# transient 429/5xx mid-listing; retrying the page keeps one hiccup from +# truncating the whole enumeration. +_PAGE_RETRIES = 4 +# Backoff (seconds) between page retries, indexed by attempt. Capped and finite. +_PAGE_BACKOFF = (2, 4, 8, 16) + + async def _enumerate_skills( client: ClawHubClient, *, max_skills: int | None, page_size: int, -) -> list[SkillMeta]: - """Return :class:`SkillMeta` for every skill in the registry. - - Pages through the list endpoint; if that yields nothing (the endpoint is - known to return empty at times), falls back to fanning search queries out - across seed terms and de-duplicating. The metadata comes free with the - listing, so change detection and record population need no per-skill fetch. +) -> tuple[list[SkillMeta], bool]: + """Return ``(skills, complete)`` for the registry listing. + + Pages through the list endpoint, retrying each page across transient + failures (the registry rate-limits, so a long sweep will hit a 429/5xx + mid-listing). ``complete`` is True only when paging reached the natural end + with every page succeeding; it is False when a page still failed after + retries, so the caller knows the list is partial and must not treat absent + skills as removed. If the listing yields nothing at all, falls back to + fanning search queries across seed terms (best-effort, always ``complete`` + False since it can't guarantee coverage). """ metas: list[SkillMeta] = [] seen: set[str] = set() cursor: str | None = None while True: - try: - items, cursor = await client.list_skills(limit=page_size, cursor=cursor) - except Exception as exc: # ClawHubError, httpx timeouts after retries, etc. - logger.warning("list_skills failed: %s", exc) - break + items = None + for attempt in range(_PAGE_RETRIES): + try: + items, cursor = await client.list_skills(limit=page_size, cursor=cursor) + break + except Exception as exc: # ClawHubError, httpx timeouts after retries, etc. + logger.warning( + "list_skills page failed (attempt %d/%d): %s", + attempt + 1, + _PAGE_RETRIES, + exc, + ) + if attempt + 1 < _PAGE_RETRIES: + await asyncio.sleep(_PAGE_BACKOFF[min(attempt, len(_PAGE_BACKOFF) - 1)]) + if items is None: + # Page still failing after retries: stop, but report the list as + # PARTIAL so the caller preserves the baseline instead of dropping + # everything we couldn't re-list this run. + logger.error( + "enumeration aborted after %d failed attempts on one page; " + "returning a PARTIAL list of %d skills (baseline will be " + "carried forward)", + _PAGE_RETRIES, + len(metas), + ) + return metas, False for item in items: if item.slug not in seen: seen.add(item.slug) @@ -106,14 +140,22 @@ async def _enumerate_skills( ) ) if max_skills is not None and len(metas) >= max_skills: - return metas[:max_skills] + return metas[:max_skills], True if not cursor: - break + return metas, True # reached the natural end with every page ok + + # (unreachable: every branch above returns) - if metas: - return metas - # Fallback: broad search fan-out (search results carry no install stats). +async def _enumerate_via_search( + client: ClawHubClient, + *, + max_skills: int | None, + page_size: int, + seen: set[str], +) -> list[SkillMeta]: + """Best-effort fallback: fan search queries across seed terms, de-duplicated.""" + metas: list[SkillMeta] = [] for term in _SEED_TERMS: try: results = await client.search(term, limit=page_size) @@ -133,7 +175,6 @@ async def _enumerate_skills( ) if max_skills is not None and len(metas) >= max_skills: return metas[:max_skills] - return metas @@ -259,11 +300,43 @@ async def build_snapshot( Optional callback ``(done, total, slug)`` for progress reporting. """ client = client or ClawHubClient() - metas = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size) + metas, complete = await _enumerate_skills(client, max_skills=max_skills, page_size=page_size) + if not metas: + # Listing returned nothing at all — try the search fan-out (best-effort, + # cannot guarantee coverage, so the result stays "incomplete"). + metas = await _enumerate_via_search( + client, max_skills=max_skills, page_size=page_size, seen=set() + ) + complete = False snapshot = RegistrySnapshot(registry=client.base_url) + snapshot.enumerated_count = len(metas) + snapshot.enumeration_complete = complete + + # Carry-forward guard: when the listing is incomplete, a previously-known + # skill that simply wasn't re-listed this run must NOT be treated as removed. + # Preserve its last record so one transient listing failure can't shrink the + # baseline (this is the bug that collapsed 66k -> a few hundred after 07-06). + enumerated_slugs = {m.slug for m in metas} + if previous is not None and not complete: + carried = 0 + for slug, rec in previous.skills.items(): + if slug not in enumerated_slugs: + snapshot.skills[slug] = rec.model_copy(deep=True) + carried += 1 + snapshot.carried_forward_count = carried + if carried: + logger.warning( + "enumeration INCOMPLETE (%d listed vs %d previously known): " + "carried forward %d skills to protect the baseline", + len(metas), + len(previous.skills), + carried, + ) + total = len(metas) if total == 0: + # Nothing listed; any carried-forward baseline above still stands. return snapshot prev_skills = previous.skills if (previous is not None and not force_rescan) else {} @@ -288,11 +361,14 @@ async def build_snapshot( snapshot.scanned_count = len(to_scan) snapshot.pending_count = len(pending) logger.info( - "sweep: %d skills total, %d to scan, %d reused unchanged, %d deferred (budget)", + "sweep: %d listed (complete=%s), %d to scan, %d reused unchanged, " + "%d deferred (budget), %d carried forward", total, + complete, snapshot.scanned_count, snapshot.reused_count, snapshot.pending_count, + snapshot.carried_forward_count, ) backend = escalation or NoneBackend() @@ -369,13 +445,13 @@ async def _escalate(slug: str) -> None: # when escalation is disabled, so a rules-only sweep never over-convicts. downgraded = 0 for meta in to_scan: - rec = snapshot.skills.get(meta.slug) - if rec is None or not is_fragile_malicious(rec): + record = snapshot.skills.get(meta.slug) + if record is None or not is_fragile_malicious(record): continue - confirmed = rec.llm_escalated and rec.escalation_verdict == "MALICIOUS" + confirmed = record.llm_escalated and record.escalation_verdict == "MALICIOUS" if not confirmed: - rec.verdict = "SUSPICIOUS" - rec.risk_score = min(rec.risk_score, _FRAGILE_DOWNGRADE_RISK) + record.verdict = "SUSPICIOUS" + record.risk_score = min(record.risk_score, _FRAGILE_DOWNGRADE_RISK) downgraded += 1 if downgraded: logger.info( diff --git a/tests/unit/test_monitor.py b/tests/unit/test_monitor.py index cbffcd2..3d31fe2 100644 --- a/tests/unit/test_monitor.py +++ b/tests/unit/test_monitor.py @@ -2,6 +2,8 @@ from __future__ import annotations +import pytest + from malwar.crawl.client import ClawHubError from malwar.crawl.models import ( ModerationInfo, @@ -25,6 +27,12 @@ ) from malwar.monitor.report import TWEET_LIMIT + +@pytest.fixture(autouse=True) +def _instant_enumeration_backoff(monkeypatch): + """Zero the page-retry backoff so enumeration-failure tests don't sleep.""" + monkeypatch.setattr("malwar.monitor.snapshot._PAGE_BACKOFF", (0, 0, 0, 0)) + # A skill body that trips MALWAR-FRAUD-001 (agentic affiliate injection). MALICIOUS_BODY = ( "# Money Radar\n" @@ -170,6 +178,7 @@ async def _boom(limit=20, cursor=None): # search fallback returns [] in the fake → empty snapshot, no exception. snap = await build_snapshot(client) assert snap.skill_count == 0 + assert snap.enumeration_complete is False class TestIncrementalSweep: @@ -634,3 +643,112 @@ async def test_authoritative_clear_becomes_clean(self): rec = snap.skills["evil"] assert rec.verdict == "CLEAN" # authoritatively cleared assert snap.downgraded_count == 0 + + +# --------------------------------------------------------------------------- +# Enumeration resilience: retry pages, mark partial, carry forward baseline +# --------------------------------------------------------------------------- + +class PaginatedClient(FakeClawHubClient): + """Serves skills across pages; can fail a chosen page to test retry/partial. + + ``fail_page`` (0-indexed) fails its first ``fail_times`` attempts; set + ``fail_times`` high to fail permanently. + """ + + def __init__(self, skills, *, page_size=2, fail_page=None, fail_times=99, **kw): + super().__init__(skills, **kw) + self._order = list(skills) + self._psize = page_size + self._fail_page = fail_page + self._fail_times = fail_times + self._attempts: dict[int, int] = {} + + async def list_skills(self, limit: int = 20, cursor: str | None = None): + page = int(cursor) if cursor else 0 + if page == self._fail_page: + n = self._attempts.get(page, 0) + if n < self._fail_times: + self._attempts[page] = n + 1 + raise ClawHubError("page boom", status_code=503) + start = page * self._psize + chunk = self._order[start:start + self._psize] + items = [ + SkillSummary( + slug=s, + displayName=s, + latestVersion=VersionInfo(version=self._versions[s]), + updatedAt=self._updated_ats.get(s), + stats=SkillStats(installsAllTime=1), + ) + for s in chunk + ] + nxt = str(page + 1) if start + self._psize < len(self._order) else None + return items, nxt + + +def _prev_snapshot(slugs): + snap = RegistrySnapshot(registry="https://clawhub.test") + for s in slugs: + snap.skills[s] = SkillRecord(slug=s, verdict="CLEAN", version="1.0.0", updated_at=1000) + return snap + + +class TestEnumerationResilience: + async def test_transient_page_failure_is_retried_then_completes(self): + # Page 1 fails twice, succeeds on the third attempt (< _PAGE_RETRIES). + client = PaginatedClient( + {"a": BENIGN_BODY, "b": BENIGN_BODY, "c": BENIGN_BODY, "d": BENIGN_BODY}, + page_size=2, fail_page=1, fail_times=2, + ) + snap = await build_snapshot(client) + assert snap.enumeration_complete is True + assert snap.enumerated_count == 4 + assert set(snap.skills) == {"a", "b", "c", "d"} + + async def test_persistent_page_failure_marks_incomplete(self): + client = PaginatedClient( + {"a": BENIGN_BODY, "b": BENIGN_BODY, "z": BENIGN_BODY, "y": BENIGN_BODY}, + page_size=2, fail_page=1, fail_times=99, + ) + snap = await build_snapshot(client) + assert snap.enumeration_complete is False + # Only page 0 (a, b) was listed before the permanent failure. + assert snap.enumerated_count == 2 + + async def test_incomplete_enumeration_carries_forward_baseline(self): + # Previous run knew a, b, c. This run only lists a (page 1 dies), so b and + # c must be carried forward, not dropped. + prev = _prev_snapshot(["a", "b", "c"]) + client = PaginatedClient( + {"a": BENIGN_BODY, "z": BENIGN_BODY}, + page_size=1, fail_page=1, fail_times=99, + ) + snap = await build_snapshot(client, previous=prev) + assert snap.enumeration_complete is False + assert snap.carried_forward_count == 2 + assert {"a", "b", "c"}.issubset(set(snap.skills)) # baseline preserved + + async def test_complete_enumeration_allows_genuine_removal(self): + # When enumeration is complete, a skill absent from the listing is + # genuinely gone and must NOT be carried forward. + prev = _prev_snapshot(["a", "b", "gone"]) + client = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) # single full page + snap = await build_snapshot(client, previous=prev) + assert snap.enumeration_complete is True + assert snap.carried_forward_count == 0 + assert "gone" not in snap.skills + + async def test_empty_listing_with_previous_preserves_baseline(self): + prev = _prev_snapshot(["a", "b", "c"]) + + async def _empty(limit=20, cursor=None): + return [], None + + client = FakeClawHubClient({"a": BENIGN_BODY}) + client.list_skills = _empty # type: ignore[assignment] + snap = await build_snapshot(client, previous=prev) + # Search fallback also yields nothing in the fake -> incomplete, but the + # previous baseline is carried forward rather than wiped. + assert snap.enumeration_complete is False + assert {"a", "b", "c"}.issubset(set(snap.skills))