mirror of
https://github.com/CloakHQ/CloakBrowser.git
synced 2026-06-23 11:41:46 +02:00
feat: add Pro tier license validation and download routing
This commit is contained in:
CloakHQ
@@ -0,0 +1,408 @@
|
||||
"""Tests for the CloakBrowser Pro license module."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from cloakbrowser.download import BinaryVerificationError, ensure_binary
|
||||
from cloakbrowser.license import (
|
||||
LicenseInfo,
|
||||
get_pro_latest_version,
|
||||
resolve_license_key,
|
||||
validate_license,
|
||||
)
|
||||
|
||||
|
||||
# ── resolve_license_key ───────────────────────────────
|
||||
|
||||
|
||||
class TestResolveLicenseKey:
|
||||
def test_explicit_param_wins(self):
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_LICENSE_KEY": "env-key"}):
|
||||
assert resolve_license_key("explicit") == "explicit"
|
||||
|
||||
def test_env_var_fallback(self):
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_LICENSE_KEY": "env-key"}):
|
||||
assert resolve_license_key() == "env-key"
|
||||
|
||||
def test_returns_none_when_absent(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
os.environ.pop("CLOAKBROWSER_LICENSE_KEY", None)
|
||||
assert resolve_license_key() is None
|
||||
|
||||
def test_empty_string_param_uses_env(self):
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_LICENSE_KEY": "env-key"}):
|
||||
assert resolve_license_key("") == "env-key"
|
||||
|
||||
def test_file_fallback(self, tmp_path):
|
||||
key_file = tmp_path / "license.key"
|
||||
key_file.write_text("file-key-123\n")
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
os.environ.pop("CLOAKBROWSER_LICENSE_KEY", None)
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
assert resolve_license_key() == "file-key-123"
|
||||
|
||||
def test_env_takes_precedence_over_file(self, tmp_path):
|
||||
key_file = tmp_path / "license.key"
|
||||
key_file.write_text("file-key")
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_LICENSE_KEY": "env-key"}):
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
assert resolve_license_key() == "env-key"
|
||||
|
||||
def test_no_file_returns_none(self, tmp_path):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
os.environ.pop("CLOAKBROWSER_LICENSE_KEY", None)
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
assert resolve_license_key() is None
|
||||
|
||||
|
||||
# ── validate_license ──────────────────────────────────
|
||||
|
||||
|
||||
class TestValidateLicense:
|
||||
def test_fresh_cache_skips_server(self, tmp_path):
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "team",
|
||||
"expires": "2026-12-01",
|
||||
"validated_at": time.time(),
|
||||
}))
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post") as mock_post:
|
||||
result = validate_license("test-key")
|
||||
|
||||
mock_post.assert_not_called()
|
||||
assert result is not None
|
||||
assert result.valid is True
|
||||
assert result.plan == "team"
|
||||
|
||||
def test_stale_cache_calls_server(self, tmp_path):
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": None,
|
||||
"validated_at": time.time() - 90000, # 25 hours ago
|
||||
}))
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": True, "plan": "solo", "expires": None}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp) as mock_post:
|
||||
result = validate_license("test-key")
|
||||
|
||||
mock_post.assert_called_once()
|
||||
assert result is not None
|
||||
assert result.valid is True
|
||||
|
||||
def test_server_success(self, tmp_path):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": True, "plan": "business", "expires": "2026-07-13"}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp):
|
||||
result = validate_license("pro-key")
|
||||
|
||||
assert result is not None
|
||||
assert result.valid is True
|
||||
assert result.plan == "business"
|
||||
assert result.expires == "2026-07-13"
|
||||
|
||||
def test_server_rejection(self, tmp_path):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": False, "plan": "solo", "expires": None}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp):
|
||||
result = validate_license("bad-key")
|
||||
|
||||
assert result is not None
|
||||
assert result.valid is False
|
||||
|
||||
def test_server_unreachable_uses_stale_cache(self, tmp_path):
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": "2026-12-01",
|
||||
"validated_at": time.time() - 90000,
|
||||
}))
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", side_effect=Exception("timeout")):
|
||||
result = validate_license("test-key")
|
||||
|
||||
assert result is not None
|
||||
assert result.valid is True
|
||||
|
||||
def test_server_unreachable_no_cache_returns_none(self, tmp_path):
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", side_effect=Exception("timeout")):
|
||||
result = validate_license("test-key")
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_cache_stores_hash_not_raw_key(self, tmp_path):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": True, "plan": "solo", "expires": None}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp):
|
||||
validate_license("secret-key-123")
|
||||
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
content = cache_path.read_text()
|
||||
assert "secret-key-123" not in content
|
||||
expected_sha = hashlib.sha256(b"secret-key-123").hexdigest()
|
||||
assert expected_sha in content
|
||||
|
||||
def test_expired_license_rejected_from_cache(self, tmp_path):
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": "2020-01-01T00:00:00+00:00",
|
||||
"validated_at": time.time(),
|
||||
}))
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
result = validate_license("test-key")
|
||||
|
||||
assert result is not None
|
||||
assert result.valid is False
|
||||
|
||||
def test_expired_license_naive_date_rejected(self, tmp_path):
|
||||
"""Date-only string (naive datetime) should also be detected as expired."""
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": "2020-01-01",
|
||||
"validated_at": time.time(),
|
||||
}))
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
result = validate_license("test-key")
|
||||
|
||||
assert result is not None
|
||||
assert result.valid is False
|
||||
|
||||
def test_wrong_key_cache_ignored(self, tmp_path):
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": "other-hash",
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": None,
|
||||
"validated_at": time.time(),
|
||||
}))
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": True, "plan": "solo", "expires": None}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp) as mock_post:
|
||||
validate_license("different-key")
|
||||
|
||||
mock_post.assert_called_once()
|
||||
|
||||
def test_corrupted_validated_at_does_not_crash(self, tmp_path):
|
||||
"""A non-numeric validated_at must be treated as an absent cache, not crash."""
|
||||
cache_path = tmp_path / ".license_cache"
|
||||
key_sha = hashlib.sha256(b"test-key").hexdigest()
|
||||
cache_path.write_text(json.dumps({
|
||||
"key_sha256": key_sha,
|
||||
"valid": True,
|
||||
"plan": "solo",
|
||||
"expires": None,
|
||||
"validated_at": "not-a-number",
|
||||
}))
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"valid": True, "plan": "solo", "expires": None}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.post", return_value=mock_resp) as mock_post:
|
||||
result = validate_license("test-key")
|
||||
|
||||
mock_post.assert_called_once() # corrupted cache ignored → server hit
|
||||
assert result is not None
|
||||
assert result.valid is True
|
||||
|
||||
|
||||
# ── get_pro_latest_version ────────────────────────────
|
||||
|
||||
|
||||
class TestGetProLatestVersion:
|
||||
def test_fetches_version(self, tmp_path):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"version": "147.0.1234.5"}
|
||||
mock_resp.raise_for_status = MagicMock()
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.get", return_value=mock_resp):
|
||||
version = get_pro_latest_version()
|
||||
|
||||
assert version == "147.0.1234.5"
|
||||
|
||||
def test_rate_limited(self, tmp_path):
|
||||
marker = tmp_path / ".last_pro_version_check"
|
||||
marker.write_text("147.0.1234.5")
|
||||
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.get") as mock_get:
|
||||
version = get_pro_latest_version()
|
||||
|
||||
mock_get.assert_not_called()
|
||||
assert version == "147.0.1234.5"
|
||||
|
||||
def test_network_error_returns_none(self, tmp_path):
|
||||
with patch("cloakbrowser.license.get_cache_dir", return_value=tmp_path):
|
||||
with patch("cloakbrowser.license.httpx.get", side_effect=Exception("network")):
|
||||
version = get_pro_latest_version()
|
||||
|
||||
assert version is None
|
||||
|
||||
|
||||
# ── Config pro parameter ──────────────────────────────
|
||||
|
||||
|
||||
class TestConfigPro:
|
||||
def test_binary_dir_pro_suffix(self, tmp_path):
|
||||
from cloakbrowser.config import get_binary_dir
|
||||
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_CACHE_DIR": str(tmp_path)}):
|
||||
normal = get_binary_dir("147.0.0.0")
|
||||
pro = get_binary_dir("147.0.0.0", pro=True)
|
||||
|
||||
assert str(normal).endswith("chromium-147.0.0.0")
|
||||
assert str(pro).endswith("chromium-147.0.0.0-pro")
|
||||
|
||||
def test_binary_dir_default_no_suffix(self, tmp_path):
|
||||
from cloakbrowser.config import get_binary_dir
|
||||
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_CACHE_DIR": str(tmp_path)}):
|
||||
normal = get_binary_dir("147.0.0.0")
|
||||
|
||||
assert not str(normal).endswith("-pro")
|
||||
|
||||
def test_effective_version_pro_marker(self, tmp_path):
|
||||
from cloakbrowser.config import get_effective_version, get_platform_tag
|
||||
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_CACHE_DIR": str(tmp_path)}):
|
||||
tag = get_platform_tag()
|
||||
marker = tmp_path / f"latest_pro_version_{tag}"
|
||||
marker.write_text("147.0.5555.1")
|
||||
|
||||
# Create the binary so effective version returns it
|
||||
from cloakbrowser.config import get_binary_path
|
||||
bp = get_binary_path("147.0.5555.1", pro=True)
|
||||
bp.parent.mkdir(parents=True, exist_ok=True)
|
||||
bp.write_text("fake")
|
||||
|
||||
version = get_effective_version(pro=True)
|
||||
|
||||
assert version == "147.0.5555.1"
|
||||
|
||||
|
||||
# ── binary_info tier reporting ────────────────────────
|
||||
|
||||
|
||||
class TestBinaryInfoTier:
|
||||
"""binary_info() reports tier from the binary actually on disk — NOT from a
|
||||
cached license, which can disagree with what's installed or the active key."""
|
||||
|
||||
def test_free_when_no_pro_binary_even_if_license_cached(self, tmp_path):
|
||||
from cloakbrowser.download import binary_info
|
||||
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_CACHE_DIR": str(tmp_path)}, clear=False):
|
||||
# A valid, fresh license is cached...
|
||||
(tmp_path / ".license_cache").write_text(json.dumps({
|
||||
"key_sha256": hashlib.sha256(b"cb_x").hexdigest(),
|
||||
"valid": True, "plan": "solo", "expires": None,
|
||||
"validated_at": time.time(),
|
||||
}))
|
||||
# ...but no Pro binary is on disk → must report free, not pro.
|
||||
info = binary_info()
|
||||
|
||||
assert info["tier"] == "free"
|
||||
|
||||
def test_pro_when_pro_binary_installed(self, tmp_path):
|
||||
from cloakbrowser.config import get_binary_path, get_platform_tag
|
||||
from cloakbrowser.download import binary_info
|
||||
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_CACHE_DIR": str(tmp_path)}, clear=False):
|
||||
tag = get_platform_tag()
|
||||
(tmp_path / f"latest_pro_version_{tag}").write_text("147.0.5555.1")
|
||||
bp = get_binary_path("147.0.5555.1", pro=True)
|
||||
bp.parent.mkdir(parents=True, exist_ok=True)
|
||||
bp.write_text("fake")
|
||||
bp.chmod(0o755)
|
||||
info = binary_info()
|
||||
|
||||
assert info["tier"] == "pro"
|
||||
assert info["version"] == "147.0.5555.1"
|
||||
|
||||
|
||||
# ── ensure_binary Pro routing (fail-closed vs fall-back) ──────────────────────
|
||||
|
||||
|
||||
class TestEnsureBinaryProRouting:
|
||||
"""A valid-license user is NEVER silently downgraded to the free binary. Both
|
||||
a tampering signal (verification failure) and a transient failure
|
||||
(network/server) surface a clear error — they differ only in the message:
|
||||
tampering is re-raised verbatim (security, no 'retry'); transient is rewrapped
|
||||
as an actionable 'Pro binary unavailable, retry' error carrying the cause."""
|
||||
|
||||
def test_verification_failure_propagates_verbatim(self):
|
||||
"""A BinaryVerificationError must surface verbatim — never reach free."""
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_DOWNLOAD_URL": ""}, clear=False), \
|
||||
patch("cloakbrowser.download.get_local_binary_override", return_value=None), \
|
||||
patch("cloakbrowser.license.resolve_license_key", return_value="cb_x"), \
|
||||
patch("cloakbrowser.license.validate_license",
|
||||
return_value=LicenseInfo(valid=True, plan="solo", expires=None)), \
|
||||
patch("cloakbrowser.download._ensure_pro_binary",
|
||||
side_effect=BinaryVerificationError("bad signature")), \
|
||||
patch("cloakbrowser.download.check_platform_available",
|
||||
side_effect=AssertionError("MUST NOT reach the free-tier path")):
|
||||
with pytest.raises(BinaryVerificationError, match="bad signature"):
|
||||
ensure_binary("cb_x")
|
||||
|
||||
def test_transient_failure_hard_errors_not_free(self):
|
||||
"""A transient Pro failure must surface a clear, actionable error carrying
|
||||
the underlying cause — NOT silently download the free binary."""
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_DOWNLOAD_URL": ""}, clear=False), \
|
||||
patch("cloakbrowser.download.get_local_binary_override", return_value=None), \
|
||||
patch("cloakbrowser.license.resolve_license_key", return_value="cb_x"), \
|
||||
patch("cloakbrowser.license.validate_license",
|
||||
return_value=LicenseInfo(valid=True, plan="solo", expires=None)), \
|
||||
patch("cloakbrowser.download._ensure_pro_binary",
|
||||
side_effect=RuntimeError("network blip")), \
|
||||
patch("cloakbrowser.download.check_platform_available",
|
||||
side_effect=AssertionError("MUST NOT reach the free-tier path")):
|
||||
with pytest.raises(RuntimeError, match="Pro binary unavailable: network blip"):
|
||||
ensure_binary("cb_x")
|
||||
@@ -21,8 +21,10 @@ from cloakbrowser.config import (
|
||||
get_platform_tag,
|
||||
)
|
||||
from cloakbrowser.download import (
|
||||
BinaryVerificationError,
|
||||
_check_wrapper_update,
|
||||
_download_and_extract,
|
||||
_download_pro_binary,
|
||||
_fetch_checksums,
|
||||
_fetch_signed_manifest,
|
||||
_get_latest_chromium_version,
|
||||
@@ -31,6 +33,7 @@ from cloakbrowser.download import (
|
||||
_should_check_for_update,
|
||||
_verify_checksum,
|
||||
_verify_download_checksum,
|
||||
_verify_pro_download,
|
||||
_verify_signature,
|
||||
_write_version_marker,
|
||||
check_for_update,
|
||||
@@ -736,6 +739,121 @@ class TestVerifyDownloadChecksumSigned:
|
||||
mocked.assert_not_called()
|
||||
|
||||
|
||||
class TestVerifyProDownloadSigned:
|
||||
"""_verify_pro_download: Pro binaries get the SAME non-bypassable signature
|
||||
check as the free official path (parity — closes the Pro M1 gap)."""
|
||||
|
||||
PRO_VERSION = "147.0.1.0"
|
||||
|
||||
def _hash(self, data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
def _tarball(self) -> str:
|
||||
return get_download_url().rsplit("/", 1)[-1]
|
||||
|
||||
def _mock_fetch(self, manifest: bytes, sig: bytes):
|
||||
"""httpx.get stub: returns the .sig for *.sig URLs, manifest otherwise."""
|
||||
def mock_get(url, **kwargs):
|
||||
resp = MagicMock()
|
||||
resp.raise_for_status = MagicMock()
|
||||
resp.content = sig if url.endswith(".sig") else manifest
|
||||
return resp
|
||||
return mock_get
|
||||
|
||||
def test_valid_pro_manifest_passes(self, tmp_path):
|
||||
priv, pub_b64 = _make_key()
|
||||
archive = tmp_path / "binary"
|
||||
archive.write_bytes(b"the real pro binary")
|
||||
manifest = (
|
||||
f"version={self.PRO_VERSION}\n"
|
||||
f"{self._hash(b'the real pro binary')} {self._tarball()}\n"
|
||||
).encode()
|
||||
sig = _sign(priv, manifest)
|
||||
with patch("cloakbrowser.download.BINARY_SIGNING_PUBKEYS", [pub_b64]), \
|
||||
patch("cloakbrowser.download.httpx.get", side_effect=self._mock_fetch(manifest, sig)):
|
||||
_verify_pro_download(archive, self.PRO_VERSION) # no raise
|
||||
|
||||
def test_skip_checksum_does_not_bypass(self, tmp_path):
|
||||
"""CLOAKBROWSER_SKIP_CHECKSUM must NOT weaken Pro verification (the point)."""
|
||||
priv, pub_b64 = _make_key()
|
||||
archive = tmp_path / "binary"
|
||||
archive.write_bytes(b"a malicious pro binary") # bytes differ from manifest
|
||||
manifest = (
|
||||
f"version={self.PRO_VERSION}\n"
|
||||
f"{self._hash(b'the real pro binary')} {self._tarball()}\n"
|
||||
).encode()
|
||||
sig = _sign(priv, manifest)
|
||||
with patch.dict(os.environ, {"CLOAKBROWSER_SKIP_CHECKSUM": "true"}), \
|
||||
patch("cloakbrowser.download.BINARY_SIGNING_PUBKEYS", [pub_b64]), \
|
||||
patch("cloakbrowser.download.httpx.get", side_effect=self._mock_fetch(manifest, sig)):
|
||||
with pytest.raises(RuntimeError, match="Checksum verification failed"):
|
||||
_verify_pro_download(archive, self.PRO_VERSION)
|
||||
|
||||
def test_missing_manifest_is_transient_not_tampering(self, tmp_path):
|
||||
"""A failed manifest FETCH is transient (router falls back to free), so it
|
||||
must be a plain RuntimeError — NOT a BinaryVerificationError, which the
|
||||
router re-raises as a hard failure."""
|
||||
archive = tmp_path / "binary"
|
||||
archive.write_bytes(b"x")
|
||||
with patch("cloakbrowser.download.httpx.get", side_effect=Exception("404")):
|
||||
with pytest.raises(RuntimeError) as ei:
|
||||
_verify_pro_download(archive, self.PRO_VERSION)
|
||||
assert not isinstance(ei.value, BinaryVerificationError)
|
||||
|
||||
def test_wrong_version_fails_downgrade(self, tmp_path):
|
||||
"""A genuinely-signed Pro manifest for a DIFFERENT version is rejected."""
|
||||
priv, pub_b64 = _make_key()
|
||||
archive = tmp_path / "binary"
|
||||
archive.write_bytes(b"the real pro binary")
|
||||
manifest = (
|
||||
f"version=1.0.0.0\n" # declares old version, we ask for PRO_VERSION
|
||||
f"{self._hash(b'the real pro binary')} {self._tarball()}\n"
|
||||
).encode()
|
||||
sig = _sign(priv, manifest)
|
||||
with patch("cloakbrowser.download.BINARY_SIGNING_PUBKEYS", [pub_b64]), \
|
||||
patch("cloakbrowser.download.httpx.get", side_effect=self._mock_fetch(manifest, sig)):
|
||||
with pytest.raises(RuntimeError, match="Version mismatch"):
|
||||
_verify_pro_download(archive, self.PRO_VERSION)
|
||||
|
||||
def test_tampered_manifest_fails_signature(self, tmp_path):
|
||||
"""A manifest tampered after signing fails the signature gate (not the hash)."""
|
||||
priv, pub_b64 = _make_key()
|
||||
archive = tmp_path / "binary"
|
||||
archive.write_bytes(b"the real pro binary")
|
||||
manifest = (
|
||||
f"version={self.PRO_VERSION}\n"
|
||||
f"{self._hash(b'the real pro binary')} {self._tarball()}\n"
|
||||
).encode()
|
||||
sig = _sign(priv, manifest)
|
||||
tampered = manifest.replace(self._tarball().encode(), b"evil.tar.gz")
|
||||
with patch("cloakbrowser.download.BINARY_SIGNING_PUBKEYS", [pub_b64]), \
|
||||
patch("cloakbrowser.download.httpx.get", side_effect=self._mock_fetch(tampered, sig)):
|
||||
with pytest.raises(RuntimeError, match="signature verification failed"):
|
||||
_verify_pro_download(archive, self.PRO_VERSION)
|
||||
|
||||
|
||||
class TestProDownloadVersionPinned:
|
||||
"""The Pro download must request the explicit version, NOT /latest, so the
|
||||
served artifact matches the version-pinned signed manifest it's verified
|
||||
against (no latest-advances TOCTOU)."""
|
||||
|
||||
def test_download_url_is_version_pinned(self):
|
||||
from cloakbrowser.config import DOWNLOAD_BASE_URL
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_download_file(url, dest, headers=None):
|
||||
captured["url"] = url
|
||||
|
||||
with patch("cloakbrowser.download._download_file", side_effect=fake_download_file), \
|
||||
patch("cloakbrowser.download._verify_pro_download"), \
|
||||
patch("cloakbrowser.download._extract_archive"):
|
||||
_download_pro_binary("147.0.1.0", "cb_key")
|
||||
|
||||
assert captured["url"] == f"{DOWNLOAD_BASE_URL}/api/download/147.0.1.0"
|
||||
assert not captured["url"].endswith("/latest")
|
||||
|
||||
|
||||
class TestVersionBinding:
|
||||
"""The 'version=<v>' line: read by new wrappers, ignored by old parsers."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user