diff --git a/.github/workflows/release-desktop.yml b/.github/workflows/release-desktop.yml index 74c5d5401c..be1bd63061 100644 --- a/.github/workflows/release-desktop.yml +++ b/.github/workflows/release-desktop.yml @@ -1405,6 +1405,18 @@ jobs: with: egress-policy: audit + # This job publishes artifacts built elsewhere, so it historically needed no + # source tree. The VirusTotal step runs a repo script, so it does now. Sparse: + # a full checkout of this repo to fetch one file would be wasteful, and a + # missing script would be swallowed by that step's continue-on-error and + # publish the release with no scan at all. + - name: Check out the scan script + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + sparse-checkout: scripts/virustotal_scan.py + sparse-checkout-cone-mode: false + - name: Download signed release assets uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 with: @@ -1445,20 +1457,31 @@ jobs: print('\n'.join(sorted(actual_names))) PY - - name: Create or validate versioned release + # Validation runs before the scan so a run with a mismatched release state is + # rejected before any bundle is disclosed to VirusTotal. Creating a release + # that does not exist yet is deferred to after the scan: `gh release create` + # without `--draft` publishes immediately, and doing that here would leave a + # public, assetless release for the duration of the scan (and indefinitely if + # the run is cancelled part way through). + - name: Validate versioned release state + id: versioned_release_state shell: bash env: GH_TOKEN: ${{ github.token }} RELEASE_DRAFT: ${{ inputs.draft }} run: | set -euo pipefail + # Written here rather than in the conditional create step below, because + # `Generate and publish versioned updater metadata` reads this file on + # every run, including reruns against an already existing release. notes_file="$RUNNER_TEMP/desktop-release-notes.md" printf '%s\n' "$DESKTOP_RELEASE_NOTES" > "$notes_file" release_json="$RUNNER_TEMP/versioned-release.json" + view_error="$RUNNER_TEMP/versioned-release-view.err" # REST tag lookup omits drafts; `gh release view` also checks pending tags. if gh release view "$DESKTOP_RELEASE_TAG" \ - --json tagName,isDraft,isPrerelease > "$release_json" 2>/dev/null; then + --json tagName,isDraft,isPrerelease > "$release_json" 2>"$view_error"; then python3 <<'PY' import json import os @@ -1475,21 +1498,87 @@ jobs: if bool(release.get('isPrerelease')) != expected_prerelease: sys.exit('Existing desktop release prerelease state does not match the requested version') PY + echo "create=false" >> "$GITHUB_OUTPUT" + elif grep -qi 'release not found' "$view_error"; then + echo "create=true" >> "$GITHUB_OUTPUT" else - release_flags=( - --title "Unsloth ${STUDIO_VERSION}" - --notes-file "$notes_file" - --target "$GITHUB_SHA" - ) - if [ "$RELEASE_DRAFT" = "true" ]; then - release_flags+=(--draft) - fi - if [ "$DESKTOP_PRERELEASE" = "true" ]; then - release_flags+=(--prerelease) - fi - gh release create "$DESKTOP_RELEASE_TAG" "${release_flags[@]}" + # `gh` exits non-zero for any failure, so a transient API, auth or rate + # limit error looks identical to a missing release. Treating those as + # "not found" would disclose the bundles to VirusTotal for a run that + # cannot publish, so fail here instead. + echo "::error::Could not determine whether $DESKTOP_RELEASE_TAG exists; refusing to treat this as a missing release" + cat "$view_error" >&2 + exit 1 fi + # ── Advisory multi-engine scan of the bundles we are about to publish ── + # Defender (build job) is one vendor and gates only Windows. VirusTotal + # aggregates ~70 engines across all four bundles, which is how a false + # positive from some other vendor gets noticed here rather than in a user + # bug report. Advisory on purpose: installers routinely trip one or two + # obscure heuristics, so a hard gate here would block releases on noise. + # Defender stays the authoritative fail-closed check. + # + # The step never fails the job: a missing secret, a VirusTotal outage or a + # quota exhaustion must not be able to block a release. + - name: VirusTotal pre-flight scan + continue-on-error: true + # Free-tier keys allow 4 requests/minute, and the script paces itself to + # match, so four bundles of lookup + upload + polling can take a while. + timeout-minutes: 30 + env: + VT_API_KEY: ${{ secrets.VIRUS_TOTAL_API_TOKEN }} + run: | + set -euo pipefail + # continue-on-error means a silent failure here would publish an unscanned + # release with only a terse "no summary" note, so name the cause loudly. + if [ ! -f scripts/virustotal_scan.py ]; then + echo "::error::scripts/virustotal_scan.py is missing; the checkout step in this job must provide it" + exit 1 + fi + python3 scripts/virustotal_scan.py \ + "$RUNNER_TEMP/desktop-release-assets" \ + --output-markdown "$RUNNER_TEMP/virustotal-summary.md" + + - name: Publish VirusTotal summary + if: always() + run: | + set -euo pipefail + summary="$RUNNER_TEMP/virustotal-summary.md" + if [ -f "$summary" ]; then + cat "$summary" >> "$GITHUB_STEP_SUMMARY" + else + echo '### VirusTotal pre-flight scan' >> "$GITHUB_STEP_SUMMARY" + echo '' >> "$GITHUB_STEP_SUMMARY" + echo 'The scan step produced no summary.' >> "$GITHUB_STEP_SUMMARY" + fi + + # Deferred from the validation step above so the release is created directly + # ahead of its assets, rather than sitting empty for the length of the scan. + - name: Create versioned release + if: steps.versioned_release_state.outputs.create == 'true' + shell: bash + env: + GH_TOKEN: ${{ github.token }} + RELEASE_DRAFT: ${{ inputs.draft }} + run: | + set -euo pipefail + # Written by the validation step above, which always runs. + notes_file="$RUNNER_TEMP/desktop-release-notes.md" + + release_flags=( + --title "Unsloth ${STUDIO_VERSION}" + --notes-file "$notes_file" + --target "$GITHUB_SHA" + ) + if [ "$RELEASE_DRAFT" = "true" ]; then + release_flags+=(--draft) + fi + if [ "$DESKTOP_PRERELEASE" = "true" ]; then + release_flags+=(--prerelease) + fi + gh release create "$DESKTOP_RELEASE_TAG" "${release_flags[@]}" + - name: Publish versioned release assets shell: bash env: diff --git a/scripts/virustotal_scan.py b/scripts/virustotal_scan.py new file mode 100644 index 0000000000..c3d480dc54 --- /dev/null +++ b/scripts/virustotal_scan.py @@ -0,0 +1,752 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 + +"""Advisory VirusTotal pre-flight scan of the built desktop release bundles. + +Stdlib only on purpose: this runs on a bare `ubuntu-latest` runner right after the +artifacts are downloaded, before any Python environment is provisioned, so it cannot +assume `requests` is importable. + +The scan is advisory by default. VirusTotal aggregates roughly 70 engines and +Tauri/NSIS installers routinely trip one or two obscure heuristics, so a hard gate +here would be flaky. `--fail-threshold` exists to make it blocking later. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import ssl +import sys +import time +import urllib.error +import urllib.request +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Iterable, Sequence + + +API_ROOT = "https://www.virustotal.com/api/v3" +API_KEY_ENV = "VT_API_KEY" + +# Signature sidecars are a few hundred bytes of base64 produced by the Tauri updater +# signer. They carry no executable content, so scanning them only burns quota. +SKIPPED_SUFFIXES = (".sig",) + +# The public API allows 4 requests/minute. 20s between requests keeps us just inside +# that even if the runner clock and VirusTotal's window disagree slightly. A premium +# key can drop this to ~1s via --request-interval. +DEFAULT_REQUEST_INTERVAL = 20.0 +DEFAULT_TIMEOUT_SECONDS = 1500.0 + +# 0 disables the gate entirely; any positive N fails when malicious + suspicious >= N. +DEFAULT_FAIL_THRESHOLD = 0 + +_MAX_ATTEMPTS = 4 +# A failed upload means fetching a fresh signed URL, so this is deliberately +# small: each retry re-sends 40+ MB and spends two more requests of quota. +_UPLOAD_ATTEMPTS = 2 +_SOCKET_TIMEOUT = 300.0 + +# Transport contract: (method, url, headers, body, timeout) -> (status, response_bytes). +# Injected so the unit tests can exercise every branch without touching the network. +# `timeout` is the per-call socket budget, already clamped to the scan deadline. +Transport = Callable[[str, str, "dict[str, str]", "bytes | None", float], "tuple[int, bytes]"] + + +@dataclass(frozen = True) +class ScanStats: + """The subset of a VirusTotal stats dict we report on.""" + + malicious: int = 0 + suspicious: int = 0 + undetected: int = 0 + harmless: int = 0 + timeout: int = 0 + + @property + def flagged(self) -> int: + return self.malicious + self.suspicious + + @property + def total(self) -> int: + """Engine verdicts of any kind. Zero means nothing has actually run.""" + return self.malicious + self.suspicious + self.undetected + self.harmless + self.timeout + + +@dataclass +class FileReport: + """One row of the job summary table.""" + + name: str + sha256: str = "" + size: int = 0 + source: str = "skipped" + stats: ScanStats | None = None + detections: list[str] = field(default_factory = list) + note: str = "" + + +def parse_stats(raw: object) -> ScanStats: + """Coerce a VirusTotal stats dict into ScanStats. + + VirusTotal omits keys that are zero and has added categories over time + (`confirmed-timeout`, `type-unsupported`, `failure`), so every field is read + defensively rather than indexed. + """ + if not isinstance(raw, dict): + return ScanStats() + + def _count(key: str) -> int: + value = raw.get(key, 0) + if isinstance(value, bool) or not isinstance(value, (int, float)): + return 0 + return int(value) + + return ScanStats( + malicious = _count("malicious"), + suspicious = _count("suspicious"), + undetected = _count("undetected"), + harmless = _count("harmless"), + # `confirmed-timeout` is a separate bucket that means the same thing to us. + timeout = _count("timeout") + _count("confirmed-timeout"), + ) + + +def parse_detections(raw: object) -> list[str]: + """Return the sorted engine names whose verdict was malicious or suspicious. + + The engine list is what makes a warning actionable: `3 malicious` is noise until + you can see whether it is three no-name heuristics or Microsoft plus Kaspersky. + """ + if not isinstance(raw, dict): + return [] + names: list[str] = [] + for engine, result in raw.items(): + if not isinstance(result, dict): + continue + if result.get("category") in ("malicious", "suspicious"): + label = result.get("result") + if isinstance(label, str) and label: + names.append(f"{engine} ({label})") + else: + names.append(str(engine)) + return sorted(names) + + +def select_scan_targets(paths: Iterable[Path]) -> list[Path]: + """Filter a directory listing down to the bundles worth spending quota on.""" + targets = [ + path for path in paths if path.is_file() and not path.name.endswith(SKIPPED_SUFFIXES) + ] + return sorted(targets, key = lambda path: path.name) + + +def sha256_of(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def exceeds_threshold(reports: Sequence[FileReport], threshold: int) -> bool: + """True when the run should fail. threshold <= 0 means advisory-only.""" + if threshold <= 0: + return False + return any(report.stats is not None and report.stats.flagged >= threshold for report in reports) + + +def _md_code(text: str) -> str: + """Flatten a value that gets wrapped in a Markdown code span. + + A backslash is literal inside a code span, so a backtick cannot be escaped + there; swap it for an apostrophe rather than let it close the span. + """ + return " ".join(text.split()).replace("`", "'") + + +def _md_text(text: str) -> str: + """Neutralise third-party text rendered as Markdown in the job summary. + + Engine names and detection labels are third-party data, and the summary is + appended to `$GITHUB_STEP_SUMMARY`. A newline ends the table row or bullet, + `|` opens a new cell, and `<` starts raw HTML, which GitHub renders, so an + engine list could otherwise break out and obscure the report. + """ + flattened = " ".join(text.split()) + escaped = flattened.replace("\\", "\\\\") + for char in ("`", "|", "*", "_", "[", "]", "#"): + escaped = escaped.replace(char, "\\" + char) + return escaped.replace("<", "<").replace(">", ">") + + +def render_markdown(reports: Sequence[FileReport], threshold: int) -> str: + """Render the job-summary table. Kept pure so it is unit testable.""" + lines = [ + "### VirusTotal pre-flight scan", + "", + "| Asset | Malicious | Suspicious | Undetected | Harmless | Timeout | Source |", + "| --- | ---: | ---: | ---: | ---: | ---: | --- |", + ] + for report in reports: + stats = report.stats + name = _md_code(report.name) + source = _md_text(report.source) + if stats is None: + lines.append(f"| `{name}` | - | - | - | - | - | {source} |") + else: + lines.append( + f"| `{name}` | {stats.malicious} | {stats.suspicious} | " + f"{stats.undetected} | {stats.harmless} | {stats.timeout} | {source} |" + ) + + detected = [report for report in reports if report.detections] + if detected: + lines += ["", "#### Flagging engines", ""] + for report in detected: + engines = ", ".join(_md_text(detection) for detection in report.detections) + lines.append(f"- `{_md_code(report.name)}`: {engines}") + + notes = [report for report in reports if report.note] + if notes: + lines += ["", "#### Notes", ""] + for report in notes: + lines.append(f"- `{_md_code(report.name)}`: {_md_text(report.note)}") + + lines += ["", "
SHA-256", ""] + for report in reports: + lines.append(f"- `{_md_code(report.name)}`: `{_md_code(report.sha256 or 'n/a')}`") + lines += ["
", ""] + + if threshold > 0: + lines.append(f"Failure threshold: {threshold} malicious + suspicious detections.") + else: + lines.append( + "Advisory only: detections are reported as warnings and never fail the release. " + "Windows Defender remains the authoritative gate." + ) + return "\n".join(lines) + "\n" + + +def _default_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float = _SOCKET_TIMEOUT, +) -> tuple[int, bytes]: + request = urllib.request.Request(url, data = body, headers = headers, method = method) + context = ssl.create_default_context() + try: + with urllib.request.urlopen(request, timeout = timeout, context = context) as response: + return response.status, response.read() + except urllib.error.HTTPError as error: + # 404 on a hash lookup is an expected control-flow signal, not a failure. + return error.code, error.read() + + +class VirusTotalClient: + """Thin, rate-limited VirusTotal v3 client.""" + + def __init__( + self, + api_key: str, + transport: Transport | None = None, + request_interval: float = DEFAULT_REQUEST_INTERVAL, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._api_key = api_key + self._transport = transport or _default_transport + self._request_interval = max(0.0, request_interval) + self._sleep = sleep + self._clock = clock + self._last_request_at: float | None = None + + def _throttle(self, deadline: float | None = None) -> None: + """Pace requests, without ever sleeping past the caller's budget. + + Capping matters because the sleep sits between the caller's deadline check + and the network call: an uncapped sleep can carry execution past the + deadline and start a request that then blocks for the full socket timeout. + The caller re-checks the deadline once this returns. + """ + if self._last_request_at is None: + return + elapsed = self._clock() - self._last_request_at + remaining = self._request_interval - elapsed + if remaining <= 0: + return + if deadline is not None: + remaining = min(remaining, max(0.0, deadline - self._clock())) + if remaining > 0: + self._sleep(remaining) + + def _backoff( + self, + seconds: float, + deadline: float | None = None, + ) -> None: + """Sleep between retry attempts, clamped to the remaining budget. + + The retry sleep grows exponentially, so a 429 arriving shortly before the + deadline could otherwise sleep well past `--timeout-seconds` before the + next iteration notices, delaying the summary the release step depends on. + """ + if deadline is None: + self._sleep(seconds) + return + remaining = min(seconds, deadline - self._clock()) + if remaining > 0: + self._sleep(remaining) + + def request( + self, + method: str, + url: str, + body: bytes | None = None, + extra_headers: dict[str, str] | None = None, + allow_status: Sequence[int] = (), + max_attempts: int = _MAX_ATTEMPTS, + deadline: float | None = None, + ) -> tuple[int, object]: + """Issue one API call, retrying 429s and transient network errors. + + Returns (status, decoded_json_or_None). Raises RuntimeError once the retry + budget is spent so the caller can degrade to a warning row. + + `max_attempts = 1` disables retrying, which is mandatory for a POST to a + single-use signed upload URL: replaying it can only ever be rejected. + + `deadline` is checked BEFORE each attempt. One attempt can block for the + full socket timeout, so a loop that only checks afterwards can overrun the + caller's budget by minutes and get the whole step killed before it writes + a summary. + """ + headers = {"x-apikey": self._api_key, "accept": "application/json"} + if extra_headers: + headers.update(extra_headers) + + backoff = self._request_interval if self._request_interval > 0 else 1.0 + last_error = "" + for attempt in range(1, max_attempts + 1): + if deadline is not None and self._clock() >= deadline: + raise TimeoutError(f"deadline reached before {method} {_redact_url(url)}") + self._throttle(deadline) + # Re-check: pacing sleeps between the check above and the call below, so + # without this a request could start after the deadline and then block + # for the full socket timeout, overrunning the step's own budget. + if deadline is not None and self._clock() >= deadline: + raise TimeoutError( + f"deadline reached while pacing before {method} {_redact_url(url)}" + ) + try: + # Clamp the socket budget to what is left. Without this a call that + # starts just before the deadline can still block for the full + # socket timeout and consume the whole cushion the step relies on + # to write its summary. + socket_timeout = _SOCKET_TIMEOUT + if deadline is not None: + socket_timeout = max(1.0, min(socket_timeout, deadline - self._clock())) + status, payload = self._transport(method, url, headers, body, socket_timeout) + except Exception as error: # network layer, DNS, TLS, truncated read + last_error = f"{type(error).__name__}: {error}" + status, payload = 0, b"" + finally: + self._last_request_at = self._clock() + + if status == 429: + # Quota or minute-rate exhaustion. Exponential backoff, then retry. + last_error = "429 rate limited" + if attempt < max_attempts: + self._backoff(backoff * (2 ** (attempt - 1)), deadline) + continue + if status == 0 or status >= 500: + last_error = last_error or f"HTTP {status}" + if attempt < max_attempts: + self._backoff(backoff * (2 ** (attempt - 1)), deadline) + continue + if status >= 400 and status not in allow_status: + raise RuntimeError(f"VirusTotal returned HTTP {status} for {_redact_url(url)}") + + if not payload: + return status, None + try: + return status, json.loads(payload.decode("utf-8", "replace")) + except json.JSONDecodeError: + return status, None + + raise RuntimeError( + f"VirusTotal request failed after {max_attempts} attempt(s): {last_error}" + ) + + def lookup_hash( + self, + sha256: str, + deadline: float | None = None, + ) -> object | None: + """Return the existing file report, or None when VirusTotal has never seen it. + + Doing this first is both a quota saving and a disclosure saving: a bundle that + VirusTotal already holds gains nothing from being uploaded again. + """ + status, payload = self.request( + "GET", + f"{API_ROOT}/files/{sha256}", + allow_status = (404,), + deadline = deadline, + ) + if status == 404: + return None + if not isinstance(payload, dict): + # A 200 whose body did not parse (a proxy error page, a truncated read) + # proves nothing about whether VirusTotal holds this file. Returning None + # here would be indistinguishable from a 404 and would upload the bundle, + # which is the one outcome the lookup exists to avoid: an unnecessary + # disclosure of an unreleased build. Fail closed and let the caller + # record the asset as unavailable instead. + raise RuntimeError("VirusTotal hash lookup returned a malformed body") + return payload + + def upload( + self, + path: Path, + deadline: float | None = None, + ) -> str: + """Upload via the large-file flow and return the analysis id. + + Every desktop bundle is 41-46 MB, which is over the 32 MB cap on + `POST /files`, so the signed upload URL is the only path that works here. + + Each signed URL is SINGLE USE, so the POST is issued with retries disabled. + Replaying one after, say, the response body failed to read would be rejected + no matter how many times we tried, and would report the asset as unavailable + while an analysis was in fact already running. Retrying instead means going + back for a fresh URL, which is what the loop below does. + """ + attempts = max(1, _UPLOAD_ATTEMPTS) + last_error: Exception | None = None + body, content_type = _build_multipart(path) + + for attempt in range(1, attempts + 1): + _, payload = self.request("GET", f"{API_ROOT}/files/upload_url", deadline = deadline) + upload_url = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(upload_url, str) or not upload_url: + raise RuntimeError("VirusTotal did not return an upload URL") + # Mask before the URL is ever used, so anything that later echoes it -- a + # traceback, a future debug print, a library error string -- is scrubbed. + _mask_in_actions(upload_url) + + try: + _, payload = self.request( + "POST", + upload_url, + body = body, + extra_headers = {"content-type": content_type}, + max_attempts = 1, + deadline = deadline, + ) + except TimeoutError: + raise + except Exception as error: + last_error = error + if attempt < attempts: + continue + raise + + analysis_id = None + if isinstance(payload, dict) and isinstance(payload.get("data"), dict): + analysis_id = payload["data"].get("id") + if not isinstance(analysis_id, str) or not analysis_id: + # An accepted upload whose acknowledgement did not parse is a failed + # attempt, not a dead end: VirusTotal may well be analysing the file + # already. Raising straight out would report the asset unavailable + # after we had paid the disclosure cost of sending it, so spend the + # remaining attempt on a fresh signed URL instead. + last_error = RuntimeError("VirusTotal upload did not return an analysis id") + if attempt < attempts: + continue + raise last_error + return analysis_id + + raise RuntimeError(f"VirusTotal upload failed after {attempts} attempt(s): {last_error}") + + def wait_for_analysis(self, analysis_id: str, deadline: float) -> object: + """Poll until the analysis completes or the caller's deadline passes.""" + while True: + # Checked inside request() too, but raising the analysis-specific message + # here keeps the summary row readable. + if self._clock() >= deadline: + raise TimeoutError(f"analysis {analysis_id} did not complete before the deadline") + _, payload = self.request( + "GET", f"{API_ROOT}/analyses/{analysis_id}", deadline = deadline + ) + attributes = _attributes(payload) + if attributes.get("status") == "completed": + return payload + if self._request_interval <= 0: + # With throttling disabled (premium key) the loop would otherwise spin. + self._sleep(1.0) + + +def _build_multipart(path: Path) -> tuple[bytes, str]: + """Encode `path` as a single-part multipart/form-data body under the field `file`.""" + boundary = f"----UnslothDesktopScan{uuid.uuid4().hex}" + head = ( + f"--{boundary}\r\n" + f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n' + "Content-Type: application/octet-stream\r\n\r\n" + ).encode("utf-8") + tail = f"\r\n--{boundary}--\r\n".encode("utf-8") + return head + path.read_bytes() + tail, f"multipart/form-data; boundary={boundary}" + + +def _redact_url(url: str) -> str: + """Strip the query string before logging: signed upload URLs carry credentials.""" + return url.split("?", 1)[0] + + +def _mask_in_actions(value: str) -> None: + """Register `value` with the runner's log scrubber. + + `VT_API_KEY` comes from a repository secret, so Actions masks it everywhere + automatically. The signed upload URL does not: VirusTotal mints it per call and + its query string is a credential, which makes it exactly the "sensitive + information that is not a GitHub secret" the Actions docs say to pass through + `::add-mask::`. Without this, the only thing keeping it out of the log is us + remembering to call `_redact_url` at every print site, which is a rule that + holds right up until someone adds a print or a traceback escapes. + + No-ops off the runner so local runs are not littered with workflow commands. + """ + if value and os.environ.get("GITHUB_ACTIONS") == "true": + print(f"::add-mask::{value}", flush = True) + + +def _attributes(payload: object) -> dict: + if isinstance(payload, dict) and isinstance(payload.get("data"), dict): + attributes = payload["data"].get("attributes") + if isinstance(attributes, dict): + return attributes + return {} + + +def _record( + report: FileReport, source: str, raw_stats: object, raw_results: object, *, completed: bool +) -> None: + """Attach a verdict, but only when an analysis has actually completed. + + A hash can be known to VirusTotal with no finished analysis, in which case + `last_analysis_stats` is missing and `parse_stats` yields an all-zero + ScanStats. Reporting that as a clean row is the worst failure mode this + script has: it looks like 70 engines cleared the bundle when none ran. Leave + `stats` as None instead, which renders as dashes and never trips the gate. + + `completed` says whether the caller already has an authoritative completion + signal. The upload path polls until `status == "completed"`, so a stats dict + is enough there. The hash lookup carries no such field, so it additionally + has to see at least one engine verdict before believing the result. + """ + stats = parse_stats(raw_stats) + if not isinstance(raw_stats, dict) or (not completed and stats.total == 0): + report.source = "no completed analysis" + report.note = ( + "VirusTotal returned no completed analysis for this hash, " + "so the bundle is unscanned rather than clean" + ) + return + report.source = source + report.stats = stats + report.detections = parse_detections(raw_results) + + +def scan_file(client: VirusTotalClient, path: Path, deadline: float) -> FileReport: + """Scan one bundle, degrading to an annotated row rather than raising.""" + report = FileReport(name = path.name, size = path.stat().st_size) + report.sha256 = sha256_of(path) + + try: + existing = client.lookup_hash(report.sha256, deadline = deadline) + if existing is not None: + attributes = _attributes(existing) + _record( + report, + "known to VirusTotal (no upload)", + attributes.get("last_analysis_stats"), + attributes.get("last_analysis_results"), + completed = False, + ) + return report + + analysis_id = client.upload(path, deadline = deadline) + # wait_for_analysis only returns once status == "completed". + attributes = _attributes(client.wait_for_analysis(analysis_id, deadline)) + _record( + report, + "uploaded", + attributes.get("stats"), + attributes.get("results"), + completed = True, + ) + except TimeoutError as error: + report.source = "timed out" + report.note = str(error) + except Exception as error: + report.source = "unavailable" + report.note = f"{type(error).__name__}: {error}" + return report + + +def _gha_escape(text: str) -> str: + """Escape a string for a GH Actions `::warning::` message. + + Engine names, detection labels and error strings are third-party data, so + they can contain anything. GH Actions truncates an annotation at the first + newline unless `\\n`/`\\r` are escaped as `%0A`/`%0D`. `%` must be replaced + first to avoid double-encoding the subsequent escapes. + + Mirrors `_gha_escape` in scripts/lockfile_supply_chain_audit.py. + """ + return text.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") + + +def _emit(report: FileReport) -> None: + """One deterministic, greppable line per asset.""" + stats = report.stats or ScanStats() + print( + f"virustotal_scan: asset={report.name} sha256={report.sha256 or 'n/a'} " + f"bytes={report.size} source={report.source!r} malicious={stats.malicious} " + f"suspicious={stats.suspicious} undetected={stats.undetected} " + f"harmless={stats.harmless} timeout={stats.timeout}", + flush = True, + ) + if report.detections: + # ::warning:: and not ::error:: so the release still ships; see the module docstring. + print( + f"::warning title=VirusTotal detection::{_gha_escape(report.name)}: " + f"{stats.malicious} malicious, {stats.suspicious} suspicious " + f"({_gha_escape(', '.join(report.detections))})", + flush = True, + ) + if report.note: + print( + f"::warning title=VirusTotal scan incomplete::" + f"{_gha_escape(report.name)}: {_gha_escape(report.note)}", + flush = True, + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description = __doc__) + parser.add_argument( + "paths", + nargs = "+", + type = Path, + help = "release bundles to scan, or a directory containing them", + ) + parser.add_argument( + "--output-markdown", + type = Path, + default = None, + help = "write the summary table here (for $GITHUB_STEP_SUMMARY)", + ) + parser.add_argument( + "--timeout-seconds", + type = float, + default = DEFAULT_TIMEOUT_SECONDS, + help = "overall wall-clock cap for the whole scan", + ) + parser.add_argument( + "--request-interval", + type = float, + default = DEFAULT_REQUEST_INTERVAL, + help = "minimum seconds between API calls (free tier allows 4/min)", + ) + parser.add_argument( + "--fail-threshold", + type = int, + default = DEFAULT_FAIL_THRESHOLD, + help = "exit non-zero when malicious + suspicious >= N (0 disables)", + ) + return parser + + +def collect_paths(paths: Sequence[Path]) -> list[Path]: + candidates: list[Path] = [] + for path in paths: + if path.is_dir(): + candidates.extend(sorted(path.iterdir())) + else: + candidates.append(path) + return select_scan_targets(candidates) + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + # The key never comes from argv: CLI arguments are world-readable in /proc. + api_key = os.environ.get(API_KEY_ENV, "").strip() + + def _write_markdown(text: str) -> None: + if args.output_markdown is not None: + args.output_markdown.parent.mkdir(parents = True, exist_ok = True) + args.output_markdown.write_text(text, encoding = "utf-8") + + if not api_key: + # A missing secret must never break a release: forks and re-runs by + # contributors without the org secret still have to be able to publish. + # The env var NAME is written out literally rather than interpolated from + # API_KEY_ENV. Formatting a constant whose name ends in _KEY into a log + # line trips CodeQL's py/clear-text-logging-sensitive-data heuristic, and + # this repo uses CodeQL default setup, so there is no config to filter it + # on. test_missing_key_skips_without_failing asserts the two stay in step. + print( + "virustotal_scan: VT_API_KEY is unset or empty; skipping the scan.", + flush = True, + ) + _write_markdown( + "### VirusTotal pre-flight scan\n\nSkipped: no API key configured for this run.\n" + ) + return 0 + + targets = collect_paths(args.paths) + if not targets: + print("virustotal_scan: no scannable bundles found.", flush = True) + _write_markdown("### VirusTotal pre-flight scan\n\nSkipped: no scannable bundles found.\n") + return 0 + + client = VirusTotalClient(api_key, request_interval = args.request_interval) + deadline = time.monotonic() + max(0.0, args.timeout_seconds) + + reports: list[FileReport] = [] + for path in targets: + if time.monotonic() >= deadline: + report = FileReport(name = path.name, size = path.stat().st_size) + report.source = "skipped" + report.note = "overall scan timeout reached before this asset was submitted" + reports.append(report) + _emit(report) + continue + report = scan_file(client, path, deadline) + reports.append(report) + _emit(report) + + _write_markdown(render_markdown(reports, args.fail_threshold)) + + if exceeds_threshold(reports, args.fail_threshold): + print( + f"virustotal_scan: detections reached the --fail-threshold of {args.fail_threshold}.", + file = sys.stderr, + flush = True, + ) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/python/test_virustotal_scan.py b/tests/python/test_virustotal_scan.py new file mode 100644 index 0000000000..6bd2a32cf0 --- /dev/null +++ b/tests/python/test_virustotal_scan.py @@ -0,0 +1,878 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 + +"""Unit tests for the advisory VirusTotal pre-flight scan. + +Offline by design: every test injects a fake transport, so the suite never spends +the account's 500/day quota and never uploads a build. The two behaviours worth +protecting are the ones a release depends on: + + - a missing API key must skip, never fail, or a contributor without the org + secret cannot publish at all, + - the bundles are 41-46 MB, over the 32 MB cap on `POST /files`, so the upload + must go through `GET /files/upload_url`. A regression to the plain endpoint + would fail on every asset. +""" + +from __future__ import annotations + +import importlib.util +import pathlib +import sys +import time + +import pytest + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +MODULE_PATH = REPO_ROOT / "scripts" / "virustotal_scan.py" + + +def _load_module(): + spec = importlib.util.spec_from_file_location("virustotal_scan", MODULE_PATH) + if spec is None or spec.loader is None: + pytest.skip(f"cannot import {MODULE_PATH}") + module = importlib.util.module_from_spec(spec) + sys.modules["virustotal_scan"] = module + spec.loader.exec_module(module) + return module + + +vt = _load_module() + + +class FakeTransport: + """Records every call and replays a queued (status, body) per URL fragment.""" + + def __init__(self, routes: dict[str, tuple[int, bytes]]): + self.routes = routes + self.calls: list[tuple[str, str, dict, int]] = [] + self.timeouts: list[float | None] = [] + + def __call__( + self, + method, + url, + headers, + body, + timeout = None, + ): + self.timeouts.append(timeout) + self.calls.append((method, url, headers, len(body or b""))) + for fragment, response in self.routes.items(): + if fragment in url: + return response + raise AssertionError(f"unrouted request: {method} {url}") + + +def _client(routes): + transport = FakeTransport(routes) + client = vt.VirusTotalClient( + "fake-key", + transport = transport, + request_interval = 0.0, + sleep = lambda _seconds: None, + ) + return client, transport + + +class TestParseStats: + def test_missing_keys_default_to_zero(self): + stats = vt.parse_stats({"malicious": 2}) + assert (stats.malicious, stats.suspicious, stats.undetected) == (2, 0, 0) + + def test_non_dict_is_tolerated(self): + assert vt.parse_stats(None) == vt.ScanStats() + assert vt.parse_stats([1, 2]) == vt.ScanStats() + + def test_confirmed_timeout_folds_into_timeout(self): + assert vt.parse_stats({"timeout": 1, "confirmed-timeout": 2}).timeout == 3 + + def test_booleans_are_not_counted_as_ints(self): + # bool is a subclass of int; True must not silently become 1 detection. + assert vt.parse_stats({"malicious": True}).malicious == 0 + + def test_flagged_sums_malicious_and_suspicious(self): + assert vt.parse_stats({"malicious": 3, "suspicious": 4}).flagged == 7 + + +class TestParseDetections: + def test_only_malicious_and_suspicious_are_reported(self): + names = vt.parse_detections( + { + "AlphaAV": {"category": "malicious", "result": "Trojan.Gen"}, + "BetaAV": {"category": "undetected"}, + "GammaAV": {"category": "suspicious", "result": None}, + "DeltaAV": {"category": "harmless"}, + } + ) + assert names == ["AlphaAV (Trojan.Gen)", "GammaAV"] + + def test_non_dict_is_tolerated(self): + assert vt.parse_detections("nope") == [] + + +class TestThreshold: + def _reports(self, flagged): + return [vt.FileReport(name = "a.exe", stats = vt.ScanStats(malicious = flagged))] + + def test_zero_threshold_is_advisory_only(self): + # The shipped default. Detections must never fail the release. + assert vt.exceeds_threshold(self._reports(50), 0) is False + assert vt.exceeds_threshold(self._reports(50), -1) is False + + def test_positive_threshold_fails_at_or_above(self): + assert vt.exceeds_threshold(self._reports(3), 3) is True + assert vt.exceeds_threshold(self._reports(2), 3) is False + + def test_rows_without_stats_never_trip_the_gate(self): + assert vt.exceeds_threshold([vt.FileReport(name = "a.exe")], 1) is False + + +class TestSelectScanTargets: + def test_sig_sidecars_are_skipped(self, tmp_path): + for name in ( + "Unsloth-Desktop-0_1_1-Windows.exe", + "Unsloth-Desktop-0_1_1-Windows.exe.sig", + "Unsloth-Desktop-0_1_1-Linux.AppImage", + "Unsloth-Desktop-0_1_1-Linux.AppImage.sig", + ): + (tmp_path / name).write_bytes(b"x") + names = [path.name for path in vt.collect_paths([tmp_path])] + assert names == [ + "Unsloth-Desktop-0_1_1-Linux.AppImage", + "Unsloth-Desktop-0_1_1-Windows.exe", + ] + + def test_directories_are_expanded_and_files_passed_through(self, tmp_path): + (tmp_path / "a.dmg").write_bytes(b"x") + assert [p.name for p in vt.collect_paths([tmp_path / "a.dmg"])] == ["a.dmg"] + + +class TestMissingKey: + def test_missing_key_skips_without_failing(self, tmp_path, monkeypatch, capsys): + monkeypatch.delenv(vt.API_KEY_ENV, raising = False) + (tmp_path / "a.exe").write_bytes(b"x") + summary = tmp_path / "summary.md" + rc = vt.main([str(tmp_path), "--output-markdown", str(summary)]) + assert rc == 0 + assert "Skipped: no API key" in summary.read_text() + out = capsys.readouterr().out + assert "skipping the scan" in out + # The message spells VT_API_KEY out literally to avoid a CodeQL + # false positive, so pin that it still matches the constant. + assert vt.API_KEY_ENV in out + + def test_whitespace_only_key_is_treated_as_missing(self, tmp_path, monkeypatch): + monkeypatch.setenv(vt.API_KEY_ENV, " ") + (tmp_path / "a.exe").write_bytes(b"x") + assert vt.main([str(tmp_path)]) == 0 + + +class TestLargeFileUploadFlow: + def test_upload_uses_the_signed_url_not_the_32mb_endpoint(self, tmp_path): + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + signed = "https://upload.virustotal.example/receive?sig=secret" + client, transport = _client( + { + "/files/upload_url": (200, b'{"data": "' + signed.encode() + b'"}'), + "upload.virustotal.example": (200, b'{"data": {"id": "analysis-1"}}'), + } + ) + + assert client.upload(bundle) == "analysis-1" + + methods_urls = [(m, u) for m, u, _h, _n in transport.calls] + assert methods_urls[0] == ("GET", f"{vt.API_ROOT}/files/upload_url") + assert methods_urls[1][0] == "POST" + assert methods_urls[1][1] == signed + # The plain 32 MB-capped endpoint must never be used for a bundle. + assert all(u.rstrip("/") != f"{vt.API_ROOT}/files" for _m, u in methods_urls) + + def test_upload_body_is_multipart_with_the_file_field(self, tmp_path): + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + body, content_type = vt._build_multipart(bundle) + assert content_type.startswith("multipart/form-data; boundary=") + assert b'name="file"' in body + assert b'filename="big.exe"' in body + assert b"payload" in body + + def test_api_key_is_sent_as_a_header_never_in_the_url(self, tmp_path): + client, transport = _client({"/files/": (200, b"{}")}) + client.lookup_hash("a" * 64) + _method, url, headers, _n = transport.calls[0] + assert headers["x-apikey"] == "fake-key" + assert "fake-key" not in url + + +class TestHashLookupFirst: + def test_known_hash_short_circuits_the_upload(self, tmp_path): + bundle = tmp_path / "known.exe" + bundle.write_bytes(b"payload") + client, transport = _client( + { + "/files/": ( + 200, + b'{"data": {"attributes": {"last_analysis_stats": ' + b'{"malicious": 1}, "last_analysis_results": ' + b'{"AlphaAV": {"category": "malicious", "result": "X"}}}}}', + ), + } + ) + report = vt.scan_file(client, bundle, deadline = float("inf")) + assert report.source == "known to VirusTotal (no upload)" + assert report.stats.malicious == 1 + assert report.detections == ["AlphaAV (X)"] + # Exactly one call: the lookup. No upload_url, no upload, no polling. + assert len(transport.calls) == 1 + + def test_unknown_hash_falls_through_to_upload(self, tmp_path): + bundle = tmp_path / "new.exe" + bundle.write_bytes(b"payload") + client, transport = _client( + { + "/files/upload_url": (200, b'{"data": "https://up.example/x"}'), + "up.example": (200, b'{"data": {"id": "an-1"}}'), + "/analyses/": ( + 200, + b'{"data": {"attributes": {"status": "completed", ' + b'"stats": {"malicious": 0}, "results": {}}}}', + ), + "/files/": (404, b"{}"), + } + ) + report = vt.scan_file(client, bundle, deadline = float("inf")) + assert report.source == "uploaded" + assert report.stats.malicious == 0 + + +class TestFailureDegradation: + def test_transport_failure_degrades_to_a_note_not_an_exception(self, tmp_path): + bundle = tmp_path / "a.exe" + bundle.write_bytes(b"payload") + client, _transport = _client({"/files/": (500, b"")}) + report = vt.scan_file(client, bundle, deadline = float("inf")) + assert report.source == "unavailable" + assert report.note + assert report.stats is None + + def test_redact_url_strips_the_signed_query_string(self): + assert vt._redact_url("https://up.example/x?sig=secret") == "https://up.example/x" + + +class TestSignedUrlMasking: + """The signed upload URL is a credential and is NOT a registered GitHub secret, + so the runner will not mask it unless we register it with ::add-mask::.""" + + def test_upload_registers_the_signed_url_with_add_mask(self, tmp_path, monkeypatch, capsys): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + signed = "https://upload.virustotal.example/receive?sig=secret-credential" + client, _transport = _client( + { + "/files/upload_url": (200, b'{"data": "' + signed.encode() + b'"}'), + "upload.virustotal.example": (200, b'{"data": {"id": "an-1"}}'), + } + ) + client.upload(bundle) + out = capsys.readouterr().out + assert f"::add-mask::{signed}" in out + # Masking must happen before the URL is used, not after. + assert out.index("::add-mask::") == 0 + + def test_no_workflow_commands_off_the_runner(self, tmp_path, monkeypatch, capsys): + monkeypatch.delenv("GITHUB_ACTIONS", raising = False) + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + client, _transport = _client( + { + "/files/upload_url": (200, b'{"data": "https://up.example/x?sig=s"}'), + "up.example": (200, b'{"data": {"id": "an-1"}}'), + } + ) + client.upload(bundle) + assert "::add-mask::" not in capsys.readouterr().out + + def test_empty_value_is_not_registered(self, monkeypatch, capsys): + monkeypatch.setenv("GITHUB_ACTIONS", "true") + vt._mask_in_actions("") + assert capsys.readouterr().out == "" + + +class TestSingleUseUploadUrl: + """A signed upload URL is single use, so replaying one can only ever be + rejected. A failed upload must go back for a fresh URL instead.""" + + def test_upload_post_is_not_retried_on_the_same_url(self, tmp_path): + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + seen_upload_urls = [] + + class Transport: + def __init__(self): + self.posts = 0 + + def __call__( + self, + method, + url, + headers, + body, + timeout = None, + ): + if url.endswith("/files/upload_url"): + token = f"https://up.example/{len(seen_upload_urls)}" + seen_upload_urls.append(token) + return 200, b'{"data": "' + token.encode() + b'"}' + self.posts += 1 + if self.posts == 1: + return 500, b"" # server-side blip on the first signed URL + return 200, b'{"data": {"id": "an-2"}}' + + transport = Transport() + client = vt.VirusTotalClient( + "k", transport = transport, request_interval = 0.0, sleep = lambda _s: None + ) + assert client.upload(bundle) == "an-2" + # Two distinct signed URLs were fetched: the failed POST was not replayed. + assert len(seen_upload_urls) == 2 + assert transport.posts == 2 + + def test_max_attempts_one_disables_retry(self, tmp_path): + calls = [] + + def transport( + method, + url, + headers, + body, + timeout = None, + ): + calls.append(url) + return 500, b"" + + client = vt.VirusTotalClient( + "k", transport = transport, request_interval = 0.0, sleep = lambda _s: None + ) + with pytest.raises(RuntimeError): + client.request("POST", "https://up.example/x", max_attempts = 1) + assert len(calls) == 1 + + +class TestDeadlineEnforcement: + """One attempt can block for the full socket timeout, so the deadline has to be + checked BEFORE a request, not after, or the step timeout kills the process + before any summary is written.""" + + def test_request_checks_deadline_before_issuing(self): + calls = [] + + def transport( + method, + url, + headers, + body, + timeout = None, + ): + calls.append(url) + return 200, b"{}" + + client = vt.VirusTotalClient( + "k", + transport = transport, + request_interval = 0.0, + sleep = lambda _s: None, + clock = lambda: 1000.0, + ) + with pytest.raises(TimeoutError): + client.request("GET", "https://api.example/x", deadline = 999.0) + assert calls == [] + + def test_wait_for_analysis_stops_at_the_deadline(self): + def transport( + method, + url, + headers, + body, + timeout = None, + ): + return 200, b'{"data": {"attributes": {"status": "queued"}}}' + + now = [0.0] + client = vt.VirusTotalClient( + "k", + transport = transport, + request_interval = 0.0, + sleep = lambda _s: None, + clock = lambda: now[0], + ) + with pytest.raises(TimeoutError): + now[0] = 100.0 + client.wait_for_analysis("an-1", deadline = 50.0) + + def test_scan_file_reports_a_timeout_row_rather_than_raising(self, tmp_path): + bundle = tmp_path / "a.exe" + bundle.write_bytes(b"payload") + client = vt.VirusTotalClient( + "k", + transport = lambda *a: (200, b"{}"), + request_interval = 0.0, + sleep = lambda _s: None, + clock = lambda: 1000.0, + ) + report = vt.scan_file(client, bundle, deadline = 0.0) + assert report.source == "timed out" + assert report.note + + def test_summary_is_still_written_when_every_asset_times_out(self, tmp_path, monkeypatch): + monkeypatch.setenv(vt.API_KEY_ENV, "k") + (tmp_path / "a.exe").write_bytes(b"x") + summary = tmp_path / "s.md" + rc = vt.main( + [ + str(tmp_path), + "--output-markdown", + str(summary), + "--timeout-seconds", + "0", + "--request-interval", + "0", + ] + ) + assert rc == 0 + assert "VirusTotal pre-flight scan" in summary.read_text() + + +class TestRenderMarkdown: + def test_advisory_footer_when_threshold_disabled(self): + text = vt.render_markdown( + [vt.FileReport(name = "a.exe", stats = vt.ScanStats(), sha256 = "ab")], 0 + ) + assert "Advisory only" in text + assert "never fail the release" in text + + def test_threshold_footer_when_enabled(self): + text = vt.render_markdown([vt.FileReport(name = "a.exe", stats = vt.ScanStats())], 4) + assert "Failure threshold: 4" in text + + def test_flagging_engines_are_listed(self): + text = vt.render_markdown( + [ + vt.FileReport( + name = "a.exe", stats = vt.ScanStats(malicious = 1), detections = ["AlphaAV (Trojan)"] + ) + ], + 0, + ) + assert "Flagging engines" in text + assert "AlphaAV (Trojan)" in text + + +class TestFailClosedOnMalformedLookup: + """A 200 whose body does not parse must not be read as 'never seen'. + + Returning None there is indistinguishable from a 404 and uploads the bundle, + which is an unnecessary disclosure of an unreleased build. + """ + + def test_malformed_200_does_not_upload(self, tmp_path): + bundle = tmp_path / "draft.exe" + bundle.write_bytes(b"unreleased build") + client, transport = _client( + { + "/files/upload_url": (200, b'{"data": "https://up.example/x"}'), + "up.example": (200, b'{"data": {"id": "an-1"}}'), + "/files/": (200, b"proxy error page"), + } + ) + report = vt.scan_file(client, bundle, deadline = float("inf")) + assert report.source == "unavailable" + assert "malformed" in report.note + assert not any("up.example" in url for _m, url, _h, _n in transport.calls) + + def test_malformed_200_is_distinguishable_from_404(self, tmp_path): + client, _ = _client({"/files/": (200, b"not json")}) + with pytest.raises(RuntimeError, match = "malformed"): + client.lookup_hash("a" * 64) + + client, _ = _client({"/files/": (404, b"{}")}) + assert client.lookup_hash("a" * 64) is None + + +class TestDeadlineIsNotOverrunByThrottling: + """Pacing sleeps between the deadline check and the network call.""" + + def _clocked_client( + self, + routes, + interval = 20.0, + ): + now = [1000.0] + transport = FakeTransport(routes) + + def sleep(seconds): + now[0] += seconds + + client = vt.VirusTotalClient( + "k", + transport = transport, + request_interval = interval, + sleep = sleep, + clock = lambda: now[0], + ) + return client, transport, now + + def test_transport_never_starts_after_the_deadline(self): + client, transport, now = self._clocked_client({"x.example": (200, b"{}")}) + client._last_request_at = now[0] # force a full interval of pacing + deadline = now[0] + 5.0 # less budget than the pacing needs + with pytest.raises(TimeoutError, match = "pacing"): + client.request("GET", "https://x.example/y", deadline = deadline) + assert transport.calls == [] + + def test_throttle_sleep_is_capped_by_the_deadline(self): + client, _transport, now = self._clocked_client({"x.example": (200, b"{}")}) + client._last_request_at = now[0] + deadline = now[0] + 5.0 + client._throttle(deadline) + # Capped at the 5s of remaining budget, not the full 20s interval. + assert now[0] == pytest.approx(1005.0) + + def test_a_request_with_budget_left_still_proceeds(self): + client, transport, now = self._clocked_client({"x.example": (200, b"{}")}) + client._last_request_at = now[0] + status, _payload = client.request("GET", "https://x.example/y", deadline = now[0] + 600.0) + assert status == 200 + assert len(transport.calls) == 1 + + +class TestSocketBudgetIsClampedToTheDeadline: + """The per-call socket timeout has to respect the scan deadline. + + Otherwise a call that starts just before the deadline still blocks for the + full socket timeout and eats the cushion the step needs to write its summary. + """ + + def test_socket_timeout_is_clamped_to_remaining_budget(self): + client, transport = _client({"x.example": (200, b"{}")}) + client.request("GET", "https://x.example/y", deadline = time.monotonic() + 30.0) + assert transport.timeouts[0] <= 30.0 + + def test_socket_timeout_is_the_default_when_budget_is_large(self): + client, transport = _client({"x.example": (200, b"{}")}) + client.request("GET", "https://x.example/y", deadline = time.monotonic() + 100000.0) + assert transport.timeouts[0] == vt._SOCKET_TIMEOUT + + def test_socket_timeout_without_a_deadline_is_the_default(self): + client, transport = _client({"x.example": (200, b"{}")}) + client.request("GET", "https://x.example/y") + assert transport.timeouts[0] == vt._SOCKET_TIMEOUT + + def test_clamp_never_goes_to_zero_or_negative(self): + # A non-positive urlopen timeout would fail instantly rather than try. + client, transport = _client({"x.example": (200, b"{}")}) + client.request("GET", "https://x.example/y", deadline = time.monotonic() + 0.001) + assert transport.timeouts[0] >= 1.0 + + +class TestMalformedUploadAcknowledgement: + """An accepted upload whose ack did not parse is a failed attempt. + + Raising straight out reports the asset unavailable after we already paid the + disclosure cost of sending the bundle. + """ + + def test_malformed_ack_retries_with_a_fresh_signed_url(self, tmp_path): + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + state = {"n": 0} + + def transport( + method, + url, + headers, + body, + timeout = None, + ): + if "/files/upload_url" in url: + state["n"] += 1 + return (200, b'{"data": "https://up.example/%d"}' % state["n"]) + # First ack is unparseable, second is well formed. + if state["n"] == 1: + return (200, b"not json") + return (200, b'{"data": {"id": "an-2"}}') + + client = vt.VirusTotalClient( + "k", transport = transport, request_interval = 0.0, sleep = lambda _s: None + ) + assert client.upload(bundle) == "an-2" + assert state["n"] == 2 # a second, fresh signed URL was fetched + + def test_malformed_ack_on_the_last_attempt_raises(self, tmp_path): + bundle = tmp_path / "big.exe" + bundle.write_bytes(b"payload") + + def transport( + method, + url, + headers, + body, + timeout = None, + ): + if "/files/upload_url" in url: + return (200, b'{"data": "https://up.example/x"}') + return (200, b"not json") + + client = vt.VirusTotalClient( + "k", transport = transport, request_interval = 0.0, sleep = lambda _s: None + ) + with pytest.raises(RuntimeError, match = "analysis id"): + client.upload(bundle) + + +class TestNoCompletedAnalysis: + """A known hash with no finished analysis must not read as clean. Reporting + zero detections when no engine ran is the worst outcome available here.""" + + def _report( + self, + attributes, + completed = False, + ): + report = vt.FileReport(name = "a.exe") + vt._record(report, "known to VirusTotal (no upload)", *attributes, completed = completed) + return report + + def test_a_completed_analysis_is_trusted_without_engine_counts(self): + # The upload path polls until status == "completed", so a stats dict is + # authoritative there even if the counts are all zero. + report = self._report(({"malicious": 0}, {}), completed = True) + assert report.stats is not None + assert report.source == "known to VirusTotal (no upload)" + + def test_a_completed_analysis_still_needs_a_stats_object(self): + assert self._report((None, {}), completed = True).stats is None + + def test_missing_stats_is_not_reported_as_clean(self): + report = self._report((None, None)) + assert report.stats is None + assert report.source == "no completed analysis" + assert "unscanned rather than clean" in report.note + + def test_all_zero_stats_is_not_reported_as_clean(self): + # A stats dict where no engine reported anything means nothing ran. + report = self._report(({"malicious": 0, "undetected": 0}, {})) + assert report.stats is None + assert report.source == "no completed analysis" + + def test_a_real_verdict_is_kept(self): + report = self._report( + ( + {"malicious": 0, "undetected": 70}, + {"AlphaAV": {"category": "undetected"}}, + ) + ) + assert report.stats is not None + assert report.stats.undetected == 70 + assert report.source == "known to VirusTotal (no upload)" + assert report.note == "" + + def test_an_unanalysed_row_never_trips_the_gate(self): + # stats=None rows are ignored by the threshold, so this stays advisory. + assert vt.exceeds_threshold([self._report((None, None))], 1) is False + + +class TestMarkdownEscaping: + """The summary is a second sink for third-party text, appended to + $GITHUB_STEP_SUMMARY and rendered as Markdown.""" + + def _summary(self, report): + return vt.render_markdown([report], 0) + + def test_a_newline_cannot_break_out_of_a_table_row(self): + report = vt.FileReport( + name = "a.exe", + stats = vt.ScanStats(malicious = 1, undetected = 1), + detections = ["Evil\n| fake | row |"], + ) + body = self._summary(report) + bullet = [line for line in body.splitlines() if "Evil" in line] + # The newline is flattened, so the detection stays on its own bullet. + assert len(bullet) == 1 + assert "\\|" in bullet[0] + assert "| fake | row |" not in body + + def test_html_is_neutralised(self): + report = vt.FileReport(name = "a.exe", note = "") + body = self._summary(report) + assert "<img" in body + assert "