From ee64eec51ae287d2b37308d9da0f84d46e9baf5b Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Fri, 7 Aug 2026 03:27:08 -0700 Subject: [PATCH] release-desktop: add a VirusTotal pre-flight scan of the release bundles (#8089) * release-desktop: add a VirusTotal pre-flight scan of the release bundles * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * virustotal_scan: register the signed upload URL with add-mask * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * virustotal_scan: check out the script, stop replaying single-use upload URLs, bound every request by the deadline * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * virustotal_scan: fail closed on malformed hash lookups and cap pacing by the deadline * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Bound VirusTotal socket calls to the deadline and scan only validated releases - Pass a per-call socket timeout through the transport, clamped to the remaining scan deadline, so a request starting just before the deadline cannot consume the full 300s cushion ahead of the step timeout. - Retry a malformed upload acknowledgement instead of aborting, since the disclosure cost of the upload has already been paid at that point. - Move the scan after 'Create or validate versioned release' so a run that is rejected has not already uploaded all four bundles. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Defer non-draft release creation past the scan and cap retry backoff - Split 'Create or validate versioned release' into a validation step that runs before the scan and a creation step that runs after it. A dispatch with draft=false and a new tag previously published an empty release that stayed assetless for the length of the scan, and permanently so if the run was cancelled part way through. - Clamp the exponential retry backoff to the remaining deadline, so a 429 or 5xx arriving late cannot sleep past --timeout-seconds before the loop notices and writes its summary. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Keep release notes unconditional, fail closed on lookup errors, fix the permission test - Write desktop-release-notes.md in the validation step, which always runs. The updater metadata step reads it on every run, so leaving the write in the conditional create step broke reruns against an existing release. - Only treat a lookup as a missing release when gh reports 'release not found'. Any other failure now fails the step, rather than proceeding to disclose the bundles for a run that cannot publish. - Point test_release_desktop_permissions at the renamed validation step and assert the deferred create step and its gate. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Escape third-party text in the VirusTotal warning annotations Engine names, detection labels and API error strings are third-party data written straight into a workflow command. Actions truncates an annotation at the first newline and mis-parses a bare %, so a crafted or merely awkward detection string could drop the engine list exactly when the scan is trying to alert a maintainer. Mirrors _gha_escape in lockfile_supply_chain_audit.py, including the replace-% first ordering. * Never report an unanalysed bundle as clean, and escape the summary - A hash known to VirusTotal can have no completed analysis, in which case last_analysis_stats is absent and parse_stats yields all zeros. That row read as 'known to VirusTotal' with zero detections, which looks like 70 engines cleared a bundle that none of them scanned. Such a row now reports 'no completed analysis' with stats left unset, so it renders as dashes and cannot trip the threshold. The upload path polls until status is completed, so it only requires a stats object. - Escape third-party engine names, detection labels and error strings in the job summary. It is appended to GITHUB_STEP_SUMMARY and rendered as Markdown, so a newline ended the row and | opened a new cell. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Avoid a CodeQL clear-text-logging false positive on the skip message Interpolating API_KEY_ENV into the skip log trips CodeQL's py/clear-text-logging-sensitive-data rule at high severity, because the constant's name ends in _KEY. It only ever holds the env var name, never the value, but the repo uses CodeQL default setup so there is no config to filter the query on. Write the name out literally and pin it against the constant in test_missing_key_skips_without_failing so the two cannot drift. --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .github/workflows/release-desktop.yml | 117 ++- scripts/virustotal_scan.py | 752 +++++++++++++++ tests/python/test_virustotal_scan.py | 878 ++++++++++++++++++ .../test_release_desktop_permissions.py | 13 +- 4 files changed, 1743 insertions(+), 17 deletions(-) create mode 100644 scripts/virustotal_scan.py create mode 100644 tests/python/test_virustotal_scan.py 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 "