BoltClam 1.0.0: ClamAV desktop protection with Bolt sign-in
Adds the PyQt6 Windows client (Homepage, Scan & results, About & updates, Settings), ClamAV scanning and quarantine, browser-download watch with an exclusive scan gate, Bolt loopback/PKCE sign-in with DPAPI key storage, Gitea release feed with checksum verification, opt-in diagnostics, and Windows/portable build tooling. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
TheMajesticMagician
@@ -0,0 +1,32 @@
|
||||
"""Test bootstrap: never touch the real BoltClam profile or Downloads."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
from boltclam import paths # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_env(tmp_path: Path, monkeypatch) -> Path:
|
||||
monkeypatch.setenv("BOLTCLAM_CONFIG_DIR", str(tmp_path / "config"))
|
||||
monkeypatch.setenv("BOLTCLAM_DATA_DIR", str(tmp_path / "data"))
|
||||
paths.ensure_dirs()
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def qapp():
|
||||
from PyQt6.QtWidgets import QApplication
|
||||
|
||||
app = QApplication.instance() or QApplication([])
|
||||
return app
|
||||
@@ -0,0 +1,198 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import http.server
|
||||
import json
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import pytest
|
||||
|
||||
from boltclam.auth import (
|
||||
AuthError,
|
||||
AuthService,
|
||||
CredentialStore,
|
||||
LoopbackAuthReceiver,
|
||||
pkce_pair,
|
||||
)
|
||||
from boltclam.config import AccountCache
|
||||
|
||||
ACCOUNT = {
|
||||
"user": {"id": "1", "email": "aaron@example.com", "name": "Aaron",
|
||||
"username": "aaron"},
|
||||
"subscription": {"tier": "premium", "label": "Premium", "status": "active"},
|
||||
"balance": {"weekly_remaining": 100, "topup": 0, "allowed": True},
|
||||
"features": {"licensed": True},
|
||||
}
|
||||
|
||||
|
||||
class _Handler(http.server.BaseHTTPRequestHandler):
|
||||
def _json(self, status: int, payload: dict) -> None:
|
||||
data = json.dumps(payload).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(data)))
|
||||
self.end_headers()
|
||||
self.wfile.write(data)
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
if self.path.startswith("/boltclam/account"):
|
||||
key = self.headers.get("X-Desk-Api-Key", "")
|
||||
if key in ("bolt_live_good", "bolt_live_new"):
|
||||
self._json(200, ACCOUNT)
|
||||
else:
|
||||
self._json(401, {"error": "invalid_credentials"})
|
||||
return
|
||||
self._json(404, {"error": "not_found"})
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
raw = self.rfile.read(length) if length else b"{}"
|
||||
if self.path == "/desk/auth/oauth/exchange":
|
||||
payload = json.loads(raw or b"{}")
|
||||
self.server.exchanges.append(payload) # type: ignore[attr-defined]
|
||||
self._json(200, {"api_key": "bolt_live_new", "user": ACCOUNT["user"]})
|
||||
return
|
||||
if self.path == "/desk/auth/logout":
|
||||
self._json(200, {})
|
||||
return
|
||||
self._json(404, {"error": "not_found"})
|
||||
|
||||
def log_message(self, fmt: str, *args) -> None:
|
||||
return
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def web_server():
|
||||
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
|
||||
server.exchanges = [] # type: ignore[attr-defined]
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
yield server
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
|
||||
def _auth(tmp_path, base_url) -> AuthService:
|
||||
return AuthService(
|
||||
base_url, install_id="test-install",
|
||||
credentials=CredentialStore(tmp_path / "credentials.dat"),
|
||||
account_cache=AccountCache(tmp_path / "account.json"))
|
||||
|
||||
|
||||
def test_pkce_pair_matches_s256():
|
||||
verifier, challenge = pkce_pair()
|
||||
expected = base64.urlsafe_b64encode(
|
||||
hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
|
||||
assert challenge == expected
|
||||
assert 43 <= len(verifier) <= 128
|
||||
|
||||
|
||||
def test_credential_store_fallback_round_trip(tmp_path):
|
||||
store = CredentialStore(tmp_path / "credentials.dat")
|
||||
store.save("bolt_live_secret")
|
||||
assert store.load() == "bolt_live_secret"
|
||||
store.delete()
|
||||
assert store.load() is None
|
||||
|
||||
|
||||
def test_key_login_restore_and_logout(tmp_path, web_server):
|
||||
base = f"http://127.0.0.1:{web_server.server_address[1]}"
|
||||
auth = _auth(tmp_path, base)
|
||||
session = auth.login_with_key("bolt_live_good")
|
||||
assert session.api_key == "bolt_live_good"
|
||||
assert session.account["user"]["email"] == "aaron@example.com"
|
||||
assert auth.stored_key() == "bolt_live_good"
|
||||
|
||||
restored = auth.restore()
|
||||
assert restored.account["subscription"]["tier"] == "premium"
|
||||
assert restored.offline is False
|
||||
|
||||
auth.logout()
|
||||
assert auth.stored_key() is None
|
||||
assert auth.account_cache.load() is None
|
||||
|
||||
|
||||
def test_key_login_rejects_a_bad_key(tmp_path, web_server):
|
||||
base = f"http://127.0.0.1:{web_server.server_address[1]}"
|
||||
auth = _auth(tmp_path, base)
|
||||
with pytest.raises(AuthError):
|
||||
auth.login_with_key("bolt_live_wrong")
|
||||
|
||||
|
||||
def test_browser_login_completes_the_loopback_pkce_handoff(tmp_path, web_server):
|
||||
base = f"http://127.0.0.1:{web_server.server_address[1]}"
|
||||
auth = _auth(tmp_path, base)
|
||||
opened: dict[str, str] = {}
|
||||
|
||||
def open_url(url: str) -> bool:
|
||||
opened["url"] = url
|
||||
parsed = urlparse(url)
|
||||
query = parse_qs(parsed.query)
|
||||
redirect_uri = query["redirect_uri"][0]
|
||||
state = query["state"][0]
|
||||
|
||||
def callback() -> None:
|
||||
time.sleep(0.2)
|
||||
with urllib.request.urlopen(
|
||||
f"{redirect_uri}?code=one-time&state={state}", timeout=5) as response:
|
||||
response.read()
|
||||
|
||||
threading.Thread(target=callback, daemon=True).start()
|
||||
return True
|
||||
|
||||
session = auth.browser_login(open_url, timeout=10)
|
||||
assert session.api_key == "bolt_live_new"
|
||||
assert auth.stored_key() == "bolt_live_new"
|
||||
exchange = web_server.exchanges[-1] # type: ignore[attr-defined]
|
||||
assert exchange["client"] == "boltclam"
|
||||
assert exchange["code"] == "one-time"
|
||||
assert exchange["code_verifier"]
|
||||
assert exchange["redirect_uri"].startswith("http://127.0.0.1:")
|
||||
assert "device_name=BoltClam" in opened["url"]
|
||||
|
||||
|
||||
def test_loopback_rejects_the_wrong_state():
|
||||
receiver = LoopbackAuthReceiver("expected-state")
|
||||
receiver.start()
|
||||
try:
|
||||
try:
|
||||
urllib.request.urlopen(
|
||||
f"{receiver.redirect_uri}?code=bad&state=wrong", timeout=5).read()
|
||||
assert False, "expected a 400"
|
||||
except urllib.error.HTTPError as exc:
|
||||
assert exc.code == 400
|
||||
assert receiver.event.is_set() is False
|
||||
|
||||
with urllib.request.urlopen(
|
||||
f"{receiver.redirect_uri}?code=good&state=expected-state",
|
||||
timeout=5) as response:
|
||||
assert response.status == 200
|
||||
result = receiver.wait(2)
|
||||
assert result["code"] == "good"
|
||||
finally:
|
||||
receiver.close()
|
||||
|
||||
|
||||
def test_restore_offline_uses_the_cached_account(tmp_path):
|
||||
with socket.socket() as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
port = sock.getsockname()[1]
|
||||
credential_path = tmp_path / "credentials.dat"
|
||||
CredentialStore(credential_path).save("bolt_live_cached")
|
||||
cache = AccountCache(tmp_path / "account.json")
|
||||
cache.save(ACCOUNT)
|
||||
auth = AuthService(
|
||||
f"http://127.0.0.1:{port}", install_id="offline",
|
||||
credentials=CredentialStore(credential_path), account_cache=cache)
|
||||
session = auth.restore()
|
||||
assert session.offline is True
|
||||
assert session.account["user"]["email"] == "aaron@example.com"
|
||||
@@ -0,0 +1,157 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import stat
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from boltclam.clamav import (
|
||||
ClamAVService,
|
||||
build_freshclam_args,
|
||||
build_scan_args,
|
||||
database_age,
|
||||
database_files,
|
||||
find_clamscan,
|
||||
parse_clamscan_line,
|
||||
parse_extra_options,
|
||||
parse_version_output,
|
||||
scan_targets,
|
||||
)
|
||||
|
||||
|
||||
def _script(path: Path, body: str = "") -> Path:
|
||||
path.write_text("#!/bin/sh\n" + body, encoding="utf-8")
|
||||
path.chmod(path.stat().st_mode | stat.S_IXUSR)
|
||||
return path
|
||||
|
||||
|
||||
def test_parse_version_output():
|
||||
assert parse_version_output("ClamAV 1.4.3/27452/Wed Oct 22 07:17:23 2026") == (
|
||||
"1.4.3", "27452", "Wed Oct 22 07:17:23 2026")
|
||||
assert parse_version_output("ClamAV 1.4.0") == ("1.4.0", "", "")
|
||||
assert parse_version_output("nonsense") == ("", "", "")
|
||||
|
||||
|
||||
def test_parse_extra_options_handles_windows_quotes():
|
||||
options = parse_extra_options(
|
||||
'--max-filesize=200M --exclude-dir="C:\\Program Files\\Skip" '
|
||||
'--database="C:\\My DB"')
|
||||
assert options == [
|
||||
"--max-filesize=200M",
|
||||
"--exclude-dir=C:\\Program Files\\Skip",
|
||||
"--database=C:\\My DB",
|
||||
]
|
||||
|
||||
|
||||
def test_parse_extra_options_refuses_destructive_flags():
|
||||
for text in ("--remove=yes", "--move=C:\\Quarantine", "--copy=/tmp"):
|
||||
with pytest.raises(ValueError):
|
||||
parse_extra_options(text)
|
||||
|
||||
|
||||
def test_parse_extra_options_rejects_malformed_quoting():
|
||||
with pytest.raises(ValueError):
|
||||
parse_extra_options('--exclude-dir="unterminated')
|
||||
|
||||
|
||||
def test_build_scan_args_uses_managed_database_and_excludes(tmp_path):
|
||||
db = tmp_path / "db"
|
||||
db.mkdir()
|
||||
(db / "main.cvd").write_bytes(b"x")
|
||||
quarantine = tmp_path / "quarantine"
|
||||
temp = tmp_path / "temp"
|
||||
args = build_scan_args(
|
||||
Path("/usr/bin/clamscan"), [tmp_path / "target"],
|
||||
database_dir=db, quarantine_dir=quarantine, temp_dir=temp,
|
||||
extra_options=["--max-filesize=200M"])
|
||||
assert args[0].endswith("clamscan")
|
||||
assert "--recursive=yes" in args
|
||||
assert "--infected" in args
|
||||
assert f"--database={db}" in args
|
||||
assert any(arg.startswith("--exclude-dir=") for arg in args)
|
||||
assert "--max-filesize=200M" in args
|
||||
assert args[-1] == str(tmp_path / "target")
|
||||
|
||||
|
||||
def test_build_scan_args_omits_database_when_no_signatures(tmp_path):
|
||||
args = build_scan_args(Path("/usr/bin/clamscan"), [tmp_path])
|
||||
assert not any(arg.startswith("--database=") for arg in args)
|
||||
|
||||
|
||||
def test_parse_clamscan_line_detection_windows_path():
|
||||
event = parse_clamscan_line(
|
||||
r"C:\Users\aaron\Downloads\invoice.pdf.exe: Win.Trojan.Agent-123 FOUND")
|
||||
assert event == {
|
||||
"kind": "detection",
|
||||
"path": r"C:\Users\aaron\Downloads\invoice.pdf.exe",
|
||||
"signature": "Win.Trojan.Agent-123",
|
||||
}
|
||||
|
||||
|
||||
def test_parse_clamscan_line_summary_progress_and_errors():
|
||||
assert parse_clamscan_line("Scanned files: 18234")["key"] == "scanned_files"
|
||||
assert parse_clamscan_line("Scanned files: 18234")["value"] == "18234"
|
||||
assert parse_clamscan_line("Engine version: 1.4.3")["value"] == "1.4.3"
|
||||
assert parse_clamscan_line("Scanning /home/a/file")["kind"] == "progress"
|
||||
assert parse_clamscan_line("/home/a/file: Access denied")["kind"] == "error"
|
||||
assert parse_clamscan_line("/home/a/file: Size limit exceeded")["kind"] == "skipped"
|
||||
assert parse_clamscan_line("") is None
|
||||
|
||||
|
||||
def test_scan_targets_deduplicates_injected_profiles(tmp_path):
|
||||
one = tmp_path / "one"
|
||||
two = tmp_path / "two"
|
||||
one.mkdir()
|
||||
two.mkdir()
|
||||
targets = scan_targets("homes", profiles=[one, two, one, Path("/does/not/exist")])
|
||||
assert targets == [one.resolve(), two.resolve()]
|
||||
|
||||
|
||||
def test_scan_targets_full_uses_injected_drives(tmp_path):
|
||||
drive = tmp_path / "drive"
|
||||
drive.mkdir()
|
||||
assert scan_targets("full", drives=[drive]) == [drive.resolve()]
|
||||
|
||||
|
||||
def test_database_detection_and_age(tmp_path):
|
||||
db = tmp_path / "db"
|
||||
db.mkdir()
|
||||
assert database_files(db) == []
|
||||
main = db / "main.cvd"
|
||||
main.write_bytes(b"signatures")
|
||||
old = time.time() - 3 * 86400
|
||||
os.utime(main, (old, old))
|
||||
assert database_files(db) == [main]
|
||||
date, age = database_age(db)
|
||||
assert date
|
||||
assert 2.8 < age < 3.2
|
||||
|
||||
|
||||
def test_clamav_service_reads_a_fake_engine(tmp_path):
|
||||
fake = _script(tmp_path / "clamscan", 'echo "ClamAV 1.4.3/27452/Wed Oct 22 07:17:23 2026"\n')
|
||||
service = ClamAVService(str(fake))
|
||||
info = service.info(force=True)
|
||||
assert info.available is True
|
||||
assert info.engine_version == "1.4.3"
|
||||
assert info.db_version == "27452"
|
||||
assert info.path == fake
|
||||
assert service.info(force=True).summary.startswith("1.4.3")
|
||||
|
||||
|
||||
def test_find_clamscan_accepts_a_file_or_directory(tmp_path):
|
||||
fake = _script(tmp_path / "clamscan")
|
||||
found, error = find_clamscan(str(fake))
|
||||
assert found == fake and error == ""
|
||||
found, error = find_clamscan(str(tmp_path))
|
||||
assert found == fake and error == ""
|
||||
missing, error = find_clamscan(str(tmp_path / "missing"))
|
||||
assert missing is None and "does not exist" in error
|
||||
|
||||
|
||||
def test_freshclam_arguments_target_managed_database(tmp_path):
|
||||
args = build_freshclam_args(Path("/usr/bin/freshclam"), tmp_path / "db")
|
||||
assert "--foreground" in args
|
||||
assert "--stdout" in args
|
||||
assert f"--datadir={tmp_path / 'db'}" in args
|
||||
@@ -0,0 +1,70 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from boltclam.config import AccountCache, AppSettings, SettingsStore
|
||||
|
||||
|
||||
def test_defaults_are_normalized_and_install_id_is_generated(isolated_env):
|
||||
settings = SettingsStore().load()
|
||||
assert settings.update_channel == "release"
|
||||
assert settings.scan_scope == "homes"
|
||||
assert settings.auto_action == "quarantine"
|
||||
assert settings.install_id
|
||||
assert len(settings.install_id) == 32
|
||||
|
||||
|
||||
def test_invalid_values_fall_back_to_defaults(isolated_env):
|
||||
settings = AppSettings(
|
||||
update_channel="nightly",
|
||||
scan_scope="somewhere",
|
||||
auto_action="explode",
|
||||
debug_logging=True,
|
||||
logging_enabled=False,
|
||||
api_base_url="",
|
||||
).normalized()
|
||||
assert settings.update_channel == "release"
|
||||
assert settings.scan_scope == "homes"
|
||||
assert settings.auto_action == "quarantine"
|
||||
assert settings.debug_logging is False
|
||||
assert settings.api_base_url == "https://api.themajesticnetwork.com"
|
||||
assert settings.install_id
|
||||
|
||||
|
||||
def test_round_trip_and_atomic_save(isolated_env):
|
||||
store = SettingsStore()
|
||||
settings = store.load()
|
||||
saved = store.update(update_channel="beta", extra_options="--max-filesize=200M",
|
||||
extra_watch_dirs=["~/Downloads/Test", ""])
|
||||
reloaded = SettingsStore().load()
|
||||
assert reloaded.update_channel == "beta"
|
||||
assert reloaded.extra_options == "--max-filesize=200M"
|
||||
assert len(reloaded.extra_watch_dirs) == 1
|
||||
assert saved.install_id == reloaded.install_id
|
||||
assert json.loads(store.path.read_text())["update_channel"] == "beta"
|
||||
|
||||
|
||||
def test_corrupt_settings_are_backed_up_and_defaults_load(isolated_env):
|
||||
store = SettingsStore()
|
||||
store.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
store.path.write_text("{not json", encoding="utf-8")
|
||||
loaded = SettingsStore().load()
|
||||
assert loaded.update_channel == "release"
|
||||
backups = list(store.path.parent.glob("settings.corrupt-*.json"))
|
||||
assert len(backups) == 1
|
||||
|
||||
|
||||
def test_public_snapshot_does_not_expose_install_id(isolated_env):
|
||||
settings = SettingsStore().load()
|
||||
public = settings.public_snapshot()
|
||||
assert "install_id" not in public
|
||||
assert public["update_channel"] == "release"
|
||||
|
||||
|
||||
def test_account_cache_round_trip_and_clear(isolated_env):
|
||||
cache = AccountCache()
|
||||
account = {"user": {"email": "a@example.com"}, "subscription": {"tier": "premium"}}
|
||||
cache.save(account)
|
||||
assert cache.load() == account
|
||||
cache.clear()
|
||||
assert cache.load() is None
|
||||
@@ -0,0 +1,79 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
from boltclam import paths
|
||||
from boltclam.config import SettingsStore
|
||||
from boltclam.logging_setup import redact
|
||||
from boltclam.logs import build_bundle, redact_paths, sanitize_text, upload_bundle
|
||||
from boltclam.models import ClamAVInfo, Detection
|
||||
from boltclam.storage import Storage
|
||||
from boltclam.system_info import collect
|
||||
|
||||
|
||||
def test_redaction_covers_tokens_and_emails():
|
||||
text = "key=bolt_live_abcdef password=hunter2 user aaron@example.com"
|
||||
redacted = sanitize_text(text)
|
||||
assert "bolt_live_abcdef" not in redacted
|
||||
assert "hunter2" not in redacted
|
||||
assert "aaron@example.com" not in redacted
|
||||
assert "[email]" in redacted or "[redacted]" in redacted
|
||||
|
||||
|
||||
def test_path_redaction_uses_userprofile(isolated_env):
|
||||
sample = str(Path.home() / "Downloads" / "bad.exe")
|
||||
assert "%USERPROFILE%" in redact_paths(sample)
|
||||
assert "Downloads" in redact_paths(sample)
|
||||
|
||||
|
||||
def test_bundle_contains_redacted_logs_and_scan_summary(isolated_env):
|
||||
settings = SettingsStore().load()
|
||||
storage = Storage()
|
||||
scan_id = storage.create_scan("homes", [str(Path.home() / "Downloads")])
|
||||
detection = Detection(path=str(Path.home() / "Downloads" / "bad.exe"),
|
||||
signature="Win.Test", size=10)
|
||||
detection.id = storage.add_detection(scan_id, detection)
|
||||
storage.finish_scan(scan_id, status="completed", scanned_files=10,
|
||||
infected_files=1, errors=0, duration_seconds=2, cancelled=False)
|
||||
paths.logs_dir().mkdir(parents=True, exist_ok=True)
|
||||
(paths.logs_dir() / "boltclam.log").write_text(
|
||||
f"token=bolt_live_supersecret user aaron@example.com path={Path.home()}/x\n",
|
||||
encoding="utf-8")
|
||||
info = ClamAVInfo(available=True, path=Path("/fake/clamscan"),
|
||||
engine_version="1.4.3", db_version="27452",
|
||||
db_path=Path("/fake/db"), db_date="2026-10-22",
|
||||
db_age_days=1.0)
|
||||
snapshot = collect(settings, info, storage, realtime_running=False, force=True)
|
||||
bundle = build_bundle(settings, info, system_snapshot=snapshot,
|
||||
account={"user": {"email": "aaron@example.com"}},
|
||||
realtime_roots=[str(Path.home() / "Downloads")])
|
||||
assert bundle.is_file()
|
||||
with zipfile.ZipFile(bundle) as archive:
|
||||
names = set(archive.namelist())
|
||||
assert {"manifest.json", "boltclam-logs.txt", "system-summary.json",
|
||||
"README.txt"} <= names
|
||||
logs = archive.read("boltclam-logs.txt").decode()
|
||||
assert "bolt_live_supersecret" not in logs
|
||||
assert "aaron@example.com" not in logs
|
||||
assert "%USERPROFILE%" in logs
|
||||
manifest = json.loads(archive.read("manifest.json"))
|
||||
assert manifest["app"] == "BoltClam"
|
||||
assert manifest["account"]["email"] == "aaron@example.com"
|
||||
summary = json.loads(archive.read("system-summary.json"))
|
||||
assert summary["stats"]["latest_scan"]["id"] == scan_id
|
||||
|
||||
|
||||
class FakeApi:
|
||||
def upload_diagnostics(self, path, metadata=None):
|
||||
return {"id": "42/test.zip", "bytes": Path(path).stat().st_size}
|
||||
|
||||
|
||||
def test_upload_bundle_returns_server_id(isolated_env):
|
||||
settings = SettingsStore().load()
|
||||
storage = Storage()
|
||||
info = ClamAVInfo(available=False, error="not installed")
|
||||
bundle = build_bundle(settings, info)
|
||||
result = upload_bundle(FakeApi(), bundle, settings)
|
||||
assert result["id"] == "42/test.zip"
|
||||
@@ -0,0 +1,100 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from boltclam.quarantine import QuarantineError, QuarantineService, delete_file, hash_file
|
||||
from boltclam.storage import Storage
|
||||
|
||||
|
||||
def _service(tmp_path) -> QuarantineService:
|
||||
storage = Storage(tmp_path / "history.db")
|
||||
return QuarantineService(storage, tmp_path / "quarantine")
|
||||
|
||||
|
||||
def test_quarantine_moves_records_and_writes_metadata(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
source = tmp_path / "malware.exe"
|
||||
source.write_bytes(b"malware bytes")
|
||||
digest = hash_file(source)
|
||||
item = service.quarantine_file(source, "Win.Test", scan_id=7, reason="Unit test")
|
||||
|
||||
assert not source.exists()
|
||||
destination = Path(item.quarantine_path)
|
||||
assert destination.is_file()
|
||||
assert destination.name.endswith(".quarantine")
|
||||
assert item.sha256 == digest
|
||||
assert service.storage.get_quarantine(item.id).signature == "Win.Test"
|
||||
metadata = json.loads((service.root / f"{item.id}.json").read_text())
|
||||
assert metadata["original_path"] == str(source)
|
||||
assert metadata["reason"] == "Unit test"
|
||||
|
||||
|
||||
def test_restore_to_original_or_chosen_path(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
source = tmp_path / "bad.exe"
|
||||
source.write_bytes(b"x")
|
||||
item = service.quarantine_file(source, "Sig")
|
||||
target = service.restore(item.id)
|
||||
assert target.read_bytes() == b"x"
|
||||
assert service.storage.get_quarantine(item.id).status == "restored"
|
||||
|
||||
source.write_bytes(b"y")
|
||||
item2 = service.quarantine_file(source, "Sig")
|
||||
chosen = tmp_path / "restored" / "safe.exe"
|
||||
restored = service.restore(item2.id, chosen)
|
||||
assert restored == chosen
|
||||
assert chosen.read_bytes() == b"y"
|
||||
|
||||
|
||||
def test_delete_quarantined_item_and_missing_original(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
source = tmp_path / "bad.exe"
|
||||
source.write_bytes(b"x")
|
||||
item = service.quarantine_file(source, "Sig")
|
||||
service.delete(item.id)
|
||||
assert not Path(item.quarantine_path).exists()
|
||||
assert service.storage.get_quarantine(item.id).status == "deleted"
|
||||
with pytest.raises(QuarantineError):
|
||||
service.restore(item.id)
|
||||
|
||||
|
||||
def test_refuses_symlink_and_directory(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
directory = tmp_path / "folder"
|
||||
directory.mkdir()
|
||||
with pytest.raises(QuarantineError):
|
||||
service.quarantine_file(directory, "Sig")
|
||||
real = tmp_path / "real.exe"
|
||||
real.write_bytes(b"x")
|
||||
link = tmp_path / "link.exe"
|
||||
try:
|
||||
link.symlink_to(real)
|
||||
except OSError:
|
||||
pytest.skip("symlinks unavailable")
|
||||
with pytest.raises(QuarantineError):
|
||||
service.quarantine_file(link, "Sig")
|
||||
with pytest.raises(QuarantineError):
|
||||
delete_file(directory)
|
||||
|
||||
|
||||
def test_delete_file_clears_readonly_bit(tmp_path):
|
||||
target = tmp_path / "readonly.exe"
|
||||
target.write_bytes(b"x")
|
||||
os.chmod(target, stat.S_IREAD)
|
||||
delete_file(target)
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
def test_restore_refuses_to_overwrite(tmp_path):
|
||||
service = _service(tmp_path)
|
||||
source = tmp_path / "bad.exe"
|
||||
source.write_bytes(b"x")
|
||||
item = service.quarantine_file(source, "Sig")
|
||||
source.write_bytes(b"replacement")
|
||||
with pytest.raises(QuarantineError):
|
||||
service.restore(item.id)
|
||||
@@ -0,0 +1,70 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from boltclam import realtime
|
||||
from boltclam.realtime import browser_download_roots, snapshot
|
||||
from boltclam.winlock import copy_for_scan
|
||||
|
||||
|
||||
def test_snapshot_finds_final_files_and_skips_partials(tmp_path):
|
||||
root = tmp_path / "Downloads"
|
||||
root.mkdir()
|
||||
(root / "old.exe").write_bytes(b"abc")
|
||||
(root / "new.crdownload").write_bytes(b"partial")
|
||||
(root / ".hidden").write_bytes(b"hidden")
|
||||
sub = root / "Sub"
|
||||
sub.mkdir()
|
||||
(sub / "inside.zip").write_bytes(b"zip")
|
||||
|
||||
found = snapshot([root])
|
||||
names = {Path(value[2]).name for value in found.values()}
|
||||
assert names == {"old.exe", "inside.zip"}
|
||||
|
||||
|
||||
def test_chromium_download_preference_is_watched(tmp_path, monkeypatch):
|
||||
local = tmp_path / "Local"
|
||||
roaming = tmp_path / "Roaming"
|
||||
monkeypatch.setenv("LOCALAPPDATA", str(local))
|
||||
monkeypatch.setenv("APPDATA", str(roaming))
|
||||
download = tmp_path / "Custom Downloads"
|
||||
download.mkdir()
|
||||
profile = local / "Google" / "Chrome" / "User Data" / "Default"
|
||||
profile.mkdir(parents=True)
|
||||
(profile / "Preferences").write_text(json.dumps({
|
||||
"download": {"default_directory": str(download)}}), encoding="utf-8")
|
||||
monkeypatch.setattr(realtime.paths, "browser_download_roots", lambda: [])
|
||||
|
||||
roots = browser_download_roots()
|
||||
assert download.resolve() in [path.resolve() for path in roots]
|
||||
|
||||
|
||||
def test_firefox_download_preference_is_watched(tmp_path, monkeypatch):
|
||||
local = tmp_path / "Local"
|
||||
roaming = tmp_path / "Roaming"
|
||||
monkeypatch.setenv("LOCALAPPDATA", str(local))
|
||||
monkeypatch.setenv("APPDATA", str(roaming))
|
||||
download = tmp_path / "Firefox Downloads"
|
||||
download.mkdir()
|
||||
profile = roaming / "Mozilla" / "Firefox" / "Profiles" / "abc.default"
|
||||
profile.mkdir(parents=True)
|
||||
escaped = str(download).replace("\\", "\\\\")
|
||||
(profile / "prefs.js").write_text(
|
||||
f'user_pref("browser.download.dir", "{escaped}");\n', encoding="utf-8")
|
||||
monkeypatch.setattr(realtime.paths, "browser_download_roots", lambda: [])
|
||||
|
||||
roots = browser_download_roots()
|
||||
assert download.resolve() in [path.resolve() for path in roots]
|
||||
|
||||
|
||||
def test_copy_for_scan_preserves_bytes(tmp_path):
|
||||
source = tmp_path / "download.exe"
|
||||
source.write_bytes(b"sample bytes")
|
||||
destination = tmp_path / "copy.bin"
|
||||
locker, exclusive = copy_for_scan(source, destination, attempts=1, delay=0)
|
||||
try:
|
||||
assert destination.read_bytes() == b"sample bytes"
|
||||
assert isinstance(exclusive, bool)
|
||||
finally:
|
||||
locker.close()
|
||||
@@ -0,0 +1,65 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from boltclam.models import Detection, QuarantineItem, RealtimeEvent
|
||||
from boltclam.storage import Storage
|
||||
|
||||
|
||||
def _storage(tmp_path) -> Storage:
|
||||
return Storage(tmp_path / "history.db")
|
||||
|
||||
|
||||
def test_scan_and_detection_lifecycle(tmp_path):
|
||||
storage = _storage(tmp_path)
|
||||
scan_id = storage.create_scan("homes", [r"C:\Users"], engine_version="1.4.3",
|
||||
db_version="27452")
|
||||
assert scan_id == 1
|
||||
detection = Detection(path=r"C:\Users\a\bad.exe", signature="Win.Test", size=42)
|
||||
detection.id = storage.add_detection(scan_id, detection)
|
||||
storage.finish_scan(scan_id, status="completed", scanned_files=100,
|
||||
infected_files=1, errors=2, duration_seconds=12.5,
|
||||
cancelled=False)
|
||||
|
||||
record = storage.get_scan(scan_id)
|
||||
assert record is not None
|
||||
assert record.scope == "homes"
|
||||
assert record.scope_label == "User homes"
|
||||
assert record.scanned_files == 100
|
||||
assert record.errors == 2
|
||||
assert record.paths == [r"C:\Users"]
|
||||
assert record.detections[0].signature == "Win.Test"
|
||||
|
||||
stats = storage.stats()
|
||||
assert stats["active_threats"] == 1
|
||||
assert stats["latest_scan"].id == scan_id
|
||||
|
||||
storage.update_detection_action(detection.id, "quarantined", "Q-1")
|
||||
assert storage.stats()["active_threats"] == 0
|
||||
assert storage.recent_detections()[0]["action"] == "quarantined"
|
||||
|
||||
|
||||
def test_quarantine_rows_and_counts(tmp_path):
|
||||
storage = _storage(tmp_path)
|
||||
item = QuarantineItem(
|
||||
id="Q-1", original_path=r"C:\bad.exe", quarantine_path=str(tmp_path / "q.bin"),
|
||||
signature="Win.Test", sha256="a" * 64, size=10, created_at="2026-10-25T10:00:00Z",
|
||||
scan_id=1)
|
||||
storage.add_quarantine(item)
|
||||
assert storage.quarantine_count() == 1
|
||||
assert storage.get_quarantine("Q-1").signature == "Win.Test"
|
||||
storage.set_quarantine_status("Q-1", "restored")
|
||||
assert storage.quarantine_count() == 0
|
||||
assert storage.list_quarantine()[0].status == "restored"
|
||||
|
||||
|
||||
def test_realtime_events_and_diagnostics_are_serializable(tmp_path):
|
||||
storage = _storage(tmp_path)
|
||||
storage.add_realtime_event(RealtimeEvent(
|
||||
created_at="2026-10-25T10:00:00Z", path=r"C:\Users\a\bad.exe",
|
||||
signature="Win.Test", action="quarantined", detail="Locked"))
|
||||
events = storage.list_realtime_events()
|
||||
assert events[0].action == "quarantined"
|
||||
snapshot = storage.diagnostics_snapshot()
|
||||
json.dumps(snapshot)
|
||||
assert snapshot["realtime"][0]["signature"] == "Win.Test"
|
||||
@@ -0,0 +1,86 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from PyQt6.QtCore import Qt
|
||||
|
||||
from boltclam.api_client import ApiClient
|
||||
from boltclam.auth import AuthService, CredentialStore
|
||||
from boltclam.clamav import ClamAVService
|
||||
from boltclam.config import AccountCache, SettingsStore
|
||||
from boltclam.context import AppContext
|
||||
from boltclam.quarantine import QuarantineService
|
||||
from boltclam.storage import Storage
|
||||
from boltclam.ui.login_dialog import LoginDialog
|
||||
from boltclam.ui.main_window import MainWindow
|
||||
|
||||
ACCOUNT = {
|
||||
"user": {"id": "1", "email": "aaron@example.com", "name": "Aaron",
|
||||
"username": "aaron"},
|
||||
"subscription": {"tier": "premium", "label": "Premium", "status": "active",
|
||||
"current_period_end": "2026-12-01T00:00:00Z"},
|
||||
"balance": {"weekly_remaining": 100, "topup": 0, "allowed": True},
|
||||
"features": {"licensed": True},
|
||||
}
|
||||
|
||||
|
||||
class FakeApi:
|
||||
def get_account(self):
|
||||
return ACCOUNT
|
||||
|
||||
def get_updates(self, *args, **kwargs):
|
||||
raise RuntimeError("offline")
|
||||
|
||||
|
||||
def _context(tmp_path) -> AppContext:
|
||||
store = SettingsStore()
|
||||
store.save(store.load())
|
||||
settings = store.update(last_update_check=time.time(), logging_enabled=False,
|
||||
realtime_enabled=False)
|
||||
storage = Storage()
|
||||
quarantine = QuarantineService(storage, tmp_path / "quarantine")
|
||||
auth = AuthService(
|
||||
"http://127.0.0.1:9", install_id=settings.install_id,
|
||||
credentials=CredentialStore(tmp_path / "credentials.dat"),
|
||||
account_cache=AccountCache(tmp_path / "account.json"))
|
||||
return AppContext(
|
||||
settings_store=store, settings=settings, storage=storage,
|
||||
quarantine=quarantine, clamav=ClamAVService(settings.clamav_path),
|
||||
auth=auth, session_key="bolt_live_fake", account=ACCOUNT,
|
||||
install_id=settings.install_id, api=FakeApi())
|
||||
|
||||
|
||||
def test_main_window_builds_four_pages_and_switches_tabs(qapp, isolated_env):
|
||||
context = _context(isolated_env)
|
||||
window = MainWindow(context)
|
||||
try:
|
||||
window.show()
|
||||
qapp.processEvents()
|
||||
assert window.stack.count() == 4
|
||||
assert [button.text() for button in window.nav_buttons] == [
|
||||
"Homepage", "Scan & results", "About & updates", "Settings"]
|
||||
assert window.scan_page.tabs.currentIndex() == 0
|
||||
window.scan_page.tabs.setCurrentIndex(1)
|
||||
assert window.scan_page.tabs.currentIndex() == 1
|
||||
window._navigate(3)
|
||||
assert window.stack.currentIndex() == 3
|
||||
assert window.settings_page.channel_combo.currentData() == "release"
|
||||
assert window.about_page.rows["account"].value_label.text() == "Aaron"
|
||||
finally:
|
||||
window._allow_close = True
|
||||
window.close()
|
||||
qapp.processEvents()
|
||||
|
||||
|
||||
def test_login_dialog_builds_without_starting_a_worker(qapp, isolated_env):
|
||||
auth = AuthService(
|
||||
"http://127.0.0.1:9", install_id="test",
|
||||
credentials=CredentialStore(isolated_env / "credentials.dat"),
|
||||
account_cache=AccountCache(isolated_env / "account.json"))
|
||||
dialog = LoginDialog(auth, notice="Test notice")
|
||||
try:
|
||||
assert "BoltClam" in dialog.windowTitle()
|
||||
assert dialog.browser_button.isEnabled()
|
||||
assert dialog.session is None
|
||||
finally:
|
||||
dialog.reject()
|
||||
@@ -0,0 +1,121 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from boltclam.models import UpdateAsset, UpdateFeed, UpdateRelease
|
||||
from boltclam.updates import (
|
||||
UpdateError,
|
||||
UpdateService,
|
||||
choose_asset,
|
||||
is_newer,
|
||||
parse_checksums,
|
||||
parse_version,
|
||||
)
|
||||
|
||||
|
||||
def test_version_parsing_and_comparison():
|
||||
assert parse_version("v1.2.3") == ((1, 2, 3), ())
|
||||
assert parse_version("1.2.3-beta.1")
|
||||
assert is_newer("v1.2.0", "1.1.9") is True
|
||||
assert is_newer("v1.0.0", "1.0.0") is False
|
||||
assert is_newer("v1.0.0", "1.0.0-beta.2") is True
|
||||
assert is_newer("v1.0.0-beta.2", "1.0.0") is False
|
||||
assert is_newer("v1.0.0-beta.2", "1.0.0-beta.1") is True
|
||||
assert is_newer("not-a-version", "1.0.0") is False
|
||||
|
||||
|
||||
def _release(tag="v1.2.0", prerelease=False, assets=None) -> UpdateRelease:
|
||||
return UpdateRelease(id="2", tag=tag, name="Test", notes="notes",
|
||||
prerelease=prerelease, published_at="2026-10-20T00:00:00Z",
|
||||
html_url="https://git.example", assets=assets or [])
|
||||
|
||||
|
||||
def test_choose_asset_prefers_a_setup_executable():
|
||||
release = _release(assets=[
|
||||
UpdateAsset("1", "BoltClam-1.2.0.zip", 5, "application/zip", "u1"),
|
||||
UpdateAsset("2", "BoltClam-1.2.0-Setup.exe", 5, "application/octet-stream", "u2"),
|
||||
])
|
||||
assert choose_asset(release).name == "BoltClam-1.2.0-Setup.exe"
|
||||
|
||||
|
||||
def test_choose_asset_falls_back_to_zip():
|
||||
release = _release(assets=[
|
||||
UpdateAsset("1", "notes.txt", 5, "text/plain", "u1"),
|
||||
UpdateAsset("2", "BoltClam-windows.zip", 5, "application/zip", "u2"),
|
||||
])
|
||||
assert choose_asset(release).name == "BoltClam-windows.zip"
|
||||
assert choose_asset(_release(assets=[])) is None
|
||||
|
||||
|
||||
def test_feed_from_api_preserves_release_fields():
|
||||
feed = UpdateFeed.from_api({
|
||||
"channel": "beta",
|
||||
"latest": {"id": "3", "tag": "v2.0.0-beta", "name": "Beta",
|
||||
"notes": "notes", "prerelease": True,
|
||||
"published_at": "2026-10-22T00:00:00Z",
|
||||
"assets": [{"id": "9", "name": "a.exe", "size": 12,
|
||||
"content_type": "x", "download_url": "u"}]},
|
||||
"releases": [],
|
||||
"checked_at": "2026-10-25T00:00:00Z",
|
||||
})
|
||||
assert feed.channel == "beta"
|
||||
assert feed.latest.tag == "v2.0.0-beta"
|
||||
assert feed.latest.assets[0].name == "a.exe"
|
||||
|
||||
|
||||
def test_parse_checksums():
|
||||
digest = "a" * 64
|
||||
assert parse_checksums(f"{digest} BoltClam-Setup.exe\n") == {
|
||||
"BoltClam-Setup.exe": digest}
|
||||
|
||||
|
||||
class FakeApi:
|
||||
def __init__(self, checksum_text: str | None = None, payload: bytes = b"payload"):
|
||||
self.checksum_text = checksum_text
|
||||
self.payload = payload
|
||||
|
||||
def download_asset(self, url, destination, progress=None):
|
||||
destination = Path(destination)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
data = self.checksum_text.encode() if self.checksum_text and "SUM" in url else self.payload
|
||||
destination.write_bytes(data)
|
||||
return destination, hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _download_release(checksum: str) -> UpdateRelease:
|
||||
asset = UpdateAsset("1", "BoltClam-Setup.exe", 7, "application/octet-stream",
|
||||
"https://api.example/asset")
|
||||
checksum_asset = UpdateAsset("2", "SHA256SUMS.txt", 100, "text/plain",
|
||||
"https://api.example/SUM")
|
||||
return _release(assets=[asset, checksum_asset])
|
||||
|
||||
|
||||
def test_download_verifies_a_matching_checksum(isolated_env):
|
||||
payload = b"payload"
|
||||
digest = hashlib.sha256(payload).hexdigest()
|
||||
release = _download_release(digest)
|
||||
service = UpdateService(FakeApi(f"{digest} BoltClam-Setup.exe\n", payload))
|
||||
result = service.download(release, release.assets[0])
|
||||
assert result.path.read_bytes() == payload
|
||||
assert result.verified is True
|
||||
assert result.expected_sha256 == digest
|
||||
|
||||
|
||||
def test_download_rejects_a_bad_checksum(isolated_env):
|
||||
release = _download_release("b" * 64)
|
||||
service = UpdateService(FakeApi(f"{'b' * 64} BoltClam-Setup.exe\n", b"payload"))
|
||||
with pytest.raises(UpdateError):
|
||||
service.download(release, release.assets[0])
|
||||
|
||||
|
||||
def test_check_returns_the_feed(isolated_env):
|
||||
class Api:
|
||||
def get_updates(self, channel, force=False):
|
||||
return {"channel": channel, "latest": None, "releases": []}
|
||||
|
||||
feed = UpdateService(Api(), "beta").check()
|
||||
assert feed.channel == "beta"
|
||||
assert feed.latest is None
|
||||
Reference in New Issue
Block a user