fdroid-repo/fdroid-sync.py

#!/usr/bin/env python3
"""Reconcile a self-hosted F-Droid repo with the latest Forgejo releases.

Nothing is built here. Each app's APK is the one its own CI built and signed,
and its listing text comes from the fastlane/ tree in the same repo at the same
tag, so the store entry and the binary can never describe different releases.

A run is a pure function of what the forge currently holds: it works out what
should be in the repo, makes it so, and rebuilds the index only if something
actually moved. That is what makes it safe to drive from a webhook, which has
no delivery guarantee and no retry — a missed delivery costs latency, not
correctness, because the next run (or the timer) converges anyway.

Usage: fdroid-sync <config.json> [app-id ...]
"""

import hashlib
import json
import os
import shutil
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

# Both halves of a release filename come from the git tag, which the app
# workflows write as v<versionName>+<versionCode> (Flutter's own pubspec
# version syntax). The code is what F-Droid orders versions by and what names
# the changelog file, so a tag without one is an error rather than a guess.
TAG_SEP = "+"

# How long to keep waiting for an APK that is still uploading. The release
# webhook fires when the release object is created, which in these workflows is
# one API call *before* the APK is attached to it, so a webhook-triggered run
# routinely arrives at a release with no asset yet. That is a race to wait out,
# not an error: a 50-80 MB upload off a CI runner takes a while.
ASSET_WAIT_SECONDS = 600
ASSET_POLL_SECONDS = 20

# A run can be triggered while an earlier one is still going, and systemd
# merges the second `start` into the first rather than queueing it, so the
# release that triggered it would be missed until the daily timer. Re-checking
# the forge after a pass that changed something closes that window; a pass that
# changes nothing ends the loop, so the steady state is still a single pass.
MAX_PASSES = 3

DEFAULT_LOCALE = "en-US"

# Per-read, not per-transfer, so this is not a cap on how long an 80 MB APK may
# take — only on how long the forge may go silent mid-stream.
HTTP_TIMEOUT = 60


def log(*args):
    print(*args, flush=True)


class AppError(Exception):
    """A failure confined to one app, so the other three still get a chance."""


# --------------------------------------------------------------------------
# forge API


def api_get(url):
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
        return json.load(resp)


def download(url, dest: Path, expect_size=None):
    """Fetch to a sibling temp file, then rename.

    Never streams straight onto the destination: a half-written file under
    repo/ is one `fdroid update` away from being published as a real release,
    and a phone that downloads it gets a signature failure it cannot explain.
    """
    tmp = dest.with_name(dest.name + ".part")
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp, open(tmp, "wb") as out:
        shutil.copyfileobj(resp, out, 1 << 20)
    size = tmp.stat().st_size
    if expect_size is not None and size != expect_size:
        tmp.unlink()
        raise AppError(f"{dest.name}: got {size} bytes, forge advertised {expect_size}")
    tmp.rename(dest)
    return size


def latest_release(forge, repo):
    """The newest published, non-draft, non-prerelease release."""
    url = f"{forge}/api/v1/repos/{repo}/releases/latest"
    try:
        return api_get(url)
    except urllib.error.HTTPError as e:
        if e.code == 404:
            raise AppError("no published release yet") from None
        raise


def version_code(tag):
    name, sep, code = tag.lstrip("v").partition(TAG_SEP)
    if not sep or not code.isdigit():
        raise AppError(
            f"tag {tag!r} is not v<versionName>{TAG_SEP}<versionCode>; "
            "cannot tell which version this is"
        )
    return name, int(code)


def apk_asset(release):
    apks = [a for a in release.get("assets", []) if a["name"].endswith(".apk")]
    if len(apks) > 1:
        raise AppError(
            f"release {release['tag_name']} has {len(apks)} APKs "
            f"({', '.join(a['name'] for a in apks)}); expected exactly one"
        )
    return apks[0] if apks else None


def await_apk_asset(forge, repo, release):
    """Poll until the APK shows up, or give up loudly."""
    asset = apk_asset(release)
    if asset:
        return asset
    deadline = time.monotonic() + ASSET_WAIT_SECONDS
    log(f"    {release['tag_name']}: no APK attached yet, waiting for the upload")
    while time.monotonic() < deadline:
        time.sleep(ASSET_POLL_SECONDS)
        asset = apk_asset(latest_release(forge, repo))
        if asset:
            return asset
    raise AppError(
        f"release {release['tag_name']} still has no APK after "
        f"{ASSET_WAIT_SECONDS}s; did the build fail after creating the release?"
    )


