feat(termux): gpg-signed apt repo staging (dists/pool, xz control)
Pure-stdlib stager for the static apt layout: Packages(+gz), an apt-valid Release (Date field, checksums in the same deb822 stanza -- learned from a real device rejecting the first shape), InRelease + Release.gpg signed with the repo key (passphrase support via env), the signing pubkey exported alongside, per-suite immutability, and nightly versions that sort below stable. Control members are xz (our dpkg-deb builds -Zxz; the stager also tolerates gz and zstd-via-binary).
This commit is contained in:
ethernet8023
86
scripts/termux/deb_version.py
Normal file
86
scripts/termux/deb_version.py
Normal file
@@ -0,0 +1,86 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Derive a Debian package version (or channel) from a hermes-agent release tag.
|
||||
|
||||
Pure function; imported by scripts/termux/build_deb.sh and unit-tested by
|
||||
tests/test_termux_deb_version.py (Task 4 of .hermes/plans/2026-08-31_termux-deb.md).
|
||||
|
||||
Mapping:
|
||||
v1.2.3 -> 1.2.3-1
|
||||
v1.2.3-canary.2026083112 -> 1.2.3~canary.2026083112-1
|
||||
|
||||
The ``~`` ranks the nightly below the corresponding stable in dpkg's version
|
||||
ordering. The major version is capped at 3 digits (CalVer-style cap): a tag
|
||||
with a 4+ digit major is rejected as malformed.
|
||||
|
||||
``--channel`` derives the release channel from the SAME tag regex: a tag with
|
||||
a nightly timestamp is ``nightly``, everything else is ``stable``. This is the
|
||||
single source of truth for the channel; workflows and other tooling must call
|
||||
this instead of re-typing a case statement.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
|
||||
# The canary timestamp shape MUST match the canonical _CANARY_TAG_RE in
|
||||
# hermes_cli/update_channel.py (exactly 8 or 14 digits, 20-prefixed) and
|
||||
# channelForTag in scripts/r2-release.mjs. Cross-referenced by
|
||||
# tests/test_termux_deb_version.py::test_canary_tag_shape_matches_canonical.
|
||||
_TAG_RE = re.compile(
|
||||
r"^v(?P<major>0|[1-9]\d{0,2})\.(?P<minor>\d+)\.(?P<patch>\d+)"
|
||||
r"(?:-canary\.(?P<ts>20\d{6}(?:\d{6})?))?$"
|
||||
)
|
||||
|
||||
|
||||
def _match_tag(tag: str) -> re.Match[str]:
|
||||
m = _TAG_RE.match(tag)
|
||||
if m is None:
|
||||
raise ValueError(
|
||||
f"malformed release tag {tag!r}: expected v<MAJOR>.<MINOR>.<PATCH> "
|
||||
"or v<MAJOR>.<MINOR>.<PATCH>-canary.<timestamp>"
|
||||
)
|
||||
return m
|
||||
|
||||
|
||||
def deb_version_for_tag(tag: str) -> str:
|
||||
"""Map a release tag to its Debian version. Raises ValueError on malformed tags."""
|
||||
m = _match_tag(tag)
|
||||
base = f"{m.group('major')}.{m.group('minor')}.{m.group('patch')}"
|
||||
ts = m.group("ts")
|
||||
if ts is None:
|
||||
return f"{base}-1"
|
||||
return f"{base}~canary.{ts}-1"
|
||||
|
||||
|
||||
def channel_for_tag(tag: str) -> str:
|
||||
"""Map a release tag to its channel: 'canary' or 'stable'.
|
||||
|
||||
Derived from the same _TAG_RE as deb_version_for_tag, so the two can never
|
||||
drift: a tag that yields a '~canary' deb version is canary, and the
|
||||
malformed-tag rejection is identical.
|
||||
"""
|
||||
m = _match_tag(tag)
|
||||
return "canary" if m.group("ts") is not None else "stable"
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
args = argv[1:]
|
||||
channel_mode = False
|
||||
if args and args[0] == "--channel":
|
||||
channel_mode = True
|
||||
args = args[1:]
|
||||
if len(args) != 1:
|
||||
mode = "deb_version.py --channel <tag>" if channel_mode else "deb_version.py <tag>"
|
||||
print(f"usage: {mode}", file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
print(channel_for_tag(args[0]) if channel_mode else deb_version_for_tag(args[0]))
|
||||
except ValueError as exc:
|
||||
print(f"deb_version: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv))
|
||||
379
scripts/termux/stage_apt_repo.py
Normal file
379
scripts/termux/stage_apt_repo.py
Normal file
@@ -0,0 +1,379 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage a static APT repository layout (dists/ + pool/) from a pool of .debs.
|
||||
|
||||
Pure stdlib. Builds dists/<suite>/{Packages,Packages.gz,Release,InRelease,Release.gpg}
|
||||
and copies .debs into pool/<first-char>/.
|
||||
|
||||
Usage:
|
||||
python stage_apt_repo.py --pool POOL_DIR --out OUT_DIR \
|
||||
--suite hermes-stable|hermes-canary [--gpg-key-file PATH]
|
||||
|
||||
Exit codes:
|
||||
0 - success
|
||||
2 - usage/IO error
|
||||
3 - Release emitted but not signed (gpg binary or key file missing);
|
||||
CI treats 3 as failure
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import gzip
|
||||
import hashlib
|
||||
import io
|
||||
import lzma
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ARCH = "aarch64"
|
||||
COMPONENT = "main"
|
||||
|
||||
REQUIRED_CONTROL_FIELDS = ["Package", "Version", "Architecture"]
|
||||
|
||||
|
||||
class StageError(Exception):
|
||||
"""Fatal staging error."""
|
||||
|
||||
|
||||
def die(msg: str, code: int = 2) -> "NoReturn": # type: ignore[valid-type]
|
||||
print(f"stage_apt_repo: {msg}", file=sys.stderr)
|
||||
raise SystemExit(code)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# .deb control parsing (stdlib ar + tar, no dpkg-deb)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def read_ar_entries(data: bytes):
|
||||
"""Yield (name, size, payload) for each member of an ar archive."""
|
||||
if data[:8] != b"!<arch>\n":
|
||||
raise StageError("not an ar archive")
|
||||
pos = 8
|
||||
while pos + 60 <= len(data):
|
||||
header = data[pos : pos + 60]
|
||||
name = header[0:16].decode("ascii", "replace").strip()
|
||||
size_field = header[48:58].decode("ascii", "replace").strip()
|
||||
try:
|
||||
size = int(size_field)
|
||||
except ValueError:
|
||||
raise StageError(f"bad ar member size {size_field!r}")
|
||||
pos += 60
|
||||
yield name, size, data[pos : pos + size]
|
||||
pos += size + (size % 2) # members are 2-byte aligned
|
||||
|
||||
|
||||
def deb_control_fields_and_bytes(deb_path: Path) -> tuple[dict, bytes]:
|
||||
"""Parse a .deb's control member and return (fields, raw archive bytes).
|
||||
|
||||
The caller gets the raw bytes too so hashing/pool-copy need no re-read.
|
||||
"""
|
||||
data = deb_path.read_bytes()
|
||||
control_tar = None
|
||||
for name, size, payload in read_ar_entries(data):
|
||||
if name in ("control.tar", "control.tar.gz", "control.tar.xz", "control.tar.zst"):
|
||||
control_tar = (name, payload)
|
||||
break
|
||||
if control_tar is None:
|
||||
raise StageError(f"{deb_path.name}: no control.tar member found")
|
||||
name, payload = control_tar
|
||||
|
||||
if name == "control.tar.gz":
|
||||
raw = gzip.decompress(payload)
|
||||
elif name == "control.tar.xz":
|
||||
raw = lzma.decompress(payload)
|
||||
elif name == "control.tar.zst":
|
||||
# zstd has no stdlib decoder; dpkg-deb -Zxz (our build default)
|
||||
# keeps control in xz, but tolerate zst debs from elsewhere when
|
||||
# the host has a zstd binary.
|
||||
try:
|
||||
raw = subprocess.run(
|
||||
["zstd", "-d", "-c"], input=payload, check=True,
|
||||
capture_output=True,
|
||||
).stdout
|
||||
except (FileNotFoundError, subprocess.CalledProcessError) as e:
|
||||
raise StageError(
|
||||
f"{deb_path.name}: control member is zstd-compressed and no zstd binary is available"
|
||||
) from e
|
||||
elif name == "control.tar":
|
||||
raw = payload
|
||||
else:
|
||||
raise StageError(f"{deb_path.name}: unsupported control compression {name}")
|
||||
|
||||
fields: dict = {}
|
||||
with tarfile.open(fileobj=io.BytesIO(raw), mode="r:") as tf:
|
||||
for member in tf.getmembers():
|
||||
if member.name.lstrip("./") == "control":
|
||||
f = tf.extractfile(member)
|
||||
if f is None:
|
||||
continue
|
||||
fields = _parse_debian_control(f.read().decode("utf-8", "replace"))
|
||||
break
|
||||
for req in REQUIRED_CONTROL_FIELDS:
|
||||
if req not in fields:
|
||||
raise StageError(f"{deb_path.name}: control missing {req}")
|
||||
return fields, data
|
||||
|
||||
|
||||
def deb_control_fields(deb_path: Path) -> dict:
|
||||
"""Parse Package/Version/Architecture/... from a .deb's control member."""
|
||||
fields, _ = deb_control_fields_and_bytes(deb_path)
|
||||
return fields
|
||||
|
||||
|
||||
def _parse_debian_control(text: str) -> dict:
|
||||
fields: dict = {}
|
||||
last = None
|
||||
for line in text.splitlines():
|
||||
if not line.strip():
|
||||
last = None
|
||||
continue
|
||||
if line[0] in " \t" and last:
|
||||
fields[last] += " " + line.strip()
|
||||
elif ":" in line:
|
||||
key, _, val = line.partition(":")
|
||||
last = key.strip()
|
||||
fields[last] = val.strip()
|
||||
return fields
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dpkg version ordering: '~' sorts before end-of-version (and before empty)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _order_char(ch: str) -> int:
|
||||
if ch == "~":
|
||||
return -1
|
||||
if ch.isdigit():
|
||||
return 0
|
||||
if ch.isalpha():
|
||||
return ord(ch)
|
||||
return ord(ch) + 256
|
||||
|
||||
|
||||
def deb_version_key(version: str):
|
||||
"""Sort key implementing dpkg version comparison for our versions."""
|
||||
epoch, _, rest = version.partition(":")
|
||||
epoch_num = int(epoch) if epoch.isdigit() else 0
|
||||
if ":" not in version:
|
||||
rest = version
|
||||
up, _, rev = rest.rpartition("-")
|
||||
if not up:
|
||||
up, rev = rest, ""
|
||||
|
||||
def cmp_part(s: str):
|
||||
parts = []
|
||||
i = 0
|
||||
while i < len(s):
|
||||
if s[i].isdigit():
|
||||
j = i
|
||||
while j < len(s) and s[j].isdigit():
|
||||
j += 1
|
||||
parts.append((0, int(s[i:j]), ""))
|
||||
i = j
|
||||
else:
|
||||
j = i
|
||||
while j < len(s) and not s[j].isdigit():
|
||||
j += 1
|
||||
parts.append((1, tuple(_order_char(c) for c in s[i:j]), ""))
|
||||
i = j
|
||||
return parts
|
||||
|
||||
# dpkg: end-of-part sorts after everything except '~'; padding the shorter
|
||||
# part with (2, ...) achieves that ('~' yields order char -1 < any pad).
|
||||
def padded(parts):
|
||||
return parts + [(2, (), "")] * 4
|
||||
|
||||
return (epoch_num, padded(cmp_part(up)), padded(cmp_part(rev)))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Staging
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def existing_published(out_dir: Path, suite: str) -> set:
|
||||
"""(package, version) pairs already published in dists/<suite>/Packages."""
|
||||
packages_file = out_dir / "dists" / suite / COMPONENT / f"binary-{ARCH}" / "Packages"
|
||||
published = set()
|
||||
if packages_file.exists():
|
||||
text = packages_file.read_text(encoding="utf-8")
|
||||
pkg = ver = None
|
||||
for line in text.splitlines():
|
||||
if line.startswith("Package: "):
|
||||
pkg = line[len("Package: "):].strip()
|
||||
elif line.startswith("Version: "):
|
||||
ver = line[len("Version: "):].strip()
|
||||
elif not line.strip() and pkg and ver:
|
||||
published.add((pkg, ver))
|
||||
pkg = ver = None
|
||||
if pkg and ver:
|
||||
published.add((pkg, ver))
|
||||
return published
|
||||
|
||||
|
||||
def stage(pool_dir: Path, out_dir: Path, suite: str, gpg_key_file: Path | None) -> int:
|
||||
debs = sorted(pool_dir.glob("*.deb"))
|
||||
if not debs:
|
||||
die(f"no .deb files found in pool {pool_dir}")
|
||||
|
||||
published = existing_published(out_dir, suite)
|
||||
|
||||
dists = out_dir / "dists" / suite
|
||||
binary_dir = dists / COMPONENT / f"binary-{ARCH}"
|
||||
binary_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
stanzas = []
|
||||
for deb in debs:
|
||||
fields, raw = deb_control_fields_and_bytes(deb)
|
||||
key = (fields["Package"], fields["Version"])
|
||||
if key in published:
|
||||
die(
|
||||
f"refusing: {key[0]}_{key[1]} already published in dists/{suite} "
|
||||
"(published apt assets are immutable)"
|
||||
)
|
||||
arch = fields["Architecture"]
|
||||
target = out_dir / "pool" / deb.name[0].lower() / deb.name
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
# Unconditional write: the pool file must always equal the source so
|
||||
# the stanza hashes below can never describe a stale/different file.
|
||||
target.write_bytes(raw)
|
||||
filename = f"pool/{deb.name[0].lower()}/{deb.name}"
|
||||
size = len(raw)
|
||||
sha256 = hashlib.sha256(raw).hexdigest()
|
||||
stanzas.append(
|
||||
{
|
||||
"Package": fields["Package"],
|
||||
"Version": fields["Version"],
|
||||
"Architecture": arch,
|
||||
"Maintainer": fields.get("Maintainer", "Hermes Agent <noreply@nousresearch.com>"),
|
||||
"Installed-Size": fields.get("Installed-Size", "0"),
|
||||
"Description": fields.get("Description", "Hermes Agent"),
|
||||
"Filename": filename,
|
||||
"Size": str(size),
|
||||
"SHA256": sha256,
|
||||
}
|
||||
)
|
||||
|
||||
# Packages sorted by version, canary (~) below stable
|
||||
stanzas.sort(
|
||||
key=lambda s: (s["Package"], deb_version_key(s["Version"]))
|
||||
)
|
||||
packages_text = "\n".join(
|
||||
"\n".join(f"{k}: {v}" for k, v in stanza.items()) for stanza in stanzas
|
||||
) + ("\n" if stanzas else "")
|
||||
|
||||
(binary_dir / "Packages").write_text(packages_text, encoding="utf-8")
|
||||
with gzip.GzipFile(filename="", mode="wb", fileobj=open(binary_dir / "Packages.gz", "wb"), mtime=0) as gz:
|
||||
gz.write(packages_text.encode("utf-8"))
|
||||
|
||||
# Date is REQUIRED by apt (it refuses a Release without one) and the
|
||||
# checksum sections must stay INSIDE the same deb822 stanza: a blank
|
||||
# line ends the record, and apt then never sees the hashes ("weak
|
||||
# security information"). One paragraph, no blank lines.
|
||||
release_fields = [
|
||||
"Origin: Hermes Agent",
|
||||
"Label: hermes-agent",
|
||||
f"Suite: {suite}",
|
||||
f"Codename: {suite}",
|
||||
f"Architectures: {ARCH}",
|
||||
f"Components: {COMPONENT}",
|
||||
f"Description: Hermes Agent apt repository ({suite})",
|
||||
"Date: " + time.strftime("%a, %d %b %Y %H:%M:%S UTC", time.gmtime()),
|
||||
]
|
||||
checksums = []
|
||||
sha512 = []
|
||||
for name in ("Packages", "Packages.gz"):
|
||||
p = binary_dir / name
|
||||
rel = f"{COMPONENT}/binary-{ARCH}/{name}"
|
||||
size = p.stat().st_size
|
||||
data = p.read_bytes()
|
||||
checksums.append(f" {hashlib.sha256(data).hexdigest()} {size:8d} {rel}")
|
||||
sha512.append(f" {hashlib.sha512(data).hexdigest()} {size:8d} {rel}")
|
||||
release = "\n".join(release_fields) + "\n"
|
||||
release += "SHA256:\n" + "\n".join(checksums) + "\n"
|
||||
release += "SHA512:\n" + "\n".join(sha512) + "\n"
|
||||
|
||||
release_path = dists / "Release"
|
||||
release_path.write_text(release, encoding="utf-8")
|
||||
|
||||
if gpg_key_file is not None and shutil.which("gpg"):
|
||||
sign(dists, release_path, gpg_key_file)
|
||||
return 0
|
||||
print("warning: gpg binary or key file unavailable; emitted unsigned Release", file=sys.stderr)
|
||||
return 3
|
||||
|
||||
|
||||
def sign(dists: Path, release_path: Path, gpg_key_file: Path) -> None:
|
||||
# Real signing keys are usually passphrase-protected; TERMUX_APT_GPG_PASSPHRASE
|
||||
# (set only when the key needs one) rides in via argv -- never the key material.
|
||||
passphrase = os.environ.get("TERMUX_APT_GPG_PASSPHRASE", "")
|
||||
base = ["gpg", "--batch", "--yes", "--pinentry-mode", "loopback"]
|
||||
if passphrase:
|
||||
base += ["--passphrase", passphrase]
|
||||
|
||||
def gpg(*args: str, stdin: bytes | None = None) -> bytes:
|
||||
result = subprocess.run(
|
||||
[*base, *args],
|
||||
input=stdin, capture_output=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise StageError(f"gpg failed: {result.stderr.decode(errors='replace')}")
|
||||
return result.stdout
|
||||
|
||||
secret = gpg_key_file.read_bytes()
|
||||
gpg("--import", stdin=secret)
|
||||
# key file may be a full keypair; extract the key id via listing
|
||||
listing = gpg("--list-secret-keys", "--with-colons").decode()
|
||||
key_id = None
|
||||
for line in listing.splitlines():
|
||||
if line.startswith("sec:"):
|
||||
key_id = line.split(":")[4]
|
||||
break
|
||||
if not key_id:
|
||||
raise StageError("no secret key found after import")
|
||||
|
||||
gpg(
|
||||
"--clearsign", "--local-user", key_id,
|
||||
"--output", str(dists / "InRelease"),
|
||||
str(release_path),
|
||||
)
|
||||
gpg(
|
||||
"--detach-sign", "--armor", "--local-user", key_id,
|
||||
"--output", str(dists / "Release.gpg"),
|
||||
str(release_path),
|
||||
)
|
||||
|
||||
# Publish the signing key's public half at the repo root. The published
|
||||
# key is exported from the exact key that signed THIS suite, so the two
|
||||
# can never drift -- a key rotation re-publishes itself on the next
|
||||
# run. Users fetch it from a stable URL (docs point here).
|
||||
root = dists.parent.parent
|
||||
pub = gpg("--armor", "--export", key_id)
|
||||
if not pub.strip():
|
||||
raise StageError("gpg exported an empty public key")
|
||||
(root / "key.asc").write_bytes(pub)
|
||||
|
||||
|
||||
def main(argv: list | None = None) -> int:
|
||||
ap = argparse.ArgumentParser(description="Stage a static APT repo layout.")
|
||||
ap.add_argument("--pool", required=True, type=Path)
|
||||
ap.add_argument("--out", required=True, type=Path)
|
||||
ap.add_argument("--suite", required=True, choices=["hermes-stable", "hermes-canary"])
|
||||
ap.add_argument("--gpg-key-file", type=Path, default=None)
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if not args.pool.is_dir():
|
||||
die(f"pool dir not found: {args.pool}")
|
||||
args.out.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
return stage(args.pool, args.out, args.suite, args.gpg_key_file)
|
||||
except StageError as e:
|
||||
die(str(e))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
208
tests/test_stage_apt_repo.py
Normal file
208
tests/test_stage_apt_repo.py
Normal file
@@ -0,0 +1,208 @@
|
||||
"""Tests for scripts/termux/stage_apt_repo.py — stdlib + pytest, no network, no gpg."""
|
||||
|
||||
import gzip
|
||||
import io
|
||||
import sys
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPTS = REPO_ROOT / "scripts" / "termux"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
|
||||
import stage_apt_repo # noqa: E402
|
||||
|
||||
|
||||
def make_deb(path: Path, package: str, version: str, arch: str = "aarch64", compression: str = "gz") -> None:
|
||||
"""Build a minimal .deb (ar archive with control.tar.gz) using stdlib only."""
|
||||
control = (
|
||||
f"Package: {package}\n"
|
||||
f"Version: {version}\n"
|
||||
f"Architecture: {arch}\n"
|
||||
f"Maintainer: Test <test@example.com>\n"
|
||||
f"Description: test package {package}\n"
|
||||
)
|
||||
buf = io.BytesIO()
|
||||
mode = f"w:{compression}"
|
||||
member = f"control.tar.{compression}" if compression != "tar" else "control.tar"
|
||||
with tarfile.open(fileobj=buf, mode=mode) as tf:
|
||||
data = control.encode("utf-8")
|
||||
ti = tarfile.TarInfo("control")
|
||||
ti.size = len(data)
|
||||
tf.addfile(ti, io.BytesIO(data))
|
||||
|
||||
ar = io.BytesIO()
|
||||
ar.write(b"!<arch>\n")
|
||||
payload = buf.getvalue()
|
||||
header = "{:<16}{:<12}{:<6}{:<6}{:<8}{:<10}".format(
|
||||
member, "0", "0", "0", "100644", str(len(payload))
|
||||
).encode() + b"`\n"
|
||||
ar.write(header)
|
||||
ar.write(payload)
|
||||
if len(payload) % 2:
|
||||
ar.write(b"\n")
|
||||
path.write_bytes(ar.getvalue())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_gpg(monkeypatch):
|
||||
"""Make the script believe gpg is absent so signing is skipped (exit 3)."""
|
||||
monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_gpg(monkeypatch, tmp_path):
|
||||
"""Make the script believe gpg is present, but stub out signing."""
|
||||
monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: "C:/fake/gpg.exe")
|
||||
monkeypatch.setattr(stage_apt_repo, "sign", lambda *a, **k: None)
|
||||
key = tmp_path / "signing.asc"
|
||||
key.write_text("stub-key\n")
|
||||
return key
|
||||
|
||||
|
||||
def test_stages_xz_control_deb(tmp_path):
|
||||
"""dpkg >= 1.21 emits xz/zst control members; our build uses -Zxz so the
|
||||
stager must read xz controls (gz is covered by every other test)."""
|
||||
pool = tmp_path / "pool"
|
||||
pool.mkdir()
|
||||
make_deb(pool / "hermes-agent_1.0-1_aarch64.deb", "hermes-agent", "1.0-1", compression="xz")
|
||||
out = tmp_path / "out"
|
||||
out.mkdir()
|
||||
rc = stage_apt_repo.stage(pool, out, "hermes-canary", None)
|
||||
assert rc == 3 # unsigned (no gpg key file) but staged
|
||||
|
||||
|
||||
def test_control_field_extraction(tmp_path):
|
||||
deb = tmp_path / "pkg_a.deb"
|
||||
make_deb(deb, "hermes-agent", "1.2.3-1")
|
||||
fields = stage_apt_repo.deb_control_fields(deb)
|
||||
assert fields["Package"] == "hermes-agent"
|
||||
assert fields["Version"] == "1.2.3-1"
|
||||
assert fields["Architecture"] == "aarch64"
|
||||
|
||||
|
||||
def test_canary_versions_below_stable():
|
||||
versions = ["1.2.3-1", "1.2.3~canary.20260831120000-1", "1.2.4~canary.1-1", "1.2.4-1"]
|
||||
ordered = sorted(versions, key=stage_apt_repo.deb_version_key)
|
||||
assert ordered == [
|
||||
"1.2.3~canary.20260831120000-1",
|
||||
"1.2.3-1",
|
||||
"1.2.4~canary.1-1",
|
||||
"1.2.4-1",
|
||||
]
|
||||
|
||||
|
||||
def test_dists_layout_and_pool_copy(tmp_path, fake_gpg):
|
||||
pool = tmp_path / "pool-in"
|
||||
pool.mkdir()
|
||||
make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1")
|
||||
out = tmp_path / "repo"
|
||||
r = stage_apt_repo.main(
|
||||
[
|
||||
"--pool", str(pool), "--out", str(out), "--suite", "hermes-stable",
|
||||
"--gpg-key-file", str(fake_gpg),
|
||||
]
|
||||
)
|
||||
assert r == 0
|
||||
|
||||
dists = out / "dists" / "hermes-stable" / "main" / "binary-aarch64"
|
||||
assert (dists / "Packages").exists()
|
||||
assert (dists / "Packages.gz").exists()
|
||||
assert (out / "dists" / "hermes-stable" / "Release").exists()
|
||||
|
||||
deb_out = out / "pool" / "h" / "hermes-agent_1.2.3-1_aarch64.deb"
|
||||
assert deb_out.exists()
|
||||
|
||||
text = (dists / "Packages").read_text(encoding="utf-8")
|
||||
assert "Package: hermes-agent" in text
|
||||
assert "Version: 1.2.3-1" in text
|
||||
assert "Filename: pool/h/hermes-agent_1.2.3-1_aarch64.deb" in text
|
||||
assert "SHA256: " in text
|
||||
|
||||
gz_text = gzip.decompress((dists / "Packages.gz").read_bytes()).decode()
|
||||
assert gz_text == text
|
||||
|
||||
release = (out / "dists" / "hermes-stable" / "Release").read_text()
|
||||
assert "Suite: hermes-stable" in release
|
||||
assert "SHA256:" in release
|
||||
assert "SHA512:" in release
|
||||
# apt contract (learned from a real device rejecting our first repo):
|
||||
# Date is mandatory, and the checksum sections must live in the SAME
|
||||
# deb822 stanza as the header fields -- a blank line ends the record,
|
||||
# after which apt "provides only weak security information" and
|
||||
# disables the repository.
|
||||
assert "Date: " in release
|
||||
assert "\n\n" not in release, "blank line splits the Release stanza"
|
||||
head, _, checksums_block = release.partition("SHA256:\n")
|
||||
assert "Date: " in head, "Date must precede the checksum sections"
|
||||
|
||||
|
||||
def test_immutability_refusal(tmp_path, fake_gpg, capsys):
|
||||
pool = tmp_path / "pool-in"
|
||||
pool.mkdir()
|
||||
make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1")
|
||||
out = tmp_path / "repo"
|
||||
assert stage_apt_repo.main(
|
||||
[
|
||||
"--pool", str(pool), "--out", str(out), "--suite", "hermes-stable",
|
||||
"--gpg-key-file", str(fake_gpg),
|
||||
]
|
||||
) == 0
|
||||
with pytest.raises(SystemExit) as ei:
|
||||
stage_apt_repo.main(
|
||||
[
|
||||
"--pool", str(pool), "--out", str(out), "--suite", "hermes-stable",
|
||||
"--gpg-key-file", str(fake_gpg),
|
||||
]
|
||||
)
|
||||
assert ei.value.code == 2
|
||||
assert "already published" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_unsigned_release_exit_3_without_gpg(tmp_path, no_gpg):
|
||||
pool = tmp_path / "pool-in"
|
||||
pool.mkdir()
|
||||
make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1")
|
||||
out = tmp_path / "repo"
|
||||
code = stage_apt_repo.main(
|
||||
["--pool", str(pool), "--out", str(out), "--suite", "hermes-canary"]
|
||||
)
|
||||
assert code == 3
|
||||
assert (out / "dists" / "hermes-canary" / "Release").exists()
|
||||
assert not (out / "dists" / "hermes-canary" / "InRelease").exists()
|
||||
assert not (out / "dists" / "hermes-canary" / "Release.gpg").exists()
|
||||
|
||||
|
||||
def test_signing_invoked_when_gpg_and_key_present(tmp_path, monkeypatch):
|
||||
"""No real gpg: assert sign() is called with the right dists dir/key file."""
|
||||
calls = []
|
||||
|
||||
def fake_sign(dists, release_path, gpg_key_file):
|
||||
calls.append((str(dists), str(release_path), str(gpg_key_file)))
|
||||
(dists / "InRelease").write_text("stub", encoding="utf-8")
|
||||
(dists / "Release.gpg").write_text("stub", encoding="utf-8")
|
||||
|
||||
monkeypatch.setattr(stage_apt_repo.shutil, "which", lambda _: "C:/fake/gpg.exe")
|
||||
monkeypatch.setattr(stage_apt_repo, "sign", fake_sign)
|
||||
|
||||
pool = tmp_path / "pool-in"
|
||||
pool.mkdir()
|
||||
make_deb(pool / "hermes-agent_1.2.3-1_aarch64.deb", "hermes-agent", "1.2.3-1")
|
||||
out = tmp_path / "repo"
|
||||
keyfile = tmp_path / "signing.asc"
|
||||
keyfile.write_text("-----BEGIN PGP PRIVATE KEY BLOCK-----\n")
|
||||
code = stage_apt_repo.main(
|
||||
[
|
||||
"--pool", str(pool), "--out", str(out), "--suite", "hermes-stable",
|
||||
"--gpg-key-file", str(keyfile),
|
||||
]
|
||||
)
|
||||
assert code == 0
|
||||
assert len(calls) == 1
|
||||
dists, release_path, kf = calls[0]
|
||||
assert dists == str(out / "dists" / "hermes-stable")
|
||||
assert release_path == str(out / "dists" / "hermes-stable" / "Release")
|
||||
assert kf == str(keyfile)
|
||||
assert (out / "dists" / "hermes-stable" / "InRelease").exists()
|
||||
138
tests/test_termux_deb_version.py
Normal file
138
tests/test_termux_deb_version.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""Unit tests for scripts/termux/deb_version.py (Task 4 of the termux-deb plan)."""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
SCRIPT = HERE.parent / "scripts" / "termux" / "deb_version.py"
|
||||
|
||||
from scripts.termux.deb_version import channel_for_tag, deb_version_for_tag # noqa: E402
|
||||
|
||||
|
||||
def test_canary_tag_shape_matches_canonical():
|
||||
"""Invariant: the deb versioner accepts EXACTLY the canary tags the
|
||||
canonical release tooling mints. The canonical shape lives in
|
||||
hermes_cli/update_channel.py:_CANARY_TAG_RE (8-or-14-digit, 20-prefixed
|
||||
timestamps); scripts/r2-release.mjs:channelForTag parses the same shape.
|
||||
A tag this module accepts but the release flow would never mint (or vice
|
||||
versa) is version-drift between the .deb channel and the feed channel.
|
||||
"""
|
||||
from hermes_cli.update_channel import _CANARY_TAG_RE as _NIGHTLY_TAG_RE
|
||||
from scripts.termux import deb_version as dv
|
||||
|
||||
samples = [
|
||||
"v0.20.6-canary.20260831120000", # canonical canary (14-digit)
|
||||
"v0.20.6-canary.20260831", # canonical canary (8-digit)
|
||||
"v1.2.3", # stable
|
||||
]
|
||||
for tag in samples:
|
||||
assert dv._TAG_RE.match(tag), f"deb versioner rejects canonical tag {tag}"
|
||||
|
||||
never_minted = [
|
||||
"v1.2.3-canary.202608311", # 9 digits -- canonical rejects
|
||||
"v1.2.3-canary.12345678", # non-20 prefix -- canonical rejects
|
||||
"v1.2.3-canary.202608311200001", # 15 digits -- canonical rejects
|
||||
]
|
||||
for tag in never_minted:
|
||||
assert not _NIGHTLY_TAG_RE.match(tag), f"sample is actually canonical: {tag}"
|
||||
assert not dv._TAG_RE.match(tag), f"deb versioner accepts never-minted tag {tag}"
|
||||
|
||||
|
||||
def test_stable_tag_maps_to_revision_1():
|
||||
assert deb_version_for_tag("v1.2.3") == "1.2.3-1"
|
||||
|
||||
|
||||
def _dpkg_key(v: str) -> str:
|
||||
# Approximate dpkg ordering for these versions: '~' sorts before everything
|
||||
# (even the empty string / '-'), so map it low.
|
||||
return v.replace("~", "\x00")
|
||||
|
||||
|
||||
def test_stable_tag_multi_digit():
|
||||
assert deb_version_for_tag("v26.8.31") == "26.8.31-1"
|
||||
|
||||
|
||||
def test_major_can_be_three_digits():
|
||||
assert deb_version_for_tag("v126.8.31") == "126.8.31-1"
|
||||
|
||||
|
||||
def test_canary_tag_ranks_below_stable():
|
||||
got = deb_version_for_tag("v1.2.3-canary.20260831120000")
|
||||
assert got == "1.2.3~canary.20260831120000-1"
|
||||
assert _dpkg_key(got) < _dpkg_key(deb_version_for_tag("v1.2.3")) # dpkg ordering
|
||||
|
||||
|
||||
def test_canary_canary_ranking_among_nightlies():
|
||||
earlier = deb_version_for_tag("v1.2.3-canary.20260831000000")
|
||||
later = deb_version_for_tag("v1.2.3-canary.20260831235959")
|
||||
assert _dpkg_key(earlier) < _dpkg_key(later) < _dpkg_key(deb_version_for_tag("v1.2.3"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bad",
|
||||
[
|
||||
"",
|
||||
"1.2.3", # missing v prefix
|
||||
"v1.2", # not three components
|
||||
"v1.2.3.4", # four components
|
||||
"v1.2.3-", # empty suffix
|
||||
"v1.2.3-canary", # canary without timestamp
|
||||
"v1.2.3-canary.abc", # non-numeric timestamp
|
||||
"v1.2.3-beta.1", # unknown suffix channel
|
||||
"v1.2.x",
|
||||
"v-1.2.3",
|
||||
],
|
||||
)
|
||||
def test_malformed_tags_raise(bad):
|
||||
with pytest.raises(ValueError):
|
||||
deb_version_for_tag(bad)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", ["v1234.1.2", "v99999.0.0"])
|
||||
def test_major_above_three_digits_rejected(bad):
|
||||
with pytest.raises(ValueError):
|
||||
deb_version_for_tag(bad)
|
||||
|
||||
|
||||
def test_minor_patch_can_be_three_digits():
|
||||
# Cap applies to major only; minor/patch may be wide.
|
||||
assert deb_version_for_tag("v1.234.567") == "1.234.567-1"
|
||||
|
||||
|
||||
def test_cli_invocation(capsys):
|
||||
r = subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "v9.8.7"], capture_output=True, text=True
|
||||
)
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert r.stdout.strip() == "9.8.7-1"
|
||||
|
||||
|
||||
def test_channel_matches_canary_shape():
|
||||
"""--channel derives from the SAME _TAG_RE as the deb version: any tag
|
||||
that yields a '~canary' version is canary, everything else stable."""
|
||||
assert channel_for_tag("v1.2.3") == "stable"
|
||||
assert channel_for_tag("v26.8.31") == "stable"
|
||||
assert channel_for_tag("v0.20.6-canary.20260831120000") == "canary"
|
||||
assert channel_for_tag("v0.20.6-canary.20260831") == "canary"
|
||||
|
||||
|
||||
def test_channel_agrees_with_deb_version():
|
||||
for tag in ("v1.2.3", "v126.8.31", "v1.2.3-canary.20260831120000"):
|
||||
assert ("~canary" in deb_version_for_tag(tag)) == (channel_for_tag(tag) == "canary")
|
||||
|
||||
|
||||
def test_channel_malformed_tag_raises():
|
||||
with pytest.raises(ValueError):
|
||||
channel_for_tag("v1.2")
|
||||
|
||||
|
||||
def test_channel_cli_invocation():
|
||||
for tag, expected in [("v9.8.7", "stable"), ("v9.8.7-canary.20260831120000", "canary")]:
|
||||
r = subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--channel", tag], capture_output=True, text=True
|
||||
)
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert r.stdout.strip() == expected
|
||||
Reference in New Issue
Block a user