From 81cf7b214111410b5273609a994d09d6b762e674 Mon Sep 17 00:00:00 2001 From: Gerardo Betancourt Date: Wed, 28 Oct 2026 12:14:09 -0400 Subject: [PATCH] Verify complete signed releases and publish through canonical Gitea --- tools/README-APT.md | 35 +++++ tools/build_apt_repository.py | 68 ++++++++- tools/publish_gitea.py | 227 +++++++++++++++++++++++++++++ tools/test_build_apt_repository.py | 37 ++++- tools/test_publish_gitea.py | 72 +++++++++ 5 files changed, 435 insertions(+), 4 deletions(-) create mode 100755 tools/publish_gitea.py create mode 100644 tools/test_publish_gitea.py diff --git a/tools/README-APT.md b/tools/README-APT.md index 8fcc410..48be9d5 100644 --- a/tools/README-APT.md +++ b/tools/README-APT.md @@ -97,3 +97,38 @@ disposable signing identity in a temporary directory. It tests actual isolated APT acceptance for each suite and rejects altered packages/indexes/signatures, unexpected versions, path traversal and bundled extra primary trust keys. It never uses the official private signing key or changes system APT sources. + + +## One signing session for the release + +For a complete release, also pass --source with the reviewed +FrameBridge-VERSION-source.tar.gz and --dependency-source twice with the two +existing dependency-source archives. Their original sibling .asc signatures +are verified and copied unchanged. The generator creates a downloads staging +subdirectory with the three DEBs, source, dependency sources and SHA256SUMS. +The same input manifest covers all of these bytes. The owner-terminal sign +command signs the three DEBs, project source and checksums in the same session +as the three APT suites. Existing dependency signatures are preserved. + +Map the staging downloads subdirectory to the existing public /downloads/; +the rest of the repository goes under /debian/. Do not publish a nested second +/debian/downloads/ route as the official download location. Verify every +signature with the pinned manifest before activating either path. + +## Canonical Gitea publication + +publish_gitea.py is an administrator tool for the existing GERANAS Gitea, not a +second Git host. Its default mode checks a root-frozen bundle SHA-256, exact old +main, ancestor relationship and the new release tag. --apply requires an +exclusive journal in a root-owned private directory. Run only the reviewed +root-frozen tool and bundle using the normal authorized administrator context. + +The tool creates a narrowly scoped, uniquely named temporary token using the +installed Gitea administrator CLI. It keeps the token only in process memory, +feeds normal Git ASKPASS through a private Unix socket and performs an atomic +push without a force flag. It verifies exact refs, revokes the temporary token +through normal Basic authentication and verifies HTTP 401 afterward. The +journal contains the nonsecret token name so an interrupted transaction can be +reviewed. If any step is uncertain, inspect that transaction rather than +replaying it. No token value is placed in URLs, arguments, environment variables, +files or reports. diff --git a/tools/build_apt_repository.py b/tools/build_apt_repository.py index bb4f379..fc79756 100755 --- a/tools/build_apt_repository.py +++ b/tools/build_apt_repository.py @@ -18,6 +18,10 @@ import tempfile PACKAGE = "framebridge-original-expanded" FINGERPRINT = "2B1F01BC806C0B33CB6CEDDE52859CEBC91B8808" +DEPENDENCY_SOURCE_ASSETS = { + "FrameBridge-0.3.1-Community-Preview1-dependency-sources.tar.gz": (82606998, "64540a85bd65eac3f394d8c0a967cacbc00e112676d57d644dcd1bb691c1926a"), + "FrameBridge-0.3.1-tunnel-client-dependency-sources.tar.gz": (26941745, "6445d4f0e773aba833fe5c610fbae54102b92de2ce636438f769f21e890c2bd9"), +} PROFILES = { "debian": ("trixie", "Debian 13", "debian13-amd64-py313", "3.13"), "ubuntu": ("noble", "Ubuntu 24.04", "ubuntu24.04-amd64-py312", "3.12"), @@ -73,6 +77,52 @@ def release_bytes(suite, label, date, files): for name, data in sorted(files.items())) return ("\n".join(lines) + "\n").encode() + +def verify_detached(public_key, signature, payload): + with tempfile.TemporaryDirectory(prefix="framebridge-asset-verify-") as tmp: + tmp = Path(tmp) + ring = tmp / "release.gpg" + ring.write_bytes(run(["gpg", "--no-options", "--homedir", tmp, "--batch", "--output", "-", + "--dearmor", public_key]).stdout) + result = run(["gpgv", "--homedir", tmp, "--keyring", ring, "--status-fd", "1", signature, payload]) + require(f"[GNUPG:] VALIDSIG {FINGERPRINT} ".encode() in result.stdout, "Unexpected dependency-source signer") + +def prepare_downloads(args, inputs): + source = getattr(args, "source", None) + dependencies = getattr(args, "dependency_source", []) + if source is None: + require(not dependencies, "Dependency assets require the corresponding project-source archive") + return [] + require(source.is_file() and not source.is_symlink() and + source.name == f"FrameBridge-{args.version}-source.tar.gz", "Unexpected project-source archive") + require({p.name for p in dependencies} == set(DEPENDENCY_SOURCE_ASSETS) and + len(dependencies) == len(DEPENDENCY_SOURCE_ASSETS), "Exactly the reviewed two dependency-source archives required") + for path in dependencies: + expected_size, expected_sha = DEPENDENCY_SOURCE_ASSETS[path.name] + require(path.is_file() and not path.is_symlink(), "Unsafe dependency-source archive") + data = path.read_bytes() + require(len(data) == expected_size and digest(data) == expected_sha, "Dependency-source archive hash mismatch") + verify_detached(args.public_key, path.with_name(path.name + ".asc"), path) + directory = args.output / "downloads" + directory.mkdir() + assets = [] + def add(name, data, needs_signing): + target = directory / name + require(not target.exists(), "Duplicate release filename") + target.write_bytes(data) + assets.append({"filename": name, "size_bytes": len(data), "sha256": digest(data), + "signature": name + ".asc", "needs_signing": needs_signing}) + for _, _, _, profile, _, data in inputs: + add(f"FrameBridge-{args.version}-{profile}.deb", data, True) + add(source.name, source.read_bytes(), True) + for path in dependencies: + add(path.name, path.read_bytes(), False) + shutil.copyfile(path.with_name(path.name + ".asc"), directory / (path.name + ".asc")) + shutil.copyfile(args.public_key, directory / "framebridge-release-key.asc") + sums = "".join(item["sha256"] + " " + item["filename"] + "\n" for item in assets) + add("SHA256SUMS", sums.encode(), True) + return assets + def prepare(args): output = args.output require(not output.exists(), "Output must be a new directory; never overwrite a published repository") @@ -117,12 +167,13 @@ def prepare(args): records.append({"distribution": distro, "label": label, "suite": suite, "profile": profile, "filename": relative, "size": len(data), "sha256": digest(data)}) + downloads = prepare_downloads(args, inputs) shutil.copyfile(args.public_key, output / "framebridge-archive-keyring.asc") files = {p.relative_to(output).as_posix(): {"bytes": p.stat().st_size, "sha256": digest(p.read_bytes())} for p in sorted(output.rglob("*")) if p.is_file()} manifest = {"schema": 1, "package": PACKAGE, "version": args.version, "fingerprint": FINGERPRINT, "prepared_utc": date.isoformat(), - "artifacts": records, "files": files} + "artifacts": records, "download_assets": downloads, "files": files} dump(output / "APT-INPUT-MANIFEST.json", manifest) manifest_hash = digest((output / "APT-INPUT-MANIFEST.json").read_bytes()) print(json.dumps({"prepared": True, "signed": False, "directory": str(output), @@ -151,11 +202,19 @@ def check_inputs(output, expected): public_fingerprint(output / "framebridge-archive-keyring.asc") return manifest -def signature_paths(output): +def signature_paths(output, sign_only=False): for suite, *_ in PROFILES.values(): root = output / "dists" / suite yield root / "Release", root / "InRelease", True yield root / "Release", root / "Release.gpg", False + manifest = json.loads((output / "APT-INPUT-MANIFEST.json").read_text()) + for asset in manifest.get("download_assets", []): + if sign_only and not asset["needs_signing"]: + continue + filename = asset["filename"] + require(Path(filename).name == filename and Path(asset["signature"]).name == asset["signature"], + "Unsafe download signature path") + yield output / "downloads" / filename, output / "downloads" / asset["signature"], False def verify_signatures(output, manifest): with tempfile.TemporaryDirectory(prefix="framebridge-apt-verify-") as tmp: @@ -183,7 +242,7 @@ def sign(args): require(args.key_home.is_dir() and not args.key_home.is_symlink(), "Unsafe signing home") key_stat = args.key_home.stat() require(key_stat.st_uid == os.getuid() and key_stat.st_mode & 0o077 == 0, "Signing home must be owner-private") - targets = list(signature_paths(args.output)) + targets = list(signature_paths(args.output, sign_only=True)) require(all(not target.exists() for _, target, _ in targets), "Signatures already exist; do not overwrite") common = ["gpg", "--no-options", "--homedir", args.key_home, "--pinentry-mode", "default", "--local-user", FINGERPRINT, "--digest-algo", "SHA256"] @@ -212,6 +271,9 @@ def main(): prepare_parser.add_argument("--" + distro, type=Path, required=True) prepare_parser.add_argument("--public-key", type=Path, required=True) prepare_parser.add_argument("--epoch", type=int) + prepare_parser.add_argument("--source", type=Path, help="Reviewed corresponding project-source tar.gz") + prepare_parser.add_argument("--dependency-source", type=Path, action="append", default=[], + help="Reviewed dependency-source archive with its unchanged sibling .asc") for name in ("sign", "verify"): sub = commands.add_parser(name) sub.add_argument("--output", type=Path, required=True) diff --git a/tools/publish_gitea.py b/tools/publish_gitea.py new file mode 100755 index 0000000..eff1bba --- /dev/null +++ b/tools/publish_gitea.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +"""Publish a reviewed release by normal authenticated fast-forward Git push.""" +import argparse +import base64 +import hashlib +import json +import os +from pathlib import Path +import re +import shutil +import socket +import subprocess +import sys +import tempfile +import threading +import urllib.error +import urllib.parse +import urllib.request +import uuid + +REPOSITORY = "http://127.0.0.1:3005/gerardo/framebridge.git" +API = "http://127.0.0.1:3005/api/v1" +CONTAINER = "geranas-gitea" + +def require(value, message): + if not value: + raise ValueError(message) + +def run(args, **kwargs): + return subprocess.run([str(x) for x in args], check=True, capture_output=True, **kwargs) + +def remote_refs(): + raw = run(["git", "-c", "credential.helper=", "ls-remote", REPOSITORY]).stdout.decode() + return dict((ref, sha) for sha, ref in (line.split() for line in raw.splitlines())) + +def api_request(method, path, token): + request = urllib.request.Request(API + path, method=method, + headers={"Authorization": "Basic " + base64.b64encode(("gerardo:" + token).encode()).decode()}) + with urllib.request.urlopen(request, timeout=15) as response: + return response.status, response.read() + +def check_bundle(args, source): + data = args.bundle.read_bytes() + require(hashlib.sha256(data).hexdigest() == args.bundle_sha256, "Bundle hash mismatch") + run(["git", "clone", "--bare", "--no-hardlinks", args.bundle, source]) + run(["git", "--git-dir", source, "fsck", "--full"]) + actual = run(["git", "--git-dir", source, "rev-parse", args.commit + "^{commit}"]).stdout.decode().strip() + require(actual == args.commit, "Reviewed release commit missing") + run(["git", "--git-dir", source, "merge-base", "--is-ancestor", args.old_commit, args.commit]) + resolved = run(["git", "--git-dir", source, "rev-parse", args.tag + "^{}"]).stdout.decode().strip() + require(resolved == args.commit, "Release tag does not point to reviewed commit") + refs = remote_refs() + require(refs.get("refs/heads/main") == args.old_commit, "Canonical main changed; review before publishing") + require("refs/tags/" + args.tag not in refs, "Release tag already exists; never overwrite") + return refs + +def record(path, state): + temporary = path.with_suffix(".partial") + with open(temporary, "x", encoding="utf-8") as stream: + json.dump(state, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + temporary.replace(path) + +def askpass_server(directory, token): + address = str(directory / "credential.sock") + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(address) + os.chmod(address, 0o600) + server.listen(2) + server.settimeout(0.5) + stop = threading.Event() + def serve(): + while not stop.is_set(): + try: + connection, _ = server.accept() + except socket.timeout: + continue + with connection: + connection.settimeout(3) + request = connection.recv(32) + if request == b"username": + connection.sendall(b"gerardo\n") + elif request == b"password": + connection.sendall(token.encode() + b"\n") + thread = threading.Thread(target=serve, daemon=True) + thread.start() + client = directory / "askpass" + client.write_text( + "#!/usr/bin/python3\nimport socket,sys\n" + "p=sys.argv[1].lower()\n" + "kind=b'username' if p.startswith('username') else b'password' if p.startswith('password') else None\n" + "if kind is None: raise SystemExit(1)\n" + "s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM)\n" + f"s.connect({address!r})\ns.sendall(kind)\ns.shutdown(socket.SHUT_WR)\n" + "data=b''\n" + "while True:\n chunk=s.recv(4096)\n if not chunk: break\n data+=chunk\n" + "sys.stdout.buffer.write(data)\n") + client.chmod(0o700) + return server, stop, thread, client + +def root_directory(path): + require(path.is_absolute() and path.resolve(strict=True) == path, "Root staging path must be absolute without symlink parents") + require(path.is_dir() and path.stat().st_mode & 0o077 == 0, "Root staging directory must be private") + for current in (path, *path.parents): + metadata = current.stat() + require(metadata.st_uid == 0 and metadata.st_mode & 0o022 == 0, + "Root staging path has a non-root or shared-writable ancestor") + +def publish(args): + require(os.geteuid() == 0 and socket.gethostname().lower() == "geranas", + "Run on GERANAS using the authorized normal administrator context") + for sha in (args.bundle_sha256,): + require(re.fullmatch(r"[a-f0-9]{64}", sha), "Invalid bundle SHA-256") + for sha in (args.commit, args.old_commit): + require(re.fullmatch(r"[a-f0-9]{40}", sha), "Invalid source commit") + require(re.fullmatch(r"v[0-9]+\.[0-9]+\.[0-9]+", args.tag), "Invalid release tag") + require(args.bundle.is_file() and not args.bundle.is_symlink(), "Unsafe bundle path") + bundle_stat = args.bundle.stat() + require(bundle_stat.st_uid == 0 and bundle_stat.st_mode & 0o022 == 0 and bundle_stat.st_nlink == 1, + "Bundle must be a root-owned reviewed regular file without shared write access") + root_directory(args.bundle.parent) + os.umask(0o077) + with tempfile.TemporaryDirectory(prefix="fb-gitea-push-", dir="/run") as tmp: + directory = Path(tmp) + source = directory / "source.git" + before = check_bundle(args, source) + state = {"schema": 1, "repository": "gerardo/framebridge", "old_commit": args.old_commit, + "commit": args.commit, "tag": args.tag, "bundle_sha256": args.bundle_sha256, + "phase": "CHECKED", "push_performed": False, "token_revoked": False} + if not args.apply: + print(json.dumps(state, sort_keys=True)) + return + require(args.journal is not None and args.journal.parent.is_dir(), "Explicit existing journal directory required") + root_directory(args.journal.parent) + require(not args.journal.exists() and not args.journal.is_symlink(), "Existing transaction; never replay blindly") + # Reserve the exact new transaction before minting a credential. + descriptor = os.open(args.journal, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + os.close(descriptor) + record(args.journal, state) + token_name = "framebridge-release-" + uuid.uuid4().hex + state["temporary_token_name"] = token_name + state["phase"] = "TOKEN_CREATION_REQUESTED" + record(args.journal, state) + token = None + failure = None + server = stop = thread = None + try: + # Capture the token only in this process's memory. Never log this result. + result = run(["docker", "exec", "--user", "git", CONTAINER, "gitea", + "--config", "/etc/gitea/app.ini", "admin", "user", "generate-access-token", + "--username", "gerardo", "--token-name", token_name, "--raw", + "--scopes", "write:repository,write:user"]) + token = result.stdout.decode().strip() + require(re.fullmatch(r"[a-fA-F0-9]{40,128}", token), "Unexpected credential response; token needs owner review") + state["phase"] = "TEMPORARY_TOKEN_CREATED" + record(args.journal, state) + # The installed token-management route requires Basic authentication. + # Gitea accepts a PAT as the Basic password; prove that gate before pushing. + status, _ = api_request("GET", "/users/gerardo/tokens?limit=1", token) + require(status == 200, "Normal Basic token-management authentication failed") + state["token_management_auth_verified"] = True + record(args.journal, state) + server, stop, thread, client = askpass_server(directory, token) + env = dict(os.environ, GIT_ASKPASS=str(client), GIT_TERMINAL_PROMPT="0", LC_ALL="C") + # No force flag. The normal receive-pack checks and Gitea hooks run. + run(["git", "--git-dir", source, "-c", "credential.helper=", "push", "--atomic", + REPOSITORY, args.commit + ":refs/heads/main", "refs/tags/" + args.tag], env=env, timeout=120) + state["push_performed"] = True + state["phase"] = "PUSH_COMPLETED" + after = remote_refs() + require(after.get("refs/heads/main") == args.commit, "Main ref readback mismatch") + resolved_tag = after.get("refs/tags/" + args.tag + "^{}", after.get("refs/tags/" + args.tag)) + require(resolved_tag == args.commit, "Release tag readback mismatch") + state["refs_verified"] = True + record(args.journal, state) + except Exception as error: + # Commands never contain the credential; do not print captured HTTP/Git output. + failure = f"{type(error).__name__}: {error}" + state["failure"] = failure + finally: + if stop is not None: + stop.set() + thread.join(timeout=2) + server.close() + if token is not None: + try: + status, _ = api_request("DELETE", "/users/gerardo/tokens/" + + urllib.parse.quote(token_name, safe=""), token) + require(status == 204, "Unexpected token revocation response") + try: + api_request("GET", "/user", token) + except urllib.error.HTTPError as error: + require(error.code == 401, "Revoked credential did not return HTTP 401") + else: + raise ValueError("Credential remains accepted after revocation") + state["token_revoked"] = True + except Exception as error: + state["revocation_failure"] = f"{type(error).__name__}: {error}" + failure = failure or "Temporary token revocation requires review" + token = None + state["phase"] = ("COMPLETED" if state.get("refs_verified") and + state["token_revoked"] and failure is None else "REVIEW_REQUIRED") + record(args.journal, state) + print(json.dumps(state, sort_keys=True)) + require(failure is None, failure) + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bundle", type=Path, required=True) + parser.add_argument("--bundle-sha256", required=True) + parser.add_argument("--commit", required=True) + parser.add_argument("--old-commit", required=True) + parser.add_argument("--tag", required=True) + parser.add_argument("--journal", type=Path) + parser.add_argument("--apply", action="store_true") + args = parser.parse_args() + try: + publish(args) + except Exception as error: + print(f"FrameBridge Gitea publication: {type(error).__name__}: {error}", file=sys.stderr) + return 1 + return 0 + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/test_build_apt_repository.py b/tools/test_build_apt_repository.py index a930113..dc3a2a4 100644 --- a/tools/test_build_apt_repository.py +++ b/tools/test_build_apt_repository.py @@ -71,7 +71,7 @@ class RepositoryTests(unittest.TestCase): return hashlib.sha256(raw).hexdigest(), json.loads(raw) def sign_test_releases(self): - for source, signed, clearsign in builder.signature_paths(self.output): + for source, signed, clearsign in builder.signature_paths(self.output, sign_only=True): subprocess.run(["gpg", "--no-options", "--homedir", str(self.key_home), "--batch", "--pinentry-mode", "loopback", "--passphrase", "", "--local-user", builder.FINGERPRINT, "--digest-algo", "SHA256", "--output", str(signed), @@ -138,6 +138,41 @@ class RepositoryTests(unittest.TestCase): self.prepare(public_key=bundle) self.assertFalse(self.output.exists()) + + def test_one_release_manifest_covers_downloads_and_preserves_dependency_signatures(self): + source = self.output.parent / "FrameBridge-0.3.3-source.tar.gz" + source.write_bytes(b"disposable source-archive fixture") + original = builder.DEPENDENCY_SOURCE_ASSETS + dependencies = [] + signatures = {} + try: + builder.DEPENDENCY_SOURCE_ASSETS = {} + for index, name in enumerate(original): + path = self.output.parent / name + data = ("disposable dependency " + str(index)).encode() + path.write_bytes(data) + builder.DEPENDENCY_SOURCE_ASSETS[name] = (len(data), hashlib.sha256(data).hexdigest()) + signature = path.with_name(name + ".asc") + subprocess.run(["gpg", "--no-options", "--homedir", str(self.key_home), "--batch", + "--pinentry-mode", "loopback", "--passphrase", "", "--local-user", builder.FINGERPRINT, + "--armor", "--output", str(signature), "--detach-sign", str(path)], check=True, capture_output=True) + signatures[name] = signature.read_bytes() + dependencies.append(path) + expected, manifest = self.prepare(source=source, dependency_source=dependencies) + self.assertEqual(len(manifest["download_assets"]), 7) + for path in dependencies: + self.assertEqual((self.output / "downloads" / (path.name + ".asc")).read_bytes(), + signatures[path.name]) + self.sign_test_releases() + self.assertTrue(builder.verify_signatures(self.output, manifest)["verified"]) + self.assertEqual(builder.check_inputs(self.output, expected)["version"], "0.3.3") + source_copy = self.output / "downloads" / source.name + source_copy.write_bytes(b"altered") + with self.assertRaisesRegex(ValueError, "Input changed"): + builder.check_inputs(self.output, expected) + finally: + builder.DEPENDENCY_SOURCE_ASSETS = original + def test_changed_signed_release_fails_verification(self): _, manifest = self.prepare() self.sign_test_releases() diff --git a/tools/test_publish_gitea.py b/tools/test_publish_gitea.py new file mode 100644 index 0000000..f4e76a6 --- /dev/null +++ b/tools/test_publish_gitea.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +"""Check the memory-only credential transport without real credentials or networking.""" +import importlib.util +import base64 +from unittest import mock +import os +from pathlib import Path +import stat +import subprocess +import tempfile +import unittest + +spec = importlib.util.spec_from_file_location("publisher", Path(__file__).with_name("publish_gitea.py")) +publisher = importlib.util.module_from_spec(spec) +spec.loader.exec_module(publisher) + +class AskpassTests(unittest.TestCase): + def test_transport_keeps_disposable_token_out_of_script_and_environment(self): + with tempfile.TemporaryDirectory(prefix="framebridge-askpass-test-") as tmp: + token = "a" * 40 # Disposable test data, not a credential. + server, stop, thread, client = publisher.askpass_server(Path(tmp), token) + try: + self.assertEqual(stat.S_IMODE(client.stat().st_mode), 0o700) + self.assertEqual(stat.S_IMODE((Path(tmp) / "credential.sock").stat().st_mode), 0o600) + self.assertNotIn(token, client.read_text()) + for prompt, expected in [("Username for 'http://127.0.0.1:3005':", b"gerardo\n"), + ("Password for 'http://gerardo@127.0.0.1:3005':", (token + "\n").encode())]: + result = subprocess.run([str(client), prompt], capture_output=True, check=True, timeout=5) + self.assertEqual(result.stdout, expected) + self.assertEqual(result.stderr, b"") + finally: + stop.set() + thread.join(timeout=2) + server.close() + + def test_unknown_prompt_returns_no_token(self): + with tempfile.TemporaryDirectory(prefix="framebridge-askpass-test-") as tmp: + server, stop, thread, client = publisher.askpass_server(Path(tmp), "b" * 40) + try: + result = subprocess.run([str(client), "Unexpected prompt"], capture_output=True, timeout=5) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(result.stdout, b"") + finally: + stop.set() + thread.join(timeout=2) + server.close() + + def test_token_management_uses_normal_basic_auth_in_memory(self): + response = mock.MagicMock() + response.__enter__.return_value.status = 204 + response.__enter__.return_value.read.return_value = b"" + disposable = "c" * 40 + with mock.patch.object(publisher.urllib.request, "urlopen", return_value=response) as opener: + status, _ = publisher.api_request("DELETE", "/users/gerardo/tokens/disposable", disposable) + self.assertEqual(status, 204) + request = opener.call_args.args[0] + expected = "Basic " + base64.b64encode(("gerardo:" + disposable).encode()).decode() + self.assertEqual(request.get_header("Authorization"), expected) + self.assertNotIn(disposable, request.full_url) + + def test_incomplete_journal_is_never_overwritten(self): + with tempfile.TemporaryDirectory(prefix="framebridge-journal-test-") as tmp: + target = Path(tmp) / "transaction.json" + partial = target.with_suffix(".partial") + partial.write_text("unreviewed incomplete state") + with self.assertRaises(FileExistsError): + publisher.record(target, {"phase": "NEW"}) + self.assertFalse(target.exists()) + self.assertEqual(partial.read_text(), "unreviewed incomplete state") + +if __name__ == "__main__": + unittest.main(verbosity=2)