# --------------------------------------------------------------------------
# listing metadata, straight out of the app's own fastlane tree


def tree_manifest(root: Path):
    """(relpath, sha256) for every file under root, sorted.

    Used to decide whether a freshly fetched listing differs from the one
    already in place. The alternative — always replacing it — would mark every
    run as changed and rebuild the index daily for nothing, which rewrites
    every index timestamp and makes every client re-download it.
    """
    out = []
    for path in sorted(root.rglob("*")):
        if path.is_file():
            out.append((str(path.relative_to(root)), hashlib.sha256(path.read_bytes()).hexdigest()))
    return out


def extract_fastlane(forge, repo, tag, fastlane_dir, dest: Path):
    """Unpack <fastlane_dir>/metadata/android/ from the source tag into dest.

    Copied verbatim, subdirectories and all: fdroidserver reads the fastlane
    layout natively — title.txt, short_description.txt, full_description.txt,
    changelogs/<versionCode>.txt and images/{icon,featureGraphic}.png plus
    images/*Screenshots/ — so there is nothing to translate and no mapping
    table to keep in step with upstream.
    """
    url = f"{forge}/{repo}/archive/{urllib.parse.quote(tag, safe='')}.tar.gz"
    prefix = f"{fastlane_dir.strip('/')}/metadata/android/"
    with tempfile.TemporaryDirectory() as td:
        tarball = Path(td) / "src.tar.gz"
        download(url, tarball)
        found = 0
        with tarfile.open(tarball) as tf:
            for member in tf:
                if not member.isfile():
                    continue
                # Drop the archive's top-level directory, which Forgejo names
                # after the repo rather than after anything we control.
                _, _, rel = member.name.partition("/")
                if not rel.startswith(prefix):
                    continue
                rel = rel[len(prefix):]
                target = (dest / rel).resolve()
                # A tarball is attacker-controlled input in principle, and this
                # one is unpacked as root. Refuse any member that resolves
                # outside dest rather than trusting the forge to never serve a
                # `../` path.
                if not str(target).startswith(str(dest.resolve()) + os.sep):
                    raise AppError(f"archive member escapes the destination: {member.name}")
                target.parent.mkdir(parents=True, exist_ok=True)
                src = tf.extractfile(member)
                if src is None:
                    continue
                with open(target, "wb") as out:
                    shutil.copyfileobj(src, out)
                found += 1
    if not found:
        raise AppError(f"{fastlane_dir}/metadata/android/ is empty or missing at tag {tag}")
    return found


def open_apk(apk: Path):
    """androguard's APK reader, or None if androguard is not installed."""
    try:
        from androguard.core.apk import APK
    except ImportError:
        return None
    # androguard logs through loguru, not the stdlib, and at DEBUG it prints a
    # line per resource-table oddity — hundreds of them per Flutter APK, which
    # buries the one line we actually emit here.
    try:
        from loguru import logger

        logger.remove()
    except ImportError:
        pass
    return APK(str(apk))


def check_version_code(apk: Path, code):
    """Fail if the tag and the APK disagree about which version this is.

    Everything downstream keys off the version code taken from the git tag:
    it names the APK in repo/, it is what pruning sorts on, and it is what
    resolves the changelog. F-Droid, meanwhile, reads the code out of the APK's
    own manifest. Normally both come from the same pubspec line and agree, but
    nothing enforces that — a hand-moved tag is enough to split them, and then
    the repo publishes a file whose name claims one version while the index
    claims another, and pruning starts ordering by a number that means nothing.
    """
    pkg = open_apk(apk)
    if pkg is None:
        return
    actual = int(pkg.get_androidversion_code())
    if actual != code:
        raise AppError(
            f"{apk.name}: tag says versionCode {code} but the APK's manifest "
            f"says {actual}; the tag and the build do not match"
        )


def extract_apk_icon(apk: Path, dest: Path):
    """Last resort: pull the launcher icon out of the APK itself.

    An app whose fastlane tree carries no images/icon.png would otherwise be
    listed with no icon at all, because fdroidserver does not fall back to the
    APK. Two of these four apps are in exactly that state, and an iconless row
    in the client reads as a broken repo rather than as missing metadata.

    Shipping an icon in fastlane is still the better answer — it can be the
    square store icon rather than whatever density the launcher happens to
    use — so this only fills the gap, it does not replace it.
    """
    pkg = open_apk(apk)
    if pkg is None:
        log("    androguard unavailable; cannot recover an icon from the APK")
        return False
    # Descending, because the default pick is whatever the manifest points at,
    # and on every modern Android app that is an adaptive-icon XML rather than
    # a bitmap. Asking for a bounded density returns a real PNG.
    for dpi in (640, 480, 320, 240, 160):
        name = pkg.get_app_icon(max_dpi=dpi)
        if name and name.lower().endswith(".png"):
            data = pkg.get_file(name)
            dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_bytes(data)
            log(f"    recovered icon from the APK ({name}, <={dpi}dpi, {len(data)} bytes)")
            return True
    log("    no bitmap launcher icon in the APK; listing will have no icon")
    return False


