#!/usr/bin/env python3 """Publish a signed APK to Nomid; use NOMID_API_KEY or a private key file.""" import argparse import hashlib import json import os from pathlib import Path import sys import time import urllib.error import urllib.parse import urllib.request import uuid class PublishError(Exception): """Safe to display: never includes keys, presigned URLs or server bodies.""" class NoRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never forward an API key or an APK to a redirected destination. return None class Publisher: def __init__(self, base_url, api_key): parsed = urllib.parse.urlsplit(base_url) if (parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment): raise PublishError("The API base URL must be HTTPS without credentials or a query.") self.base_url = base_url.rstrip("/") + "/api/v1/library/apps/uploads" self.api_key = api_key self.opener = urllib.request.build_opener(NoRedirects()) def api(self, method, suffix, body=None): data = None if body is None else json.dumps(body).encode("utf-8") request = urllib.request.Request( self.base_url + suffix, data=data, method=method, headers={"X-API-Key": self.api_key, "Content-Type": "application/json"}, ) try: with self.opener.open(request, timeout=90) as response: raw = response.read(2 * 1024 * 1024 + 1) except urllib.error.HTTPError as exc: raise PublishError(f"API request failed (HTTP {exc.code}).") from None except (urllib.error.URLError, TimeoutError, OSError): raise PublishError("API connection failed; resume using the same session ID.") from None if len(raw) > 2 * 1024 * 1024: raise PublishError("API response exceeded the example client's size limit.") try: result = json.loads(raw) except (ValueError, UnicodeError): raise PublishError("The API did not return valid JSON.") from None if not isinstance(result, dict): raise PublishError("The API returned an unexpected response shape.") return result def upload_part(self, session_id, part_number, data): # Re-sign on each attempt: the prior short-lived URL may have expired. for attempt in range(3): signed = self.api("POST", "/parts/sign", { "sessionId": session_id, "partNumber": part_number, }) parsed = urllib.parse.urlsplit(signed.get("url", "")) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise PublishError("The API returned an invalid upload URL.") request = urllib.request.Request( signed["url"], data=data, method="PUT", headers=signed.get("requiredHeaders") or {}, ) try: with self.opener.open(request, timeout=180) as response: etag = response.headers.get("ETag") if not etag: raise PublishError("Storage did not return the uploaded part's ETag.") return {"partNumber": part_number, "etag": etag} except (urllib.error.URLError, TimeoutError, OSError): if attempt == 2: raise PublishError("Part upload failed; resume using the same session ID.") from None time.sleep(2 ** attempt) raise PublishError("Part upload did not finish.") def wait(self, session_id, timeout): deadline = time.monotonic() + timeout while True: result = self.api("GET", f"/status/{session_id}") status = result.get("status") if status == "completed": return result if status in ("failed", "aborted", "expired"): raise PublishError(f"Publication ended with status '{status}'.") if time.monotonic() >= deadline: raise PublishError("Publication wait timed out; poll/resume the same session ID.") time.sleep(min(5, max(0, deadline - time.monotonic()))) def publish(self, apk_path, tracks, session_id=None, timeout=1800): with apk_path.open("rb") as apk: size = os.fstat(apk.fileno()).st_size if size < 1: raise PublishError("The APK is empty.") digest = hashlib.file_digest(apk, "sha256").hexdigest() apk.seek(0) if session_id is None: # A discovered transfer may belong to another publisher targeting a # different track. Only an explicit session ID resumes a publication. metadata = {"filename": apk_path.name, "size": size, "mime": "application/vnd.android.package-archive", "sha256": digest} checked = self.api("POST", "/check", metadata) if not checked.get("allowed") and not checked.get("duplicateOf"): raise PublishError("This APK is not eligible for upload (size or file type).") session_id = self.api("POST", "/start", metadata)["sessionId"] session_id = str(uuid.UUID(str(session_id))) print(json.dumps({"sessionId": session_id}), flush=True) status = self.api("GET", f"/status/{session_id}") if status.get("status") == "completed": return status if status.get("status") == "processing": return self.wait(session_id, timeout) if status.get("status") in ("aborted", "expired"): raise PublishError("The upload session cannot be resumed; start a new upload.") uploaded = self.api("GET", f"/parts/{session_id}") part_size = uploaded.get("partSize") if not isinstance(part_size, int) or not 1 <= part_size <= 100 * 1024 * 1024: raise PublishError("The API returned an invalid part size.") reusable = {part["partNumber"]: part for part in uploaded.get("parts", [])} parts = [] for number, offset in enumerate(range(0, size, part_size), 1): part = reusable.get(number) if part is None: apk.seek(offset) part = self.upload_part(session_id, number, apk.read(min(part_size, size - offset))) parts.append(part) result = self.api("POST", "/complete", { "sessionId": session_id, "parts": parts, "tracks": tracks, }) if result.get("status") == "completed": return result if result.get("status") == "failed": raise PublishError("APK publication failed.") return self.wait(session_id, timeout) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("apk", type=Path) parser.add_argument("--base-url", default="https://api.nomid.tech/emm") parser.add_argument("--api-key-file", type=Path) parser.add_argument("--track", action="append", dest="tracks") parser.add_argument("--session-id", type=uuid.UUID) parser.add_argument("--timeout", type=int, default=1800, help="Publication wait in seconds") args = parser.parse_args() try: key = (args.api_key_file.read_text().strip() if args.api_key_file else os.environ.get("NOMID_API_KEY", "").strip()) if not key: raise PublishError("Set NOMID_API_KEY through CI secrets or provide --api-key-file.") if args.timeout < 1: raise PublishError("Timeout must be positive.") tracks = args.tracks or ["prod"] if any(not track.strip() or len(track) > 10 for track in tracks): raise PublishError("Track names must contain 1–10 characters.") result = Publisher(args.base_url, key).publish( args.apk, tracks, args.session_id, args.timeout, ) # Allowlist output: never print raw responses, URLs, storage paths or credentials. app = result.get("app") or {} print(json.dumps({"status": "completed", "sessionId": result.get("sessionId"), "packageName": app.get("packageName")}), flush=True) return 0 except PublishError as exc: print(str(exc), file=sys.stderr) return 1 except (OSError, ValueError, KeyError, TypeError): print("Unable to publish: check the APK, configuration and API response contract.", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())