app-stores/apple/apple.py

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 <locale>.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/<name>.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()