import base64 import json import os import pathlib import plistlib import re import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request import zipfile ROOT = "https://api.appstoreconnect.apple.com/v1/" POLL = 30 PROCESSING_LIMIT = 5400 RETRY_AFTER = 5 NOTES_LIMIT = 4000 EDITABLE = "PREPARE_FOR_SUBMISSION,DEVELOPER_REJECTED,REJECTED,METADATA_REJECTED,INVALID_BINARY" RELEASES = {"after-approval": "AFTER_APPROVAL", "manual": "MANUAL"} def fail(message): print(f"::error::{message}", flush=True) sys.exit(1) def setting(name, default=""): return os.environ.get(name, "").strip() or default def b64url(data): return base64.urlsafe_b64encode(data).rstrip(b"=").decode() def explain(body): text = body.decode(errors="replace") try: errors = json.loads(text)["errors"] return "; ".join(f"{error.get('title', '')}: {error.get('detail', '')}".strip(": ") for error in errors) except (ValueError, KeyError, TypeError, AttributeError): return text[:500] def send(method, url, body=None, headers=None): where = f"{method} {urllib.parse.urlsplit(url).path}" for attempt in (1, 2, 3): try: call = urllib.request.Request(url, data=body, method=method, headers=headers or {}) with urllib.request.urlopen(call, timeout=900) as response: return response.read() except urllib.error.HTTPError as error: if attempt < 3 and error.code in (429, 500, 502, 503, 504): time.sleep(RETRY_AFTER * attempt) continue fail(f"{where} answered {error.code}: {explain(error.read())}") except (urllib.error.URLError, OSError) as error: if attempt < 3: time.sleep(RETRY_AFTER * attempt) continue fail(f"{where} failed: {error}") def es256(key, data): with tempfile.TemporaryDirectory() as folder: path = pathlib.Path(folder, "key.p8") path.touch(0o600) path.write_text(key) result = subprocess.run(["openssl", "dgst", "-sha256", "-sign", str(path)], input=data, capture_output=True) if result.returncode != 0: fail("openssl could not sign with the API key: " + result.stderr.decode(errors="replace").strip()) der = result.stdout offset = 2 halves = [] while offset < len(der): length = der[offset + 1] halves.append(int.from_bytes(der[offset + 2:offset + 2 + length], "big").to_bytes(32, "big")) offset += 2 + length return b"".join(halves) class Connect: def __init__(self, issuer, key_id, key): self.issuer = issuer self.key_id = key_id self.key = key self.minted = 0 self.token = "" def authorization(self): now = int(time.time()) if now - self.minted > 900: header = b64url(json.dumps({"alg": "ES256", "kid": self.key_id, "typ": "JWT"}).encode()) claims = b64url(json.dumps({"iss": self.issuer, "iat": now, "exp": now + 1140, "aud": "appstoreconnect-v1"}).encode()) unsigned = f"{header}.{claims}" self.token = f"{unsigned}.{b64url(es256(self.key, unsigned.encode()))}" self.minted = now return {"Authorization": f"Bearer {self.token}"} def call(self, method, path, payload=None, query=None): headers = self.authorization() body = None if payload is not None: headers["Content-Type"] = "application/json" body = json.dumps(payload).encode() url = f"{ROOT}{path}" + (f"?{urllib.parse.urlencode(query)}" if query else "") answer = send(method, url, body, headers) return json.loads(answer) if answer else {} def ref(kind, identifier): return {"data": {"type": kind, "id": identifier}} def notes(folder): if not folder: return [] texts = [] for file in sorted(pathlib.Path(folder).glob("*.txt")): text = file.read_text(encoding="utf-8").strip() if not text: continue if len(text) > NOTES_LIMIT: fail(f"{file} has {len(text)} characters; App Store Connect takes at most {NOTES_LIMIT} per language") texts.append((file.stem, text)) if not texts: fail(f"{folder} holds no .txt file with text in it") return texts def bundle_info(ipa, bundle_id): with zipfile.ZipFile(ipa) as archive: plists = [name for name in archive.namelist() if re.fullmatch(r"Payload/[^/]+\.app/Info\.plist", name)] if len(plists) != 1: fail(f"{ipa} does not look like an iOS app: no single Payload/.app/Info.plist in it") info = plistlib.loads(archive.read(plists[0])) if info.get("CFBundleIdentifier") != bundle_id: fail(f"{ipa} is {info.get('CFBundleIdentifier')}, not {bundle_id}") if not info.get("CFBundleShortVersionString") or not info.get("CFBundleVersion"): fail(f"{ipa} carries no CFBundleShortVersionString or CFBundleVersion") if "ITSAppUsesNonExemptEncryption" not in info: fail("the app's Info.plist has no ITSAppUsesNonExemptEncryption key, so every build would wait for an export compliance answer in App Store Connect; add the key to the app's Info.plist") return info["CFBundleShortVersionString"], info["CFBundleVersion"] def app_id(connect, bundle_id): apps = connect.call("GET", "apps", query={"filter[bundleId]": bundle_id, "limit": "200"})["data"] for app in apps: if app["attributes"]["bundleId"] == bundle_id: return app["id"] fail(f"App Store Connect has no app {bundle_id}; create its record there first") def upload(connect, app, ipa, version, build): created = connect.call("POST", "buildUploads", {"data": { "type": "buildUploads", "attributes": {"cfBundleShortVersionString": version, "cfBundleVersion": build, "platform": "IOS"}, "relationships": {"app": ref("apps", app)}, }})["data"]["id"] reserved = connect.call("POST", "buildUploadFiles", {"data": { "type": "buildUploadFiles", "attributes": {"assetType": "ASSET", "fileName": ipa.name, "fileSize": ipa.stat().st_size, "uti": "com.apple.ipa"}, "relationships": {"buildUpload": ref("buildUploads", created)}, }})["data"] operations = reserved["attributes"].get("uploadOperations") or [] if not operations: fail("App Store Connect gave no upload operations for the file") with ipa.open("rb") as content: for operation in operations: content.seek(operation["offset"]) part = content.read(operation["length"]) headers = {header["name"]: header["value"] for header in operation.get("requestHeaders") or []} send(operation.get("method") or "PUT", operation["url"], part, headers) connect.call("PATCH", f"buildUploadFiles/{reserved['id']}", {"data": { "type": "buildUploadFiles", "id": reserved["id"], "attributes": {"uploaded": True}, }}) print(f"{ipa.name} uploaded in {len(operations)} parts; App Store Connect is processing it") return created def processed(connect, upload_id): deadline = time.time() + PROCESSING_LIMIT build = None seen = None while time.time() < deadline: if not build: upload_state = connect.call("GET", f"buildUploads/{upload_id}", query={"include": "build"})["data"] state = upload_state["attributes"].get("state") or {} if state.get("state") == "FAILED": problems = "; ".join(f"{error.get('code')}: {error.get('description')}" for error in state.get("errors") or []) fail(f"App Store Connect refused the build: {problems}") build = ((upload_state.get("relationships") or {}).get("build") or {}).get("data") current = state.get("state") if build: current = connect.call("GET", f"builds/{build['id']}")["data"]["attributes"]["processingState"] if current == "VALID": print("the build is processed") return build["id"] if current in ("FAILED", "INVALID"): fail(f"App Store Connect could not process the build: {current}") if current != seen: print(f"waiting: {current}", flush=True) seen = current time.sleep(POLL) fail(f"the build was not processed within {PROCESSING_LIMIT // 60} minutes; it may still appear in App Store Connect") def existing_build(connect, app, version, build): query = { "filter[app]": app, "filter[preReleaseVersion.version]": version, "filter[preReleaseVersion.platform]": "IOS", "filter[processingState]": "VALID", "sort": "-uploadedDate", "limit": "1", } if build: query["filter[version]"] = build builds = connect.call("GET", "builds", query=query)["data"] if not builds: fail(f"App Store Connect has no processed build of version {version}" + (f" ({build})" if build else "")) return builds[0]["id"] def tester_notes(connect, build, texts): current = connect.call("GET", f"builds/{build}/betaBuildLocalizations", query={"limit": "50"})["data"] known = {localization["attributes"]["locale"]: localization["id"] for localization in current} for locale, text in texts: if locale in known: connect.call("PATCH", f"betaBuildLocalizations/{known[locale]}", {"data": { "type": "betaBuildLocalizations", "id": known[locale], "attributes": {"whatsNew": text}, }}) else: connect.call("POST", "betaBuildLocalizations", {"data": { "type": "betaBuildLocalizations", "attributes": {"locale": locale, "whatsNew": text}, "relationships": {"build": ref("builds", build)}, }}) print(f"what to test: {', '.join(locale for locale, _ in texts)}") def beta_groups(connect, app, names): groups = {group["attributes"]["name"]: group for group in connect.call("GET", f"apps/{app}/betaGroups", query={"limit": "200"})["data"]} missing = [name for name in names if name not in groups] if missing: fail(f"no TestFlight group named {', '.join(missing)}; the app has {', '.join(groups) or 'none'}") return [groups[name] for name in names] def testers(connect, build, groups): external = False for group in groups: attributes = group["attributes"] if attributes.get("isInternalGroup") and attributes.get("hasAccessToAllBuilds"): print(f"{attributes['name']} already gets every build") continue connect.call("POST", f"betaGroups/{group['id']}/relationships/builds", {"data": [{"type": "builds", "id": build}]}) external = external or not attributes.get("isInternalGroup") print(f"added to {attributes['name']}") if external: connect.call("POST", "betaAppReviewSubmissions", {"data": { "type": "betaAppReviewSubmissions", "relationships": {"build": ref("builds", build)}, }}) print("sent to TestFlight review, which external testers wait for") def submit(connect, app, build, version, texts, release, phased): attributes = {"versionString": version, "releaseType": release} editable = connect.call("GET", f"apps/{app}/appStoreVersions", query={ "filter[platform]": "IOS", "filter[appVersionState]": EDITABLE, "limit": "1", })["data"] if editable: version_id = editable[0]["id"] connect.call("PATCH", f"appStoreVersions/{version_id}", {"data": { "type": "appStoreVersions", "id": version_id, "attributes": attributes, "relationships": {"build": ref("builds", build)}, }}) else: version_id = connect.call("POST", "appStoreVersions", {"data": { "type": "appStoreVersions", "attributes": dict(attributes, platform="IOS"), "relationships": {"app": ref("apps", app), "build": ref("builds", build)}, }})["data"]["id"] if texts: earlier = connect.call("GET", f"apps/{app}/appStoreVersions", query={"filter[platform]": "IOS", "limit": "2"})["data"] if len(earlier) < 2: print("::notice::the first version of an app takes no release notes; they were left out") else: current = connect.call("GET", f"appStoreVersions/{version_id}/appStoreVersionLocalizations", query={"limit": "50"})["data"] known = {localization["attributes"]["locale"]: localization["id"] for localization in current} for locale, text in texts: if locale not in known: fail(f"version {version} has no {locale} page; add the language in App Store Connect, or remove {locale}.txt") connect.call("PATCH", f"appStoreVersionLocalizations/{known[locale]}", {"data": { "type": "appStoreVersionLocalizations", "id": known[locale], "attributes": {"whatsNew": text}, }}) if phased: linked = connect.call("GET", f"appStoreVersions/{version_id}", query={"include": "appStoreVersionPhasedRelease"})["data"] if not ((linked.get("relationships") or {}).get("appStoreVersionPhasedRelease") or {}).get("data"): connect.call("POST", "appStoreVersionPhasedReleases", {"data": { "type": "appStoreVersionPhasedReleases", "attributes": {"phasedReleaseState": "INACTIVE"}, "relationships": {"appStoreVersion": ref("appStoreVersions", version_id)}, }}) waiting = connect.call("GET", f"apps/{app}/reviewSubmissions", query={ "filter[platform]": "IOS", "filter[state]": "READY_FOR_REVIEW", "limit": "1", }) if waiting["data"]: submission = waiting["data"][0]["id"] items = connect.call("GET", f"reviewSubmissions/{submission}/items", query={"include": "appStoreVersion", "limit": "50"})["data"] listed = any((((item.get("relationships") or {}).get("appStoreVersion") or {}).get("data") or {}).get("id") == version_id for item in items) else: submission = connect.call("POST", "reviewSubmissions", {"data": { "type": "reviewSubmissions", "attributes": {"platform": "IOS"}, "relationships": {"app": ref("apps", app)}, }})["data"]["id"] listed = False if not listed: connect.call("POST", "reviewSubmissionItems", {"data": { "type": "reviewSubmissionItems", "relationships": {"reviewSubmission": ref("reviewSubmissions", submission), "appStoreVersion": ref("appStoreVersions", version_id)}, }}) connect.call("PATCH", f"reviewSubmissions/{submission}", {"data": { "type": "reviewSubmissions", "id": submission, "attributes": {"submitted": True}, }}) after = "goes live when Apple approves it" if release == "AFTER_APPROVAL" else "waits for you to release it once Apple approves it" print(f"version {version} is submitted for review and {after}" + (", over seven days" if phased else "")) def main(): issuer = setting("ASC_ISSUER_ID") or fail("issuer-id is required (App Store Connect, Users and Access, Integrations)") key_id = setting("ASC_KEY_ID") or fail("key-id is required") key = setting("ASC_PRIVATE_KEY") or fail("private-key is required: the text of the .p8 file") bundle_id = setting("APPLE_BUNDLE_ID") or fail("bundle-id is required") ipa = setting("APPLE_IPA") version = setting("APPLE_VERSION") build = setting("APPLE_BUILD") group_names = [name.strip() for name in re.split(r"[,\n]", setting("APPLE_GROUPS")) if name.strip()] wants_submit = setting("APPLE_SUBMIT", "false") == "true" release = RELEASES.get(setting("APPLE_RELEASE", "after-approval")) or fail("release must be after-approval or manual") phased = setting("APPLE_PHASED", "false") == "true" texts = notes(setting("APPLE_NOTES")) if not ipa and not wants_submit: fail("give ipa to upload a build, or submit: true with version to send a build for review") if not ipa and not version: fail("submitting without ipa needs version, and build if not the newest of that version") if ipa: path = pathlib.Path(ipa) if not path.is_file(): fail(f"{ipa} is not a file") version, build = bundle_info(path, bundle_id) connect = Connect(issuer, key_id, key) app = app_id(connect, bundle_id) groups = beta_groups(connect, app, group_names) if group_names else [] if ipa: build_id = processed(connect, upload(connect, app, path, version, build)) if texts: tester_notes(connect, build_id, texts) if groups: testers(connect, build_id, groups) else: build_id = existing_build(connect, app, version, build) if wants_submit: submit(connect, app, build_id, version, texts, release, phased) if __name__ == "__main__": main()