Add generic APK URL source
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
326
src/sideload.py
326
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
|
||||
@@ -174,6 +208,283 @@ def _http_get(url: str, extra_headers: dict | None = None) -> bytes:
|
||||
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)"
|
||||
|
||||
@@ -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('<a href="1/">1</a><a href="10/">10</a><a href="http://bad/">bad</a>', entry.config["url"]),
|
||||
_generic_response('<a href="android/">android</a>', "https://downloads.example/releases/10/"),
|
||||
_generic_response('<a href="fenix-1.multi.android-arm64-v8a.apk?download=1">apk</a>', "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('<a href="a/release/">a</a><a href="b/release/">b</a>', 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(
|
||||
'<a href="http://downloads.example/good.apk">http</a>'
|
||||
'<a href="good-debug.apk">debug</a><a href="good.apk">good</a>',
|
||||
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('<a href="good.apk">good</a>', 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('<a href="good.apk">good</a>', "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()
|
||||
|
||||
Reference in New Issue
Block a user