_listing_done = {}


def sync_listing(app, cfg, tag, apk: Path, locale):
    """Refresh metadata/<packageId>/ from the app repo. True if it changed.

    Read from the source archive at the *released tag*, not from the default
    branch, so the listing always describes the build it ships with. The
    consequence is worth stating plainly: correcting a typo in
    full_description.txt on main changes nothing here. It takes a new release,
    or moving the tag.

    Refetched on every run even when the release has not moved, so a tag that
    was moved in place is still picked up, and so a listing lost to a failed
    run repairs itself. Within a single run it is done once per tag: the later
    passes exist to notice a *new* release, and repeating the fetch and the APK
    icon parse for a tag already handled would be pure waste.
    """
    if _listing_done.get(app["id"]) == tag:
        return False
    root = Path(cfg["root"])
    live = root / "metadata" / app["packageId"]
    with tempfile.TemporaryDirectory(dir=root / "tmp") as td:
        staged = Path(td) / "metadata"
        staged.mkdir()
        extract_fastlane(cfg["forge"], app["repo"], tag, app.get("fastlane", "fastlane"), staged)
        icon = staged / locale / "images" / "icon.png"
        if not icon.exists():
            extract_apk_icon(apk, icon)
        _listing_done[app["id"]] = tag
        if live.exists() and tree_manifest(live) == tree_manifest(staged):
            return False
        # Replace rather than merge, so a file deleted upstream disappears here
        # too. The swap is two renames in the same filesystem; `fdroid update`
        # is not running concurrently (the unit is a oneshot) so there is no
        # reader to tear.
        if live.exists():
            shutil.rmtree(live.with_name(live.name + ".old"), ignore_errors=True)
            live.rename(live.with_name(live.name + ".old"))
        staged.rename(live)
        shutil.rmtree(live.with_name(live.name + ".old"), ignore_errors=True)
        return True


def app_metadata(app, code, version_name):
    """The non-localized half of the listing: metadata/<packageId>.yml.

    Name, Summary and Description are deliberately not set. fdroidserver reads
    those from the localized files this script just unpacked, and a value here
    would win over them — pinning the listing to whatever was true when this
    file was written.

    Two fields here are load-bearing for the changelog, and both have to be
    right or the client shows an empty "What's New":

    CurrentVersionCode is what fdroidserver resolves changelogs/<code>.txt
    against, so without it the changelog file is never read at all.

    Builds is what gets the text into the index once read. index-v2 only
    carries whatsNew for a version matching the *last* Builds entry, and a repo
    of prebuilt APKs has no build recipes, so the entry has to be declared. It
    is metadata only — nothing here builds anything, `fdroid update` just
    matches versionCode against it.
    """
    forge_repo = app["forgeUrl"]
    lines = [
        "# Generated by fdroid-sync. Edit apps.nix, not this file.",
        f"AuthorName: {app.get('authorName', 'randogoth')}",
        f"License: {app['license']}",
        "Categories:",
    ]
    lines += [f"  - {c}" for c in app["categories"]]
    lines += [
        f"SourceCode: {forge_repo}",
        f"IssueTracker: {forge_repo}/issues",
        f"Changelog: {forge_repo}/releases",
    ]
    if app.get("webSite"):
        lines.append(f"WebSite: {app['webSite']}")
    lines += [
        f"CurrentVersion: {version_name}",
        f"CurrentVersionCode: {code}",
        "Builds:",
        f"  - versionName: {version_name}",
        f"    versionCode: {code}",
        "",
    ]
    return "\n".join(lines)


# --------------------------------------------------------------------------
# per-app reconcile


