diff --git a/README.md b/README.md index bc74a41..7151cec 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # adb-sideload-free-apps -Install free Android apps via ADB from F-Droid, GitHub Releases, GitLab Releases, or Codeberg Releases. +Install free Android apps via ADB from F-Droid, GitHub Releases, GitLab Releases, Codeberg Releases, or generic HTTPS pages. ## Requirements @@ -39,11 +39,18 @@ See `config.example.toml`. Supported sources: | `github` | `url` (repository) | | `gitlab` | `url` (projet) | | `codeberg` | `url` (repository Codeberg) | +| `generic` | `url`, `asset_pattern` (HTTPS page or directory) | Every application also requires `name` and `package`. For GitHub, GitLab, and Codeberg, set `asset_pattern` when a release contains more than one suitable APK. Debug, test and unsigned APKs are rejected by default. +The `generic` source follows HTTPS anchor links from `url`. Use ordered +`intermediate_patterns` to navigate directories; when several links match, a +pattern with numeric capture groups selects the greatest numeric tuple. +`asset_pattern` is required and must match exactly one final APK. Optional +`headers` and `sha256` are supported; header values are never logged. + `aapt2` is used to validate the APK package ID before installation, with `aapt` as a fallback for older Android SDK Build Tools. At least one of them must be available in `PATH`. The script refuses to install an APK when neither diff --git a/config.example.toml b/config.example.toml index 6108538..67bbb8a 100644 --- a/config.example.toml +++ b/config.example.toml @@ -55,3 +55,17 @@ url = "https://f-droid.org/packages/org.mozilla.fennec_fdroid/" # package = "app.fedilab.android" # url = "https://codeberg.org/tom79/Fedilab" # asset_pattern = "^Fedilab-Google-.*\\.apk$" + +# --- Generic HTTPS page or directory --- +# Follow anchor links through directories, then select exactly one APK. +# intermediate_patterns are optional and applied in order to URL path components. + +# [[apps]] +# name = "Firefox" +# source = "generic" +# package = "org.mozilla.firefox" +# url = "https://download.example.org/releases/" +# intermediate_patterns = ["(?:[0-9]+\\.)+[0-9]+/$", "android/$"] +# asset_pattern = "^firefox-.*\\.apk$" +# headers = { Accept = "text/html" } +# sha256 = "0000000000000000000000000000000000000000000000000000000000000000" diff --git a/pyproject.toml b/pyproject.toml index 81c424f..0a1859c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "adb-sideload-free-apps" version = "0.1.0" -description = "Install free Android apps via ADB from F-Droid, GitHub, GitLab, or Codeberg" +description = "Install free Android apps via ADB from F-Droid, GitHub, GitLab, Codeberg, or generic HTTPS pages" requires-python = ">=3.11" license = { text = "MIT" } readme = "README.md" diff --git a/src/sideload.py b/src/sideload.py index 157a8c0..a0459d3 100644 --- a/src/sideload.py +++ b/src/sideload.py @@ -1,4 +1,4 @@ -"""ADB Sideload Free Apps — install free Android apps from F-Droid, GitHub, GitLab, or Codeberg via ADB.""" +"""ADB Sideload Free Apps — install free Android apps from release APIs or generic HTTPS pages via ADB.""" from __future__ import annotations @@ -13,6 +13,7 @@ import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass +from html.parser import HTMLParser from pathlib import Path if sys.version_info >= (3, 11): @@ -79,6 +80,8 @@ class Config: raise ValueError("each app requires name and package") if app.source not in SOURCE_REGISTRY: raise ValueError(f"unknown source type '{app.source}'") + if app.source == "generic": + _validate_generic_config(app) pattern = app.config.get("asset_pattern") if pattern: try: @@ -129,18 +132,49 @@ def _make_apk_stem(release: ReleaseInfo) -> str: return f"{release.app_entry.app_id}_{release.version}".replace("/", "_") -def _atomic_download(url: str, dest: Path) -> Path: +def _atomic_download( + url: str, + dest: Path, + extra_headers: dict[str, str] | None = None, + *, + conditional_headers: dict[str, str] | None = None, + same_origin: bool = False, + return_metadata: bool = False, +) -> Path | tuple[Path | None, str, dict[str, str], int]: parsed = urllib.parse.urlparse(url) if parsed.scheme != "https": raise ValueError(f"only HTTPS downloads are allowed: {url}") dest.parent.mkdir(parents=True, exist_ok=True) tmp = dest.with_suffix(dest.suffix + ".tmp") try: - request = urllib.request.Request(url, headers={"User-Agent": "adb-sideload/0.1"}) - with urllib.request.urlopen(request, timeout=30) as resp: + headers = dict(extra_headers or {}) + if conditional_headers: + headers.update(conditional_headers) + headers["User-Agent"] = "adb-sideload/0.1" + request = urllib.request.Request(url, headers=headers) + opener = _generic_opener(url if same_origin else None, same_origin=same_origin) + with opener.open(request, timeout=30) as resp: + final_url = resp.geturl() + if not isinstance(final_url, str): + final_url = url + if urllib.parse.urlsplit(final_url).scheme != "https": + raise ValueError("HTTPS download redirected to a non-HTTPS URL") + response_headers = dict(resp.headers.items()) + status = resp.getcode() or 200 + if status == 304: + if return_metadata: + return None, final_url, response_headers, status + raise ValueError("received HTTP 304 without a reusable cache entry") with tmp.open("wb") as fh: shutil.copyfileobj(resp, fh) tmp.rename(dest) + if return_metadata: + return dest, final_url, response_headers, status + except urllib.error.HTTPError as exc: + if exc.code == 304 and return_metadata: + return None, url, dict(exc.headers.items()), 304 + tmp.unlink(missing_ok=True) + raise except BaseException: tmp.unlink(missing_ok=True) raise @@ -170,10 +204,287 @@ def _http_get(url: str, extra_headers: dict | None = None) -> bytes: return resp.read() except urllib.error.HTTPError as exc: raise RuntimeError(f"HTTP {exc.code} fetching {url}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"network error fetching {url}: {exc.reason}") from exc + + +def _generic_http_get(url: str, headers: dict[str, str]) -> bytes: + body, _, _, _ = _generic_fetch(url, headers) + return body + + +def _same_origin(first_url: str, second_url: str) -> bool: + first = urllib.parse.urlsplit(first_url) + second = urllib.parse.urlsplit(second_url) + first_port = first.port or 443 + second_port = second.port or 443 + return ( + first.scheme == second.scheme == "https" + and (first.hostname or "").lower() == (second.hostname or "").lower() + and first_port == second_port + ) + + +class _HttpsOnlyRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, origin_url: str | None = None, same_origin: bool = False): + super().__init__() + self.origin_url = origin_url + self.same_origin = same_origin + + def redirect_request(self, req, fp, code, msg, newurl, headers): + if urllib.parse.urlsplit(newurl).scheme != "https": + raise ValueError("generic HTTPS request redirected to a non-HTTPS URL") + if self.same_origin and self.origin_url and not _same_origin(self.origin_url, newurl): + raise ValueError("generic HTTPS redirect changed origin") + # Keep configured headers on each allowed redirect. The source is + # explicitly configured by the user, so this also supports CDN hops. + redirected = urllib.request.Request( + newurl, + headers=dict(req.header_items()), + origin_req_host=req.origin_req_host, + unverifiable=True, + method=req.get_method(), + ) + return redirected + + +def _generic_opener(origin_url: str | None = None, *, same_origin: bool = True): + return urllib.request.build_opener( + _HttpsOnlyRedirectHandler(origin_url, same_origin=same_origin) + ) + + +def _generic_fetch( + url: str, headers: dict[str, str], *, conditional: dict[str, str] | None = None +) -> tuple[bytes, str, dict[str, str], int]: + if urllib.parse.urlsplit(url).scheme != "https": + raise ValueError("generic source accepts HTTPS URLs only") + request_headers = dict(headers) + if conditional: + request_headers.update(conditional) + request_headers["User-Agent"] = "adb-sideload/0.1" + try: + req = urllib.request.Request(url, headers=request_headers) + with _generic_opener(url, same_origin=True).open(req, timeout=30) as resp: + final_url = resp.geturl() + if not isinstance(final_url, str): + final_url = url + if urllib.parse.urlsplit(final_url).scheme != "https": + raise ValueError("generic HTTPS request redirected to a non-HTTPS URL") + status = resp.getcode() or 200 + return resp.read(), final_url, dict(resp.headers.items()), status + except urllib.error.HTTPError as exc: + if exc.code == 304: + return b"", url, dict(exc.headers.items()), 304 + raise RuntimeError(f"HTTP {exc.code} fetching {url}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"network error fetching {url}: {exc.reason}") from exc +class _AnchorParser(HTMLParser): + def __init__(self): + super().__init__(convert_charrefs=True) + self.hrefs: list[str] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + if tag.lower() != "a": + return + href = dict(attrs).get("href") + if href: + self.hrefs.append(href) + + +def _anchor_urls(page_url: str, body: bytes) -> list[str]: + parser = _AnchorParser() + parser.feed(body.decode("utf-8", errors="replace")) + urls = [] + for href in parser.hrefs: + resolved = urllib.parse.urljoin(page_url, href) + if urllib.parse.urlsplit(resolved).scheme == "https": + urls.append(resolved) + return urls + + +def _decoded_path_component(url: str) -> str: + path = urllib.parse.unquote(urllib.parse.urlsplit(url).path) + return path.rsplit("/", 1)[-1] if not path.endswith("/") else path.rsplit("/", 2)[-2] + "/" + + +def _validate_generic_config(app: AppEntry) -> None: + url = app.config.get("url") + parsed = urllib.parse.urlsplit(url) if isinstance(url, str) else None + if not parsed or parsed.scheme != "https" or not parsed.netloc: + raise ValueError(f"url must be an absolute HTTPS URL for {app.name}") + pattern = app.config.get("asset_pattern") + if not isinstance(pattern, str) or not pattern: + raise ValueError(f"asset_pattern is required for generic source '{app.name}'") + try: + re.compile(pattern) + except re.error as exc: + raise ValueError(f"invalid asset_pattern for {app.name}: {exc}") from exc + intermediates = app.config.get("intermediate_patterns", []) + if not isinstance(intermediates, list) or not all(isinstance(p, str) for p in intermediates): + raise ValueError(f"intermediate_patterns must be a list of strings for {app.name}") + for intermediate in intermediates: + try: + re.compile(intermediate) + except re.error as exc: + raise ValueError(f"invalid intermediate_patterns for {app.name}: {exc}") from exc + headers = app.config.get("headers", {}) + if not isinstance(headers, dict) or not all( + isinstance(k, str) and isinstance(v, str) for k, v in headers.items() + ): + raise ValueError(f"headers must be a mapping of strings for {app.name}") + sha256 = app.config.get("sha256") + if sha256 is not None and ( + not isinstance(sha256, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", sha256) + ): + raise ValueError(f"sha256 must be exactly 64 hexadecimal characters for {app.name}") + + +# ── generic HTTPS HTML source ────────────────────────────────────────────────── + + +class GenericSource: + def __init__(self, app_entry: AppEntry): + _validate_generic_config(app_entry) + self.app_entry = app_entry + self.start_url = app_entry.config["url"] + self.asset_regex = re.compile(app_entry.config["asset_pattern"]) + self.intermediate_regexes = [ + re.compile(pattern) + for pattern in app_entry.config.get("intermediate_patterns", []) + ] + self.headers = dict(app_entry.config.get("headers", {})) + + def _links(self, page_url: str) -> tuple[str, list[str]]: + body, final_url, _, _ = _generic_fetch(page_url, self.headers) + return final_url, _anchor_urls(final_url, body) + + def _choose_intermediate(self, links: list[str], regex: re.Pattern) -> str: + matches = [] + seen = set() + for link in links: + if link in seen: + continue + seen.add(link) + match = regex.fullmatch(_decoded_path_component(link)) + if match: + matches.append((link, match)) + if not matches: + raise ValueError( + f"no intermediate link matches pattern for {self.app_entry.name}" + ) + if len(matches) == 1: + return matches[0][0] + group_count = len(matches[0][1].groups()) + numeric = [] + if group_count: + try: + for link, match in matches: + if len(match.groups()) != group_count: + raise ValueError + numeric.append((tuple(int(value) for value in match.groups()), link)) + except (TypeError, ValueError): + numeric = [] + if not numeric: + raise ValueError("multiple intermediate links match; make the pattern more precise") + greatest = max(item[0] for item in numeric) + winners = [link for values, link in numeric if values == greatest] + if len(winners) != 1: + raise ValueError("multiple intermediate links have the same numeric version") + return winners[0] + + def get_latest_release(self) -> ReleaseInfo: + page_url = self.start_url + for regex in self.intermediate_regexes: + page_url, links = self._links(page_url) + page_url = self._choose_intermediate(links, regex) + + assets = [] + page_url, links = self._links(page_url) + for link in links: + filename = _decoded_path_component(link) + if filename.endswith("/"): + continue + if not filename.lower().endswith(".apk"): + continue + if not self.asset_regex.fullmatch(filename): + continue + if re.search(r"(?i)(debug|test|unsigned)", filename): + continue + assets.append((filename, link)) + if not assets: + raise ValueError(f"no APK asset matches for {self.app_entry.name}") + if len(assets) > 1: + raise ValueError("multiple APK assets match; set asset_pattern more precisely") + filename, download_url = assets[0] + return ReleaseInfo( + app_entry=self.app_entry, + version=filename, + download_url=download_url, + expected_sha256=self.app_entry.config.get("sha256"), + ) + + def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path: + if release.expected_sha256: + cached = cache.find_cached(release.app_entry.app_id, release.expected_sha256) + if cached: + return cached + staging_dir = cache.root / release.app_entry.app_id + staging_dir.mkdir(parents=True, exist_ok=True) + metadata_path = staging_dir / "generic-cache.json" + metadata = {} + if not release.expected_sha256 and metadata_path.is_file(): + try: + metadata = json.loads(metadata_path.read_text()) + except (OSError, ValueError): + metadata = {} + cached = None + if not release.expected_sha256 and metadata.get("url") == release.download_url: + candidate = cache.cached_path(release.app_entry.app_id, metadata.get("sha256", "")) + if candidate.is_file() and _sha256_file(candidate) == metadata.get("sha256"): + cached = candidate + conditional = {} if cached else None + if cached and metadata.get("ETag"): + conditional["If-None-Match"] = metadata["ETag"] + if cached and metadata.get("Last-Modified"): + conditional["If-Modified-Since"] = metadata["Last-Modified"] + staging = None + try: + downloaded, _, response_headers, status = _atomic_download( + release.download_url, + staging_dir / "generic-download.apk", + self.headers, + conditional_headers=conditional, + same_origin=True, + return_metadata=True, + ) + if status == 304 and cached: + return cached + if status == 304 or downloaded is None: + raise ValueError("received HTTP 304 without a reusable cache entry") + staging = downloaded + actual = _sha256_file(staging) + if release.expected_sha256 and actual.lower() != release.expected_sha256.lower(): + raise ValueError(f"SHA-256 mismatch for {release.app_entry.name}") + result = cache.store(release.app_entry.app_id, staging, actual) + if not release.expected_sha256: + new_metadata = { + "url": release.download_url, + "sha256": actual, + "ETag": response_headers.get("ETag", ""), + "Last-Modified": response_headers.get("Last-Modified", ""), + } + tmp_metadata = metadata_path.with_suffix(".tmp") + tmp_metadata.write_text(json.dumps(new_metadata)) + tmp_metadata.replace(metadata_path) + return result + finally: + if staging is not None: + staging.unlink(missing_ok=True) + + # ── download cache ──────────────────────────────────────────────────────────── @@ -612,14 +923,15 @@ SOURCE_REGISTRY = { "github": GitHubSource, "gitlab": GitLabSource, "codeberg": CodebergSource, + "generic": GenericSource, } -def _make_source(app: AppEntry) -> FdroidSource | GitHubSource | GitLabSource | CodebergSource: +def _make_source(app: AppEntry) -> FdroidSource | GitHubSource | GitLabSource | CodebergSource | GenericSource: cls = SOURCE_REGISTRY.get(app.source) if cls is None: raise ValueError( - f"unknown source type '{app.source}'. Use fdroid, github, gitlab, or codeberg" + f"unknown source type '{app.source}'. Use fdroid, github, gitlab, codeberg, or generic" ) return cls(app) @@ -732,7 +1044,7 @@ def sideload( def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="sideload", - description="Install free Android apps via ADB from F-Droid, GitHub or GitLab.", + description="Install free Android apps via ADB from F-Droid, GitHub, GitLab, Codeberg, or generic HTTPS pages.", ) p.add_argument( "app", nargs="?", help="app name to install (omit for interactive selection)" diff --git a/tests/test_sideload.py b/tests/test_sideload.py index 8e627a3..51c91de 100644 --- a/tests/test_sideload.py +++ b/tests/test_sideload.py @@ -9,6 +9,7 @@ import subprocess import sys import urllib.error import urllib.request +from contextlib import contextmanager from pathlib import Path from unittest import mock @@ -43,6 +44,22 @@ def _mock_urlopen_response(data: bytes): return m +def _generic_response(data: str | bytes, url: str): + response = _mock_urlopen_response(data.encode() if isinstance(data, str) else data) + response.geturl.return_value = url + return response + + +@contextmanager +def _mock_download_opener(response=None, *, error=None): + with mock.patch("sideload._generic_opener") as opener_factory: + if error is not None: + opener_factory.return_value.open.side_effect = error + else: + opener_factory.return_value.open.return_value = response + yield + + # ── fixtures ────────────────────────────────────────────────────────────────── @@ -212,6 +229,20 @@ asset_pattern = ".*\\\\.apk$" assert cfg.apps[0].app_id == "org.schabi.newpipe" assert cfg.apps[0].config["url"].startswith("https://github.com/") + def test_generic_requires_https_url_and_asset_pattern(self, tmp_path: Path): + p = _write_config( + tmp_path / "cfg.toml", + '[[apps]]\nname="X"\nsource="generic"\npackage="x"\nurl="http://example.test/"\nasset_pattern=".*\\\\.apk"', + ) + with pytest.raises(ValueError, match="absolute HTTPS"): + sideload.Config.from_file(p) + p.write_text( + '[[apps]]\nname="X"\nsource="generic"\npackage="x"\n' + 'url="https://example.test/"\nasset_pattern=".*\\\\.apk"\nsha256="not-a-sha"' + ) + with pytest.raises(ValueError, match="64 hexadecimal"): + sideload.Config.from_file(p) + # ── DownloadCache tests ─────────────────────────────────────────────────────── @@ -375,7 +406,7 @@ class TestFdroidSource: expected_sha256=wrong_sha, ) mock_resp = _mock_urlopen_response(apk_data) - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with pytest.raises(ValueError, match="SHA-256 mismatch"): source.download_apk(release, cache) @@ -435,7 +466,7 @@ class TestGitHubSource: ) apk_data = b"github-apk-content" mock_resp = _mock_urlopen_response(apk_data) - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): result = source.download_apk(release, cache) assert result.is_file() assert result.read_bytes() == apk_data @@ -581,7 +612,7 @@ class TestGitLabSource: ) apk_data = b"gitlab-apk-content" mock_resp = _mock_urlopen_response(apk_data) - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): result = source.download_apk(release, cache) assert result.is_file() assert result.read_bytes() == apk_data @@ -590,6 +621,148 @@ class TestGitLabSource: # ── ApkValidator tests ──────────────────────────────────────────────────────── +class TestGenericSource: + def _entry(self, **config): + return sideload.AppEntry( + name="Firefox", source="generic", app_id="org.mozilla.firefox", + config={ + "url": "https://downloads.example/releases/", + "intermediate_patterns": [r"(\d+)\/$", "android/$"], + "asset_pattern": r"^fenix-.*\.apk$", + **config, + }, + ) + + def test_directory_navigation_and_numeric_version_selection(self): + entry = self._entry() + pages = [ + _generic_response('110bad', entry.config["url"]), + _generic_response('android', "https://downloads.example/releases/10/"), + _generic_response('apk', "https://downloads.example/releases/10/android/"), + ] + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.side_effect = pages + release = sideload.GenericSource(entry).get_latest_release() + assert release.download_url.endswith(".apk?download=1") + assert release.version == "fenix-1.multi.android-arm64-v8a.apk" + + def test_ambiguous_intermediate_links(self): + entry = self._entry(intermediate_patterns=[r"release/$"]) + page = _generic_response('ab', entry.config["url"]) + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.return_value = page + with pytest.raises(ValueError, match="multiple intermediate"): + sideload.GenericSource(entry).get_latest_release() + + def test_excludes_debug_and_rejects_http_assets(self): + entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$") + page = _generic_response( + 'http' + 'debuggood', + entry.config["url"], + ) + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.return_value = page + release = sideload.GenericSource(entry).get_latest_release() + assert release.download_url.endswith("good.apk") + + def test_initial_http_and_network_error(self): + bad = self._entry(url="http://downloads.example/releases/") + with pytest.raises(ValueError, match="absolute HTTPS"): + sideload.GenericSource(bad) + entry = self._entry() + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.side_effect = urllib.error.URLError("offline") + with pytest.raises(RuntimeError, match="network error"): + sideload.GenericSource(entry).get_latest_release() + + def test_headers_are_sent_but_not_exposed(self, capsys): + entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$", headers={"Authorization": "secret-token"}) + response = _generic_response('good', entry.config["url"]) + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.return_value = response + sideload.GenericSource(entry).get_latest_release() + request = opener_factory.return_value.open.call_args.args[0] + assert request.get_header("Authorization") == "secret-token" + captured = capsys.readouterr() + assert "secret-token" not in captured.out + captured.err + + def test_https_redirect_is_blocked_before_following(self): + request = urllib.request.Request( + "https://downloads.example/start", headers={"Authorization": "secret-token"} + ) + with pytest.raises(ValueError, match="non-HTTPS"): + sideload._HttpsOnlyRedirectHandler().redirect_request( + request, mock.Mock(), 302, "Found", "http://evil.example/", {} + ) + + def test_cross_host_redirect_is_blocked_and_same_origin_keeps_headers(self): + request = urllib.request.Request( + "https://downloads.example/start", headers={"Authorization": "secret-token"} + ) + handler = sideload._HttpsOnlyRedirectHandler( + "https://downloads.example/start", same_origin=True + ) + with pytest.raises(ValueError, match="changed origin"): + handler.redirect_request( + request, mock.Mock(), 302, "Found", "https://cdn.example/file", {} + ) + redirected = handler.redirect_request( + request, mock.Mock(), 302, "Found", "https://downloads.example/file", {} + ) + assert redirected.get_header("Authorization") == "secret-token" + + def test_atomic_download_rejects_downgrade_before_response_body(self): + request = urllib.request.Request("https://downloads.example/file.apk") + handler = sideload._HttpsOnlyRedirectHandler( + "https://downloads.example/file.apk", same_origin=False + ) + with pytest.raises(ValueError, match="non-HTTPS"): + handler.redirect_request( + request, mock.Mock(read=mock.Mock(side_effect=AssertionError)), + 302, "Found", "http://downloads.example/file.apk", {} + ) + + def test_redirected_page_url_is_used_for_relative_links(self): + entry = self._entry(intermediate_patterns=[] , asset_pattern=r"^good\.apk$") + response = _generic_response('good', "https://cdn.example/final/") + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.return_value = response + release = sideload.GenericSource(entry).get_latest_release() + assert release.download_url == "https://cdn.example/final/good.apk" + + def test_generic_cache_uses_conditional_request_without_sha(self, tmp_path: Path): + entry = self._entry(intermediate_patterns=[] , asset_pattern=r"^good\.apk$") + source = sideload.GenericSource(entry) + release = sideload.ReleaseInfo(entry, "good.apk", "https://downloads.example/good.apk", None) + first = _generic_response(b"apk-content", release.download_url) + first.headers = {"ETag": "v1"} + first.getcode.return_value = 200 + second = _generic_response(b"", release.download_url) + second.headers = {} + second.getcode.return_value = 304 + with mock.patch("sideload._generic_opener") as opener_factory: + opener_factory.return_value.open.side_effect = [first, second] + cache = sideload.DownloadCache(tmp_path / "cache") + cached_first = source.download_apk(release, cache) + cached_second = source.download_apk(release, cache) + assert cached_second == cached_first + assert cached_second.read_bytes() == b"apk-content" + second_request = opener_factory.return_value.open.call_args_list[1].args[0] + assert second_request.get_header("If-none-match") == "v1" + + def test_generic_cache_rejects_304_without_valid_cache(self, tmp_path: Path): + entry = self._entry(intermediate_patterns=[], asset_pattern=r"^good\.apk$") + release = sideload.ReleaseInfo(entry, "good.apk", "https://downloads.example/good.apk", None) + response = _generic_response(b"", release.download_url) + response.getcode.return_value = 304 + with _mock_download_opener(response): + with pytest.raises(ValueError, match="304"): + sideload.GenericSource(entry).download_apk( + release, sideload.DownloadCache(tmp_path / "cache") + ) + + class TestApkValidator: def test_validate_success(self, apk_file: Path): validator = sideload.ApkValidator(aapt2_bin="fake-aapt2", aapt_bin="fake-aapt") @@ -747,6 +920,13 @@ class TestMakeSource: s = sideload._make_source(codeberg_entry) assert isinstance(s, sideload.CodebergSource) + def test_generic(self): + entry = sideload.AppEntry( + name="X", source="generic", app_id="x", + config={"url": "https://example.test/", "asset_pattern": r"x\.apk"}, + ) + assert isinstance(sideload._make_source(entry), sideload.GenericSource) + def test_unknown_source_raises(self): entry = sideload.AppEntry(name="X", source="unknown", app_id="x", config={}) with pytest.raises(ValueError, match="unknown source type"): @@ -783,7 +963,7 @@ class TestSideload: with mock.patch( "sideload._http_get", return_value=json.dumps(fdroid_index).encode() ): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", side_effect=mock_run): result = sideload.sideload( app_entry, config, dry_run=True, @@ -821,7 +1001,7 @@ class TestSideload: with mock.patch( "sideload._http_get", return_value=json.dumps(fdroid_index).encode() ): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", side_effect=mock_run): result = sideload.sideload( app_entry, config, dry_run=True, @@ -865,7 +1045,7 @@ class TestSideload: with mock.patch( "sideload._http_get", return_value=json.dumps(fdroid_index).encode() ): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", side_effect=mock_run): result = sideload.sideload( app_entry, config, @@ -910,7 +1090,7 @@ class TestSideload: with mock.patch( "sideload._http_get", return_value=json.dumps(fdroid_index).encode() ): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", side_effect=mock_run): result = sideload.sideload( app_entry, config, @@ -963,7 +1143,7 @@ class TestSideload: with mock.patch( "sideload._http_get", return_value=json.dumps(fdroid_index).encode() ): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", side_effect=mock_run): result = sideload.sideload( app_entry, config, @@ -1008,7 +1188,7 @@ class TestCLI: } mock_resp = _mock_urlopen_response(apk_data) with mock.patch("sideload._http_get", return_value=json.dumps(fdroid_index).encode()): - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): with mock.patch("sideload._run", return_value=subprocess.CompletedProcess( args=[], returncode=0, stdout="org.mozilla.fennec_fdroid\n", stderr="" )): @@ -1064,7 +1244,7 @@ class TestUtilities: dest = tmp_path / "sub" / "file.apk" data = b"downloaded-content" mock_resp = _mock_urlopen_response(data) - with mock.patch("urllib.request.urlopen", return_value=mock_resp): + with _mock_download_opener(mock_resp): result = sideload._atomic_download("https://example.com/x.apk", dest) assert result == dest assert dest.read_bytes() == data @@ -1072,7 +1252,7 @@ class TestUtilities: def test_atomic_download_cleans_up_on_error(self, tmp_path: Path): dest = tmp_path / "file.apk" - with mock.patch("urllib.request.urlopen", side_effect=RuntimeError("fail")): + with _mock_download_opener(error=RuntimeError("fail")): with pytest.raises(RuntimeError): sideload._atomic_download("https://example.com/x.apk", dest) assert not dest.exists()