diff --git a/docs/crawl.md b/docs/crawl.md index baf0825..e171ad7 100644 --- a/docs/crawl.md +++ b/docs/crawl.md @@ -105,16 +105,29 @@ Sweep the registry, fast-scan skills (rule engine + threat intel, escalating fla **Budgeted, resumable baseline.** ClawHub is rate-limited (~120 req/min), so a full ~6k-skill registry can't be swept in one shot. `--max-scans N` caps how many skills are actually scanned per run; any overflow is recorded as an `UNKNOWN` placeholder and picked up on the next run. So the **first full baseline builds up over several runs** rather than dying to a timeout, and once complete, daily incremental runs only touch what changed. `scanned_count`/`reused_count`/`pending_count` on each snapshot show the split. +**Targeted, tiered escalation.** The rules pass is confident at the two ends (obviously clean, obviously malicious) but there's an **ambiguous middle** where a sneaky skill hides — it scores just under the line. Rather than pay for a deep look at *every* flagged skill, an escalation **policy** selects only that middle band (rule risk ~8–74, below the confident-malicious line), ranks by suspicion, and caps it with `--escalate-budget`. A pluggable **backend** then gives those a second opinion: + +| `--escalate-backend` | What runs | Cost | +|---|---|---| +| `none` | nothing (policy still records candidates) | free | +| `hf` | a free local Hugging Face classifier (CPU) — `pip install malwar[hf]` | free | +| `anthropic` | the full LLM analyzer (rules + LLM + false-positive suppression) | paid API | +| `tiered` | `hf` triages first; escalates to `anthropic` **only** on a hit | paid on the residual only | + +`tiered` is the money-saver: the free tier filters the band, and the LLM is spent only on what the free tier still finds suspicious. An `anthropic`/`tiered` result is *authoritative* — it can raise a sneaky skill or clear a rule-engine false positive; an `hf` result is recorded as a triage signal (`escalation_backend`/`escalation_verdict`/`escalation_score` on the snapshot) without overriding the rule verdict. `escalated_count` shows how many skills were sent to the backend. + +> **Note:** a "rule-clean but ML-anomalous" escalation path exists (`EscalationPolicy.ml_threshold`) but is **off by default** — the stock ML scorer saturates near 1.0 for both benign and malicious skills, so it can't discriminate yet. Enable it only once the model is recalibrated. + ```bash -malwar crawl monitor # incremental sweep -> snapshot -> diff -malwar crawl monitor --full # re-scan every skill (catches same-version tampering) -malwar crawl monitor --digest # also print a shareable digest + draft post -malwar crawl monitor --publish # post the digest to X when skills newly turn malicious -malwar crawl monitor --fail-on-malicious # exit non-zero when newly-flagged skills are found (CI/cron) -malwar crawl monitor --max 100 --no-escalate # quick partial run, rules only +malwar crawl monitor # incremental sweep -> snapshot -> diff (LLM on the ambiguous band) +malwar crawl monitor --no-escalate # rules only, no second opinion (fast, free) +malwar crawl monitor --escalate-backend hf # free local classifier on the ambiguous band +malwar crawl monitor --escalate-backend tiered --escalate-budget 50 # hf triage -> LLM on hits, capped +malwar crawl monitor --full # re-scan every skill (catches same-version tampering) +malwar crawl monitor --fail-on-malicious # exit non-zero when newly-flagged skills are found (CI/cron) ``` -**Options:** `--snapshot-dir`, `--full`, `--max-scans`, `--max`, `--no-escalate`, `--concurrency`, `--format`/`-f` (`console`|`json`), `--output`/`-o`, `--no-save`, `--digest`, `--publish`, `--fail-on-malicious`. Publishing to X requires the `MALWAR_X_*` credentials (see [Configuration](deployment/configuration.md#x-twitter-publishing)). +**Options:** `--snapshot-dir`, `--full`, `--max-scans`, `--max`, `--escalate-backend` (`none`|`hf`|`anthropic`|`tiered`), `--escalate-budget`, `--no-escalate`, `--concurrency`, `--format`/`-f` (`console`|`json`), `--output`/`-o`, `--no-save`, `--digest`, `--publish`, `--fail-on-malicious`. Publishing to X requires the `MALWAR_X_*` credentials (see [Configuration](deployment/configuration.md#x-twitter-publishing)). The bundled GitHub Actions workflow (`.github/workflows/registry-monitor.yml`) runs this on two cadences: **daily incremental** and a **weekly `--full`** re-scan, committing each snapshot to the `registry-snapshots` branch. It runs **rules-only (`--no-escalate`)** so a whole-registry sweep stays fast and cheap — the rule engine alone catches every malicious sample in the benchmark; use `malwar crawl scan ` for an LLM deep-dive on an individual flagged skill. diff --git a/pyproject.toml b/pyproject.toml index 185d0a6..319ac90 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,12 @@ postgres = [ cache = [ "redis[hiredis]>=5.0.0", ] +hf = [ + # Free, local second-opinion escalation tier (monitor.escalation). Heavy — + # only needed for `malwar crawl monitor --escalate-backend hf|tiered`. + "transformers>=4.44.0", + "torch>=2.2.0", +] dev = [ "pytest>=8.3.0", "pytest-asyncio>=0.24.0", @@ -115,6 +121,11 @@ module = ["frontmatter", "frontmatter.*", "aiofiles", "aiofiles.*"] ignore_missing_imports = true follow_untyped_imports = true +[[tool.mypy.overrides]] +# Optional 'hf' extra — imported lazily in monitor.escalation. +module = ["transformers"] +ignore_missing_imports = true + [tool.pytest.ini_options] testpaths = ["tests"] asyncio_mode = "auto" diff --git a/src/malwar/cli/commands/crawl.py b/src/malwar/cli/commands/crawl.py index 3c15b46..9bf23b2 100644 --- a/src/malwar/cli/commands/crawl.py +++ b/src/malwar/cli/commands/crawl.py @@ -478,8 +478,24 @@ def crawl_monitor( ] = None, no_escalate: Annotated[ bool, - typer.Option("--no-escalate", help="Never escalate flagged skills to the LLM"), + typer.Option("--no-escalate", help="Disable escalation (shorthand for --escalate-backend none)"), ] = False, + escalate_backend: Annotated[ + str, + typer.Option( + "--escalate-backend", + help="Second-opinion backend for the ambiguous band: " + "none | hf (free local classifier) | anthropic (LLM) | " + "tiered (hf triage -> anthropic on hits)", + ), + ] = "anthropic", + escalate_budget: Annotated[ + int | None, + typer.Option( + "--escalate-budget", + help="Cap how many skills are escalated per run (most-suspicious first)", + ), + ] = None, concurrency: Annotated[ int, typer.Option("--concurrency", help="Skills scanned in parallel") ] = 8, @@ -547,7 +563,8 @@ def crawl_monitor( _async_monitor( snapshot_dir=snapshot_dir, max_skills=max_skills, - escalate=not no_escalate, + escalate_backend="none" if no_escalate else escalate_backend, + escalate_budget=escalate_budget, concurrency=concurrency, fmt=fmt, output=output, @@ -566,7 +583,8 @@ async def _async_monitor( *, snapshot_dir: Path, max_skills: int | None, - escalate: bool, + escalate_backend: str, + escalate_budget: int | None, concurrency: int, fmt: CrawlFormat, output: Path | None, @@ -579,11 +597,24 @@ async def _async_monitor( ) -> int: from rich.progress import BarColumn, Progress, TextColumn - from malwar.monitor import SnapshotStore, build_snapshot, diff_snapshots + from malwar.monitor import ( + EscalationPolicy, + SnapshotStore, + build_snapshot, + diff_snapshots, + make_backend, + ) store = SnapshotStore(snapshot_dir) previous = store.load_latest() + try: + backend = make_backend(escalate_backend) + except ValueError as exc: + console.print(f"[red]{exc}[/red]") + return 2 + policy = EscalationPolicy(budget=escalate_budget) + mode = "full re-scan" if full else ("incremental" if previous else "first run") scope = f" (max {max_skills})" if max_skills else "" console.print(f"Crawling ClawHub registry{scope} [{mode}]...", style="dim") @@ -607,7 +638,8 @@ def _on_progress(done: int, total: int, slug: str) -> None: force_rescan=full, max_skills=max_skills, scan_budget=max_scans, - escalate=escalate, + escalation=backend, + escalation_policy=policy, concurrency=concurrency, on_progress=_on_progress, ) @@ -622,6 +654,8 @@ def _on_progress(done: int, total: int, slug: str) -> None: ) if snapshot.pending_count: summary += f", deferred {snapshot.pending_count} to next run" + if snapshot.escalated_count: + summary += f", escalated {snapshot.escalated_count} to '{backend.name}'" console.print(summary + ".[/dim]") diff = diff_snapshots(previous, snapshot) diff --git a/src/malwar/monitor/__init__.py b/src/malwar/monitor/__init__.py index b09a6bf..9229c9a 100644 --- a/src/malwar/monitor/__init__.py +++ b/src/malwar/monitor/__init__.py @@ -4,6 +4,12 @@ from __future__ import annotations from malwar.monitor.diff import diff_snapshots +from malwar.monitor.escalation import ( + EscalationPolicy, + EscalationResult, + make_backend, + select_candidates, +) from malwar.monitor.models import ( RegistrySnapshot, SkillChange, @@ -14,6 +20,8 @@ from malwar.monitor.snapshot import SnapshotStore, build_snapshot __all__ = [ + "EscalationPolicy", + "EscalationResult", "RegistrySnapshot", "SkillChange", "SkillRecord", @@ -21,6 +29,8 @@ "SnapshotStore", "build_snapshot", "diff_snapshots", + "make_backend", "render_digest", "render_tweet", + "select_candidates", ] diff --git a/src/malwar/monitor/escalation.py b/src/malwar/monitor/escalation.py new file mode 100644 index 0000000..6aadd38 --- /dev/null +++ b/src/malwar/monitor/escalation.py @@ -0,0 +1,304 @@ +"""Targeted escalation for the registry monitor. + +The registry sweep scans every skill with the rule engine + threat intel (fast, +free, 100% recall on the benchmark). That first pass is confident at the two +ends — obviously clean and obviously malicious — but there is an **ambiguous +middle** where a sneaky skill can hide: it scores just under the line, or looks +clean to the rules while the ML model finds it anomalous. + +Spending a deeper (paid/slow) analysis on *every* flagged skill is wasteful — +most of that spend lands on skills already decided. This module picks only the +skills where a second opinion can actually change the verdict, and runs a +pluggable **backend** on them: + +* ``none`` — no second opinion (the policy still marks candidates). +* ``hf`` — a free, local Hugging Face classifier (CPU, no API cost). +* ``anthropic`` — the full LLM analyzer (highest quality, paid). +* ``tiered`` — ``hf`` triage first, escalating to ``anthropic`` *only* the + skills the free tier still finds suspicious. Money is spent + on the residual, not the whole band. + +The policy is a pure function of the first-pass record, so it is cheap and +fully testable; the backends share one small async ``assess`` interface. +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Protocol + +from pydantic import BaseModel, Field + +from malwar.monitor.models import SkillRecord + +logger = logging.getLogger("malwar.monitor.escalation") + +# Rule risk at/above which a skill is already a confident detection — a deeper +# look would not change the verdict, so we don't spend on it. (Mirrors the +# MALICIOUS threshold in ScanResult.verdict.) +_MALICIOUS_RISK = 75 + + +@dataclass(frozen=True) +class EscalationPolicy: + """Decides which first-pass records deserve a deeper second opinion. + + A skill qualifies when it sits in the ambiguous middle — either the rule + engine flagged *something* short of a confident verdict, or the rules were + quiet but the ML model finds the skill anomalous ("looks clean but off"). + Confident-clean and confident-malicious skills are skipped: a second opinion + can't change their verdict, so spending on them is waste. + """ + + # Rule risk at/above which the rules already saw enough to warrant a look. + # Default 8 is below the CAUTION threshold (15) so we also catch + # "just-under-the-line" skills a sneaky author tuned to stay clean. + min_rule_risk: int = 8 + # Rule risk at/above which the skill is already a confident detection. + malicious_risk: int = _MALICIOUS_RISK + # ML anomaly probability at/above which a rule-clean skill is still worth a + # look (the "sneaky clean" path). Disabled by default (None): the current + # RiskScorer saturates near 1.0 for *both* benign and malicious skills, so + # it can't yet discriminate — turning it on would escalate everything. Once + # the model is recalibrated to separate the classes, set e.g. 0.6. + ml_threshold: float | None = None + # Cap on how many skills to escalate per run (most-suspicious first). None + # means no cap. Bounds the per-run cost of a paid backend. + budget: int | None = None + + def qualifies(self, record: SkillRecord) -> bool: + """True if this first-pass record sits in the ambiguous band.""" + if record.error is not None or record.verdict == "UNKNOWN": + return False # not scanned yet / failed — nothing to second-guess + if record.risk_score >= self.malicious_risk: + return False # already a confident detection + if record.risk_score >= self.min_rule_risk: + return True # rules saw something short of a verdict + # Rules were quiet — escalate only if the ML model is suspicious. + return ( + self.ml_threshold is not None + and record.ml_risk_score is not None + and record.ml_risk_score >= self.ml_threshold + ) + + def suspicion(self, record: SkillRecord) -> float: + """A 0-1 rank for ordering candidates most-suspicious first.""" + rule = record.risk_score / 100.0 + ml = record.ml_risk_score or 0.0 + return max(rule, ml) + + +def select_candidates( + records: Mapping[str, SkillRecord], + policy: EscalationPolicy, +) -> list[str]: + """Return the slugs to escalate, most-suspicious first, capped by budget.""" + candidates = [slug for slug, rec in records.items() if policy.qualifies(rec)] + candidates.sort(key=lambda slug: records[slug].risk_score, reverse=False) # stable base + candidates.sort(key=lambda slug: policy.suspicion(records[slug]), reverse=True) + if policy.budget is not None: + candidates = candidates[: policy.budget] + return candidates + + +class EscalationResult(BaseModel): + """The outcome of a backend's deeper look at one skill.""" + + backend: str + flagged: bool = False + verdict: str = "" + score: float | None = Field(default=None, description="Backend confidence [0,1]") + # True when this came from a full re-scan (rules + LLM + suppression), whose + # verdict is authoritative and should replace the first-pass verdict. False + # for a narrow classifier whose flag is only a triage signal. + authoritative: bool = False + detail: str = "" + + +class EscalationBackend(Protocol): + """A deeper second-opinion analyzer for an ambiguous skill.""" + + name: str + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: ... + + +class NoneBackend: + """No second opinion — the policy still selects candidates for reporting.""" + + name = "none" + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + return EscalationResult(backend=self.name, flagged=False, verdict="") + + +class AnthropicBackend: + """Full LLM analyzer (paid) — reruns the pipeline with the LLM layer on.""" + + name = "anthropic" + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + from malwar.sdk import scan + + result = await scan(content, file_name=file_name, use_llm=True, use_urls=True) + return EscalationResult( + backend=self.name, + flagged=result.verdict != "CLEAN", + verdict=result.verdict, + score=result.risk_score / 100.0, + authoritative=True, # full pipeline verdict (incl. LLM suppression) + detail=f"risk={result.risk_score}", + ) + + +# Default Hugging Face model for the free tier. A small prompt-injection / +# jailbreak classifier that runs on CPU. Overridable; pin a verified model id +# once confirmed against the Hub. +DEFAULT_HF_MODEL = "protectai/deberta-v3-base-prompt-injection-v2" + +# Truncate content before classification — these models have a 512-token window, +# and the malicious intent is virtually always in the instructions up top. +_HF_MAX_CHARS = 4000 + + +class HfClassifierBackend: + """Free, local Hugging Face text-classifier tier (no API cost). + + Loads a small sequence-classification model via ``transformers`` and scores + the content as malicious/benign. The model is loaded lazily and cached; if + ``transformers`` (or the model) is unavailable, ``assess`` degrades to an + ``unavailable`` result rather than raising, so it can never break a sweep. + + A ``classifier`` callable may be injected (tests, or a custom scorer): it + takes the text and returns ``(label, score)`` with ``score`` in [0, 1]. + """ + + name = "hf" + + def __init__( + self, + model: str = DEFAULT_HF_MODEL, + *, + threshold: float = 0.5, + malicious_labels: tuple[str, ...] = ("INJECTION", "MALICIOUS", "JAILBREAK", "LABEL_1"), + classifier: object | None = None, + ) -> None: + self.model = model + self.threshold = threshold + self._malicious_labels = {label.upper() for label in malicious_labels} + self._classifier = classifier # injected or lazily built pipeline + self._load_failed = False + + def _get_classifier(self) -> object | None: + if self._classifier is not None: + return self._classifier + if self._load_failed: + return None + try: # pragma: no cover - exercised only when transformers is installed + from transformers import pipeline as hf_pipeline + + self._classifier = hf_pipeline( + "text-classification", model=self.model, truncation=True + ) + return self._classifier + except Exception as exc: + self._load_failed = True + logger.warning( + "HF classifier unavailable (%s); install the 'hf' extra " + "(pip install malwar[hf]) to enable the free escalation tier.", + exc, + ) + return None + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + clf = self._get_classifier() + if clf is None: + return EscalationResult( + backend=self.name, flagged=False, verdict="", detail="unavailable" + ) + + try: + out = clf(content[:_HF_MAX_CHARS]) # type: ignore[operator] + except Exception as exc: + logger.warning("HF classification failed for %s: %s", file_name, exc) + return EscalationResult( + backend=self.name, flagged=False, verdict="", detail=f"error: {exc}" + ) + + label, score = _parse_classifier_output(out) + malicious = label.upper() in self._malicious_labels + flagged = malicious and score >= self.threshold + return EscalationResult( + backend=self.name, + flagged=flagged, + verdict="SUSPICIOUS" if flagged else "CLEAN", + score=score if malicious else (1.0 - score), + detail=f"{label}={score:.2f}", + ) + + +def _parse_classifier_output(out: object) -> tuple[str, float]: + """Normalize a transformers text-classification result to (label, score). + + Accepts ``[{"label": ..., "score": ...}]`` (the common shape), a bare dict, + or a list of per-label scores (picks the argmax). + """ + item: object = out + if isinstance(out, list) and out: + # Could be [{label,score}] or [[{label,score}, ...]] (return_all_scores). + first = out[0] + if isinstance(first, list) and first: + item = max(first, key=lambda d: d.get("score", 0.0)) + else: + item = first + if isinstance(item, dict): + return str(item.get("label", "")), float(item.get("score", 0.0)) + return "", 0.0 + + +class TieredBackend: + """Run a cheap backend first, escalating to an expensive one only on a hit. + + This is the money-saver: the free ``hf`` tier triages the ambiguous band, + and the paid ``anthropic`` tier is invoked *only* for skills the free tier + still flags — so paid spend tracks the residual, not the whole band. + """ + + name = "tiered" + + def __init__(self, cheap: EscalationBackend, expensive: EscalationBackend) -> None: + self.cheap = cheap + self.expensive = expensive + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + first = await self.cheap.assess(content, file_name=file_name) + if not first.flagged: + return first + second = await self.expensive.assess(content, file_name=file_name) + second.detail = f"{self.cheap.name}->{self.expensive.name}; {second.detail}".strip() + return second + + +def make_backend( + name: str, + *, + hf_model: str = DEFAULT_HF_MODEL, +) -> EscalationBackend: + """Build an escalation backend by name. + + ``none`` | ``hf`` | ``anthropic`` | ``tiered`` (hf -> anthropic). + """ + key = (name or "none").strip().lower() + if key in ("none", "off", ""): + return NoneBackend() + if key == "hf": + return HfClassifierBackend(model=hf_model) + if key == "anthropic": + return AnthropicBackend() + if key == "tiered": + return TieredBackend(HfClassifierBackend(model=hf_model), AnthropicBackend()) + raise ValueError( + f"unknown escalation backend {name!r} (expected none|hf|anthropic|tiered)" + ) diff --git a/src/malwar/monitor/models.py b/src/malwar/monitor/models.py index e4a7bba..4aef3f1 100644 --- a/src/malwar/monitor/models.py +++ b/src/malwar/monitor/models.py @@ -43,7 +43,16 @@ class SkillRecord(BaseModel): risk_score: int = 0 finding_rule_ids: list[str] = Field(default_factory=list) installs: int = 0 + # ML anomaly probability [0,1] from the first (rules) pass — a cheap signal + # for "looks clean but off" skills that drives targeted escalation. + ml_risk_score: float | None = None llm_escalated: bool = False + # Targeted escalation outcome (see monitor.escalation). Which second-opinion + # backend ran on this skill, its verdict, and its confidence — empty when the + # skill wasn't in the ambiguous band or escalation was disabled. + escalation_backend: str = "" + escalation_verdict: str = "" + escalation_score: float | None = None moderation_blocked: bool = False moderation_suspicious: bool = False scanned_at: str = Field( @@ -71,6 +80,8 @@ class RegistrySnapshot(BaseModel): scanned_count: int = 0 reused_count: int = 0 pending_count: int = 0 + # Skills sent to a second-opinion escalation backend this run. + escalated_count: int = 0 @property def skill_count(self) -> int: diff --git a/src/malwar/monitor/snapshot.py b/src/malwar/monitor/snapshot.py index 22f3eb7..fafdb4d 100644 --- a/src/malwar/monitor/snapshot.py +++ b/src/malwar/monitor/snapshot.py @@ -28,6 +28,12 @@ from typing import NamedTuple from malwar.crawl.client import ClawHubClient +from malwar.monitor.escalation import ( + EscalationBackend, + EscalationPolicy, + NoneBackend, + select_candidates, +) from malwar.monitor.models import RegistrySnapshot, SkillRecord from malwar.sdk import scan @@ -137,11 +143,16 @@ def _is_unchanged(prev: SkillRecord | None, version: str | None, updated_at: int return prev.version == version and prev.updated_at == updated_at -async def _scan_slug(client: ClawHubClient, meta: SkillMeta, *, escalate: bool) -> SkillRecord: - """Fetch and scan a single skill, escalating to the LLM only if flagged. +async def _scan_slug( + client: ClawHubClient, meta: SkillMeta +) -> tuple[SkillRecord, str | None]: + """Fetch and rule-scan a single skill (the fast, free first pass). Metadata (version, updated_at, display name, installs) comes from the enumeration listing, so this makes just one request per skill — the file. + Returns the record and the fetched content (or ``None`` on error) so a + later escalation pass can re-analyze without re-fetching. Escalation itself + is a separate, targeted phase (see :mod:`malwar.monitor.escalation`). """ record = SkillRecord( slug=meta.slug, @@ -155,24 +166,22 @@ async def _scan_slug(client: ClawHubClient, meta: SkillMeta, *, escalate: bool) content = await client.get_skill_file(meta.slug) except Exception as exc: # ClawHubError, httpx timeouts, etc. record.error = f"file: {exc}" - return record + return record, None record.content_sha256 = hashlib.sha256(content.encode("utf-8")).hexdigest() file_name = f"{meta.slug}/SKILL.md" try: result = await scan(content, file_name=file_name, use_llm=False, use_urls=False) - if escalate and result.verdict != "CLEAN": - result = await scan(content, file_name=file_name, use_llm=True, use_urls=True) - record.llm_escalated = True except Exception as exc: # a single bad skill must not kill the whole sweep record.error = f"scan: {exc}" - return record + return record, None record.verdict = result.verdict record.risk_score = result.risk_score + record.ml_risk_score = result.ml_risk_score record.finding_rule_ids = sorted({f.rule_id for f in result.findings}) - return record + return record, content def _pending_record(meta: SkillMeta) -> SkillRecord: @@ -199,13 +208,21 @@ async def build_snapshot( force_rescan: bool = False, max_skills: int | None = None, scan_budget: int | None = None, - escalate: bool = True, + escalation: EscalationBackend | None = None, + escalation_policy: EscalationPolicy | None = None, concurrency: int = 8, page_size: int = 50, on_progress: ProgressCallback | None = None, ) -> RegistrySnapshot: """Crawl and scan the registry into a :class:`RegistrySnapshot`. + The sweep has two phases. Phase 1 rule-scans every (changed) skill — fast, + free, one request each. Phase 2 gives a *targeted* second opinion: an + :class:`~malwar.monitor.escalation.EscalationPolicy` picks only the skills in + the ambiguous band (rules unsure, or rule-clean but ML-anomalous), and the + ``escalation`` backend re-analyzes just those — so deep (paid) analysis is + spent where it can change the verdict, not on every flagged skill. + Parameters ---------- client: @@ -224,8 +241,12 @@ async def build_snapshot( is recorded as an UNKNOWN placeholder and picked up on the next run, so a registry too large to sweep within one run's time limit is built up over successive runs rather than lost to an all-or-nothing timeout. - escalate: - Re-scan flagged skills with the full LLM + URL pipeline. + escalation: + Second-opinion backend for the ambiguous band (``None`` / ``NoneBackend`` + disables phase 2). See :func:`malwar.monitor.escalation.make_backend`. + escalation_policy: + Which first-pass records qualify for escalation; defaults to + :class:`~malwar.monitor.escalation.EscalationPolicy`. concurrency: Maximum number of skills scanned in parallel. on_progress: @@ -268,6 +289,11 @@ async def build_snapshot( snapshot.pending_count, ) + backend = escalation or NoneBackend() + keep_content = not isinstance(backend, NoneBackend) + contents: dict[str, str] = {} + + # --- Phase 1: rule-scan every skill (fast, free, one request each) --- semaphore = asyncio.Semaphore(max(1, concurrency)) done = len(snapshot.skills) lock = asyncio.Lock() @@ -276,18 +302,58 @@ async def _worker(meta: SkillMeta) -> None: nonlocal done async with semaphore: try: - record = await _scan_slug(client, meta, escalate=escalate) + record, content = await _scan_slug(client, meta) except Exception as exc: # last-resort guard; never abort the sweep record = SkillRecord(slug=meta.slug, verdict="UNKNOWN", error=f"worker: {exc}") + content = None async with lock: snapshot.skills[meta.slug] = record if record.error: snapshot.errors[meta.slug] = record.error + if keep_content and content is not None: + contents[meta.slug] = content done += 1 if on_progress is not None: on_progress(done, total, meta.slug) await asyncio.gather(*(_worker(meta) for meta in to_scan)) + + # --- Phase 2: targeted second opinion on the ambiguous band --- + if keep_content and contents: + policy = escalation_policy or EscalationPolicy() + scanned = {slug: snapshot.skills[slug] for slug in contents} + candidates = select_candidates(scanned, policy) + snapshot.escalated_count = len(candidates) + logger.info( + "escalation: %d of %d scanned skills sent to '%s' backend", + len(candidates), + len(contents), + backend.name, + ) + + esc_sem = asyncio.Semaphore(max(1, concurrency)) + + async def _escalate(slug: str) -> None: + async with esc_sem: + try: + res = await backend.assess(contents[slug], file_name=f"{slug}/SKILL.md") + except Exception as exc: # a bad escalation must not abort the sweep + logger.warning("escalation failed for %s: %s", slug, exc) + return + rec = snapshot.skills[slug] + rec.escalation_backend = res.backend + rec.escalation_verdict = res.verdict + rec.escalation_score = res.score + rec.llm_escalated = res.authoritative + # A full-pipeline second opinion is authoritative — adopt its + # verdict (it may raise a sneaky-clean, or clear a false positive). + if res.authoritative and res.verdict: + rec.verdict = res.verdict + if res.score is not None: + rec.risk_score = round(res.score * 100) + + await asyncio.gather(*(_escalate(slug) for slug in candidates)) + return snapshot diff --git a/tests/unit/test_escalation.py b/tests/unit/test_escalation.py new file mode 100644 index 0000000..05f4db8 --- /dev/null +++ b/tests/unit/test_escalation.py @@ -0,0 +1,206 @@ +"""Unit tests for targeted monitor escalation (policy + backends).""" + +from __future__ import annotations + +import pytest + +from malwar.monitor.escalation import ( + AnthropicBackend, + EscalationPolicy, + EscalationResult, + HfClassifierBackend, + NoneBackend, + TieredBackend, + _parse_classifier_output, + make_backend, + select_candidates, +) +from malwar.monitor.models import SkillRecord + + +def rec(risk: int, *, ml: float | None = None, verdict: str = "CLEAN", error: str | None = None): + return SkillRecord(slug=f"s{risk}", risk_score=risk, ml_risk_score=ml, verdict=verdict, error=error) + + +class TestEscalationPolicy: + def test_ambiguous_middle_qualifies(self): + p = EscalationPolicy() + assert p.qualifies(rec(20, verdict="CAUTION")) # CAUTION + assert p.qualifies(rec(60, verdict="SUSPICIOUS")) # SUSPICIOUS + assert p.qualifies(rec(10)) # just under the CAUTION line but >= min_rule_risk + + def test_confident_clean_skipped(self): + # Rules quiet and no ML signal -> not worth spending on. + assert not EscalationPolicy().qualifies(rec(0)) + assert not EscalationPolicy().qualifies(rec(5)) # below min_rule_risk (8) + + def test_confident_malicious_skipped(self): + # Already a confident detection; a second opinion can't change it. + assert not EscalationPolicy().qualifies(rec(90, verdict="MALICIOUS")) + assert not EscalationPolicy().qualifies(rec(75, verdict="MALICIOUS")) + + def test_sneaky_clean_qualifies_via_ml_when_enabled(self): + # Rules quiet (risk 0) but the ML model finds it anomalous — only when + # the ML path is explicitly enabled (a calibrated threshold). + p = EscalationPolicy(ml_threshold=0.6) + assert p.qualifies(rec(0, ml=0.8)) + assert not p.qualifies(rec(0, ml=0.4)) # below ml_threshold + + def test_ml_path_disabled_by_default(self): + # Default ml_threshold is None (the stock scorer isn't discriminative), + # so a rule-clean skill is never escalated on the ML signal alone. + assert not EscalationPolicy().qualifies(rec(0, ml=0.99)) + + def test_ml_threshold_none_disables_ml_path(self): + p = EscalationPolicy(ml_threshold=None) + assert not p.qualifies(rec(0, ml=0.99)) + + def test_unknown_or_errored_never_qualifies(self): + assert not EscalationPolicy().qualifies(rec(50, verdict="UNKNOWN")) + assert not EscalationPolicy().qualifies(rec(50, verdict="SUSPICIOUS", error="boom")) + + def test_suspicion_uses_max_of_rule_and_ml(self): + p = EscalationPolicy() + assert p.suspicion(rec(40, ml=0.9)) == pytest.approx(0.9) + assert p.suspicion(rec(70, ml=0.1)) == pytest.approx(0.7) + + +class TestSelectCandidates: + def test_ranks_most_suspicious_first(self): + records = { + "low": rec(20, verdict="CAUTION"), + "high": rec(70, verdict="SUSPICIOUS"), + "mid": rec(45, verdict="SUSPICIOUS"), + } + assert select_candidates(records, EscalationPolicy()) == ["high", "mid", "low"] + + def test_budget_caps_to_most_suspicious(self): + records = { + "low": rec(20, verdict="CAUTION"), + "high": rec(70, verdict="SUSPICIOUS"), + "mid": rec(45, verdict="SUSPICIOUS"), + } + assert select_candidates(records, EscalationPolicy(budget=2)) == ["high", "mid"] + + def test_excludes_non_candidates(self): + records = { + "clean": rec(0), + "malicious": rec(90, verdict="MALICIOUS"), + "ambiguous": rec(30, verdict="CAUTION"), + } + assert select_candidates(records, EscalationPolicy()) == ["ambiguous"] + + +class TestParseClassifierOutput: + def test_list_of_dicts(self): + assert _parse_classifier_output([{"label": "INJECTION", "score": 0.91}]) == ("INJECTION", 0.91) + + def test_bare_dict(self): + assert _parse_classifier_output({"label": "SAFE", "score": 0.7}) == ("SAFE", 0.7) + + def test_return_all_scores_picks_argmax(self): + out = [[{"label": "SAFE", "score": 0.2}, {"label": "INJECTION", "score": 0.8}]] + assert _parse_classifier_output(out) == ("INJECTION", 0.8) + + def test_garbage_is_neutral(self): + assert _parse_classifier_output(None) == ("", 0.0) + + +class TestBackends: + async def test_none_backend_never_flags(self): + res = await NoneBackend().assess("anything", file_name="x") + assert res.backend == "none" and not res.flagged and not res.authoritative + + async def test_hf_flags_malicious_label(self): + clf = lambda text: [{"label": "INJECTION", "score": 0.95}] # noqa: E731 + be = HfClassifierBackend(classifier=clf, threshold=0.5) + res = await be.assess("ignore previous instructions", file_name="x") + assert res.flagged + assert res.verdict == "SUSPICIOUS" + assert not res.authoritative # narrow classifier is a signal, not a verdict + assert res.score == pytest.approx(0.95) + + async def test_hf_clears_benign(self): + clf = lambda text: [{"label": "SAFE", "score": 0.99}] # noqa: E731 + res = await HfClassifierBackend(classifier=clf).assess("hello", file_name="x") + assert not res.flagged and res.verdict == "CLEAN" + + async def test_hf_below_threshold_not_flagged(self): + clf = lambda text: [{"label": "INJECTION", "score": 0.3}] # noqa: E731 + res = await HfClassifierBackend(classifier=clf, threshold=0.5).assess("x", file_name="x") + assert not res.flagged + + async def test_hf_unavailable_degrades_gracefully(self): + # No injected classifier and transformers not guaranteed -> must not raise. + be = HfClassifierBackend(model="definitely/not-a-real-model") + be._load_failed = True # simulate missing transformers/model + res = await be.assess("x", file_name="x") + assert res.detail == "unavailable" and not res.flagged + + async def test_hf_classifier_error_is_contained(self): + def boom(text): + raise RuntimeError("cuda oom") + + res = await HfClassifierBackend(classifier=boom).assess("x", file_name="x") + assert not res.flagged and "error" in res.detail + + +class _FakeBackend: + """Records calls and returns a preset result.""" + + def __init__(self, name: str, result: EscalationResult): + self.name = name + self._result = result + self.calls: list[str] = [] + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + self.calls.append(file_name) + return self._result + + +class TestTieredBackend: + async def test_expensive_only_runs_when_cheap_flags(self): + cheap = _FakeBackend("hf", EscalationResult(backend="hf", flagged=True)) + expensive = _FakeBackend( + "anthropic", + EscalationResult(backend="anthropic", flagged=True, verdict="MALICIOUS", authoritative=True), + ) + res = await TieredBackend(cheap, expensive).assess("x", file_name="s/SKILL.md") + assert expensive.calls == ["s/SKILL.md"] # escalated + assert res.backend == "anthropic" and res.authoritative + + async def test_expensive_skipped_when_cheap_clears(self): + cheap = _FakeBackend("hf", EscalationResult(backend="hf", flagged=False, verdict="CLEAN")) + expensive = _FakeBackend("anthropic", EscalationResult(backend="anthropic", flagged=True)) + res = await TieredBackend(cheap, expensive).assess("x", file_name="x") + assert expensive.calls == [] # never escalated -> no paid call + assert res.backend == "hf" and not res.flagged + + +class TestMakeBackend: + def test_names(self): + assert make_backend("none").name == "none" + assert make_backend("hf").name == "hf" + assert make_backend("anthropic").name == "anthropic" + assert make_backend("tiered").name == "tiered" + assert make_backend("").name == "none" + + def test_invalid_raises(self): + with pytest.raises(ValueError, match="unknown escalation backend"): + make_backend("gpt5") + + +class TestAnthropicBackend: + async def test_maps_scan_result(self, monkeypatch): + class _Result: + verdict = "SUSPICIOUS" + risk_score = 55 + + async def fake_scan(content, *, file_name, use_llm, use_urls): + assert use_llm and use_urls + return _Result() + + monkeypatch.setattr("malwar.sdk.scan", fake_scan) + res = await AnthropicBackend().assess("body", file_name="s/SKILL.md") + assert res.flagged and res.verdict == "SUSPICIOUS" + assert res.authoritative and res.score == pytest.approx(0.55) diff --git a/tests/unit/test_monitor.py b/tests/unit/test_monitor.py index f9fdf9e..f1e1598 100644 --- a/tests/unit/test_monitor.py +++ b/tests/unit/test_monitor.py @@ -13,6 +13,8 @@ VersionInfo, ) from malwar.monitor import ( + EscalationPolicy, + EscalationResult, RegistrySnapshot, SkillRecord, SnapshotStore, @@ -102,7 +104,7 @@ async def get_skill_file(self, slug: str, path: str = "SKILL.md", version=None) class TestBuildSnapshot: async def test_scans_all_skills(self): client = FakeClawHubClient({"good": BENIGN_BODY, "money-radar": MALICIOUS_BODY}) - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.skill_count == 2 assert snap.registry == "https://clawhub.test" assert snap.skills["good"].verdict == "CLEAN" @@ -111,7 +113,7 @@ async def test_scans_all_skills(self): async def test_records_content_hash_and_version(self): client = FakeClawHubClient({"good": BENIGN_BODY}) - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) rec = snap.skills["good"] assert rec.content_sha256 and len(rec.content_sha256) == 64 assert rec.version == "1.0.0" @@ -121,7 +123,7 @@ async def test_one_request_per_skill_no_detail_fetch(self): # Metadata comes from the listing, so a scanned skill costs a single # request (the file) — the per-skill detail endpoint is never hit. client = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.skill_count == 2 assert client.detail_fetches == [] assert sorted(client.file_fetches) == ["a", "b"] @@ -129,7 +131,7 @@ async def test_one_request_per_skill_no_detail_fetch(self): async def test_max_skills_cap(self): client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(10)}) - snap = await build_snapshot(client, escalate=False, max_skills=3) + snap = await build_snapshot(client, max_skills=3) assert snap.skill_count == 3 async def test_fetch_error_recorded_not_fatal(self): @@ -138,7 +140,7 @@ async def test_fetch_error_recorded_not_fatal(self): async def _boom(slug, path="SKILL.md", version=None): raise ClawHubError("boom") client.get_skill_file = _boom # type: ignore[assignment] - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.skills["good"].error is not None assert "good" in snap.errors @@ -152,7 +154,7 @@ async def _boom(slug, path="SKILL.md", version=None): return BENIGN_BODY client.get_skill_file = _boom # type: ignore[assignment] - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.skill_count == 2 assert snap.skills["good"].verdict == "CLEAN" assert snap.skills["bad"].error is not None @@ -166,19 +168,19 @@ async def _boom(limit=20, cursor=None): client.list_skills = _boom # type: ignore[assignment] # search fallback returns [] in the fake → empty snapshot, no exception. - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.skill_count == 0 class TestIncrementalSweep: async def test_unchanged_skills_are_not_refetched(self): client1 = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) assert day1.scanned_count == 2 assert day1.reused_count == 0 client2 = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - day2 = await build_snapshot(client2, previous=day1, escalate=False) + day2 = await build_snapshot(client2, previous=day1) # Nothing changed → nothing re-fetched, everything carried forward. assert client2.file_fetches == [] assert day2.scanned_count == 0 @@ -187,13 +189,13 @@ async def test_unchanged_skills_are_not_refetched(self): async def test_version_bump_triggers_rescan(self): client1 = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) client2 = FakeClawHubClient( {"a": BENIGN_BODY, "b": BENIGN_BODY}, versions={"a": "2.0.0", "b": "1.0.0"}, ) - day2 = await build_snapshot(client2, previous=day1, escalate=False) + day2 = await build_snapshot(client2, previous=day1) # Only the version-bumped skill is re-fetched. assert client2.file_fetches == ["a"] assert day2.scanned_count == 1 @@ -201,19 +203,19 @@ async def test_version_bump_triggers_rescan(self): async def test_updated_at_change_triggers_rescan(self): client1 = FakeClawHubClient({"a": BENIGN_BODY}, updated_ats={"a": 1000}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) client2 = FakeClawHubClient({"a": BENIGN_BODY}, updated_ats={"a": 2000}) - day2 = await build_snapshot(client2, previous=day1, escalate=False) + day2 = await build_snapshot(client2, previous=day1) assert client2.file_fetches == ["a"] assert day2.scanned_count == 1 async def test_new_skill_is_scanned_others_reused(self): client1 = FakeClawHubClient({"a": BENIGN_BODY}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) client2 = FakeClawHubClient({"a": BENIGN_BODY, "new": MALICIOUS_BODY}) - day2 = await build_snapshot(client2, previous=day1, escalate=False) + day2 = await build_snapshot(client2, previous=day1) assert client2.file_fetches == ["new"] assert day2.scanned_count == 1 assert day2.reused_count == 1 @@ -221,17 +223,17 @@ async def test_new_skill_is_scanned_others_reused(self): async def test_force_rescan_scans_everything(self): client1 = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) client2 = FakeClawHubClient({"a": BENIGN_BODY, "b": BENIGN_BODY}) - day2 = await build_snapshot(client2, previous=day1, force_rescan=True, escalate=False) + day2 = await build_snapshot(client2, previous=day1, force_rescan=True) assert sorted(client2.file_fetches) == ["a", "b"] assert day2.scanned_count == 2 assert day2.reused_count == 0 async def test_scan_budget_defers_overflow(self): client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(5)}) - snap = await build_snapshot(client, escalate=False, scan_budget=2) + snap = await build_snapshot(client, scan_budget=2) # Only 2 scanned this run; the other 3 recorded as UNKNOWN placeholders. assert snap.skill_count == 5 assert snap.scanned_count == 2 @@ -247,14 +249,14 @@ async def test_budgeted_baseline_converges_over_runs(self): # Three runs of budget 2 must fully cover a 5-skill registry. for _ in range(3): client = FakeClawHubClient(skills) - prev = await build_snapshot(client, previous=prev, escalate=False, scan_budget=2) + prev = await build_snapshot(client, previous=prev, scan_budget=2) assert prev is not None assert prev.pending_count == 0 assert all(r.verdict == "CLEAN" for r in prev.skills.values()) async def test_no_budget_scans_all(self): client = FakeClawHubClient({f"s{i}": BENIGN_BODY for i in range(4)}) - snap = await build_snapshot(client, escalate=False) + snap = await build_snapshot(client) assert snap.scanned_count == 4 assert snap.pending_count == 0 @@ -266,17 +268,87 @@ async def _boom(slug, path="SKILL.md", version=None): raise ClawHubError("boom") client1.get_skill_file = _boom # type: ignore[assignment] - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) assert day1.skills["a"].error is not None # Day 2: same version, but the errored record must be retried (not reused). client2 = FakeClawHubClient({"a": BENIGN_BODY}) - day2 = await build_snapshot(client2, previous=day1, escalate=False) + day2 = await build_snapshot(client2, previous=day1) assert client2.file_fetches == ["a"] assert day2.skills["a"].error is None assert day2.skills["a"].verdict == "CLEAN" +class _RecordingBackend: + """Escalation backend that records which skills it was asked to assess.""" + + def __init__(self, name: str, result: EscalationResult): + self.name = name + self._result = result + self.assessed: list[str] = [] + + async def assess(self, content: str, *, file_name: str) -> EscalationResult: + self.assessed.append(file_name) + return self._result + + +class TestEscalationPhase: + async def test_only_ambiguous_band_is_escalated(self): + # money-radar scores in the ambiguous band (SUSPICIOUS, < MALICIOUS); + # the benign skill is confident-clean and must not be escalated. + client = FakeClawHubClient({"clean": BENIGN_BODY, "amb": MALICIOUS_BODY}) + backend = _RecordingBackend("hf", EscalationResult(backend="hf", flagged=True)) + snap = await build_snapshot(client, escalation=backend) + assert backend.assessed == ["amb/SKILL.md"] + assert snap.escalated_count == 1 + assert snap.skills["amb"].escalation_backend == "hf" + assert snap.skills["clean"].escalation_backend == "" + + async def test_no_escalation_by_default(self): + client = FakeClawHubClient({"amb": MALICIOUS_BODY}) + snap = await build_snapshot(client) # NoneBackend + assert snap.escalated_count == 0 + assert snap.skills["amb"].escalation_backend == "" + + async def test_authoritative_backend_overrides_verdict(self): + # Simulate an LLM second opinion clearing a rule-engine false positive. + client = FakeClawHubClient({"amb": MALICIOUS_BODY}) + backend = _RecordingBackend( + "anthropic", + EscalationResult( + backend="anthropic", flagged=False, verdict="CLEAN", score=0.0, authoritative=True + ), + ) + snap = await build_snapshot(client, escalation=backend) + assert snap.skills["amb"].verdict == "CLEAN" # adopted the authoritative verdict + assert snap.skills["amb"].risk_score == 0 + assert snap.skills["amb"].llm_escalated is True + + async def test_non_authoritative_backend_records_but_keeps_verdict(self): + client = FakeClawHubClient({"amb": MALICIOUS_BODY}) + first = await build_snapshot(client) + rule_verdict = first.skills["amb"].verdict # SUSPICIOUS/CAUTION from rules + + client2 = FakeClawHubClient({"amb": MALICIOUS_BODY}) + backend = _RecordingBackend( + "hf", EscalationResult(backend="hf", flagged=True, verdict="SUSPICIOUS", authoritative=False) + ) + snap = await build_snapshot(client2, escalation=backend) + # A narrow classifier is recorded but does not overwrite the verdict. + assert snap.skills["amb"].verdict == rule_verdict + assert snap.skills["amb"].escalation_verdict == "SUSPICIOUS" + assert snap.skills["amb"].llm_escalated is False + + async def test_escalation_budget_caps_candidates(self): + # Two ambiguous skills, budget 1 -> only the most suspicious is escalated. + client = FakeClawHubClient({"a": MALICIOUS_BODY, "b": MALICIOUS_BODY}) + backend = _RecordingBackend("hf", EscalationResult(backend="hf", flagged=True)) + policy = EscalationPolicy(budget=1) + snap = await build_snapshot(client, escalation=backend, escalation_policy=policy) + assert len(backend.assessed) == 1 + assert snap.escalated_count == 1 + + # --------------------------------------------------------------------------- # diff_snapshots # --------------------------------------------------------------------------- @@ -376,7 +448,7 @@ async def test_day_over_day_detects_trojanized_update(self, tmp_path): # Day 1: two clean skills. client1 = FakeClawHubClient({"good": BENIGN_BODY, "helper": BENIGN_BODY}) - day1 = await build_snapshot(client1, escalate=False) + day1 = await build_snapshot(client1) diff1 = diff_snapshots(store.load_latest(), day1) store.save(day1) assert diff1.is_first_run @@ -387,7 +459,7 @@ async def test_day_over_day_detects_trojanized_update(self, tmp_path): {"good": BENIGN_BODY, "helper": MALICIOUS_BODY}, versions={"good": "1.0.0", "helper": "1.1.0"}, ) - day2 = await build_snapshot(client2, escalate=False) + day2 = await build_snapshot(client2) diff2 = diff_snapshots(store.load_latest(), day2) assert [r.slug for r in diff2.newly_malicious] == ["helper"]