def prune_apks(root: Path, package_id, keep):
    """Drop all but the newest `keep` versions of one app.

    F-Droid's own answer is archive_older, which moves old APKs into archive/
    and publishes a second index for them. That is a feature for a repo with
    users who pin old versions; here it would be a second index to sign and
    serve so that nobody can reach four-releases-ago builds of a launcher.
    Deleting is the honest version of the same decision, and because the index
    is regenerated from whatever APKs are present, it needs no bookkeeping.
    """
    found = []
    for apk in (root / "repo").glob(f"{package_id}_*.apk"):
        code = apk.stem[len(package_id) + 1:]
        if code.isdigit():
            found.append((int(code), apk))
    dropped = False
    for _code, apk in sorted(found, reverse=True)[keep:]:
        log(f"    pruning {apk.name} (keeping newest {keep})")
        apk.unlink()
        dropped = True
    return dropped


def reconcile(app, cfg):
    """Bring one app's APK and listing in line with its latest release."""
    root = Path(cfg["root"])
    locale = cfg.get("locale", DEFAULT_LOCALE)
    release = latest_release(cfg["forge"], app["repo"])
    tag = release["tag_name"]
    name, code = version_code(tag)
    apk_path = root / "repo" / f"{app['packageId']}_{code}.apk"

    changed = False
    if apk_path.exists():
        log(f"  {app['id']}: {tag} already published")
    else:
        asset = await_apk_asset(cfg["forge"], app["repo"], release)
        log(f"  {app['id']}: fetching {tag} ({asset['name']}, {asset['size']} bytes)")
        download(asset["browser_download_url"], apk_path, expect_size=asset["size"])
        try:
            check_version_code(apk_path, code)
        except AppError:
            # Unlink first: left in place it would be picked up by the next
            # run as "already published" and never checked again.
            apk_path.unlink()
            raise
        changed = True

    if sync_listing(app, cfg, tag, apk_path, locale):
        log(f"  {app['id']}: listing updated from {tag}")
        changed = True

    meta = root / "metadata" / f"{app['packageId']}.yml"
    want = app_metadata(app, code, name)
    if not meta.exists() or meta.read_text() != want:
        meta.write_text(want)
        log(f"  {app['id']}: metadata set to version {name} ({code})")
        changed = True

    if prune_apks(root, app["packageId"], cfg.get("keep", 3)):
        changed = True
    return changed


# --------------------------------------------------------------------------


def build_index(cfg):
    """Sign a fresh index over whatever is now in repo/.

    Deliberately without --create-metadata: an APK with no metadata/<id>.yml
    means apps.nix and the repo disagree, and the useful outcome is a loud
    failure, not a listing auto-filled with the working directory's name as its
    category (which is what the skeleton generator actually does).
    """
    log("building and signing the index")
    env = dict(os.environ)
    env["FDROID_KEYSTORE_PASS"] = Path(cfg["keystorePassFile"]).read_text().strip()
    # fdroidserver reads keypass separately even when it is the same secret.
    env["FDROID_KEY_PASS"] = env["FDROID_KEYSTORE_PASS"]
    subprocess.run([cfg["fdroid"], "update"], cwd=cfg["root"], env=env, check=True)


def main(argv):
    if len(argv) < 2:
        print(__doc__, file=sys.stderr)
        return 2
    cfg = json.loads(Path(argv[1]).read_text())
    root = Path(cfg["root"])
    for sub in ("repo", "metadata", "tmp"):
        (root / sub).mkdir(parents=True, exist_ok=True)

    wanted = set(argv[2:])
    apps = [a for a in cfg["apps"] if not wanted or a["id"] in wanted]
    unknown = wanted - {a["id"] for a in cfg["apps"]}
    if unknown:
        print(f"unknown app(s): {', '.join(sorted(unknown))}", file=sys.stderr)
        return 2
    for app in apps:
        app["forgeUrl"] = f"{cfg['forge']}/{app['repo']}"

    # An index that was never built is a change in itself, otherwise a first
    # run on a fresh server that finds every APK already in place (restored
    # from a backup, say) would serve no index at all.
    changed = not (root / "repo" / "index-v2.json").exists()
    failed = []
    for attempt in range(1, MAX_PASSES + 1):
        log(f"checking {len(apps)} app(s) against {cfg['forge']} (pass {attempt})")
        moved = False
        failed = []
        for app in apps:
            try:
                if reconcile(app, cfg):
                    moved = True
            except AppError as e:
                log(f"  {app['id']}: {e}")
                failed.append(app["id"])
            # Broad on purpose: one app's surprise must not sink the other three.
            except Exception as e:
                log(f"  {app['id']}: unexpected {type(e).__name__}: {e}")
                failed.append(app["id"])
        changed |= moved
        if not moved:
            break

    if changed:
        build_index(cfg)
    else:
        log("nothing changed; index left alone")

    if failed:
        # Loud, but only after everything that could be published has been.
        log(f"FAILED: {', '.join(failed)}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))