verae-mcp-timestamping-server/scripts/testflight-enroll/enroll_lib.py

#!/usr/bin/env python3
"""App Store Connect TestFlight external-tester enrollment helpers.

Secrets come from env (never hard-coded):
  ASC_KEY_ID, ASC_ISSUER_ID, ASC_KEY_PATH
Optional:
  ASC_APP_ID (default 6817018890)
  ASC_BETA_GROUP_ID (default fcfa572e-6494-4781-97b2-728f73e5b3ae)
"""
from __future__ import annotations

import json
import os
import re
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

EMAIL_RE = re.compile(r"^[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}$")

DEFAULT_APP_ID = "6817018890"
DEFAULT_GROUP_ID = "fcfa572e-6494-4781-97b2-728f73e5b3ae"
ASC_BASE = "https://api.appstoreconnect.apple.com"


def validate_email(email: str) -> str:
    email = (email or "").strip()
    if not email or len(email) > 254 or not EMAIL_RE.match(email):
        raise ValueError("invalid_email")
    return email.lower()


def _require_env() -> tuple[str, str, str]:
    key_id = os.environ.get("ASC_KEY_ID", "").strip()
    issuer = os.environ.get("ASC_ISSUER_ID", "").strip()
    path = os.environ.get("ASC_KEY_PATH", "").strip()
    if not key_id or not issuer or not path:
        raise RuntimeError("missing_asc_env")
    if not os.path.isfile(path):
        raise RuntimeError("missing_asc_key_file")
    return key_id, issuer, path


def make_token() -> str:
    """ES256 JWT for App Store Connect API (max ~20 min)."""
    import jwt  # PyJWT

    key_id, issuer, path = _require_env()
    with open(path, "r", encoding="utf-8") as f:
        private_key = f.read()
    now = int(time.time())
    payload = {
        "iss": issuer,
        "iat": now,
        "exp": now + 15 * 60,
        "aud": "appstoreconnect-v1",
    }
    return jwt.encode(
        payload,
        private_key,
        algorithm="ES256",
        headers={"kid": key_id, "typ": "JWT"},
    )


def _request(method: str, path: str, body: dict | None = None, token: str | None = None) -> tuple[int, Any]:
    token = token or make_token()
    url = path if path.startswith("http") else ASC_BASE + path
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", "Bearer " + token)
    req.add_header("Accept", "application/json")
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=45) as resp:
            raw = resp.read().decode("utf-8") or "{}"
            return resp.status, json.loads(raw) if raw.strip() else {}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            parsed = json.loads(raw) if raw.strip() else {}
        except json.JSONDecodeError:
            parsed = {"raw": raw}
        return e.code, parsed


def find_tester_by_email(email: str, token: str) -> dict | None:
    q = urllib.parse.urlencode({
        "filter[email]": email,
        "filter[apps]": os.environ.get("ASC_APP_ID", DEFAULT_APP_ID),
        "limit": "5",
    })
    status, data = _request("GET", f"/v1/betaTesters?{q}", token=token)
    if status != 200:
        return None
    items = data.get("data") or []
    return items[0] if items else None


def create_tester(email: str, token: str, first: str = "Verae", last: str = "Tester") -> tuple[int, Any]:
    app_id = os.environ.get("ASC_APP_ID", DEFAULT_APP_ID)
    group_id = os.environ.get("ASC_BETA_GROUP_ID", DEFAULT_GROUP_ID)
    body = {
        "data": {
            "type": "betaTesters",
            "attributes": {
                "email": email,
                "firstName": first[:30] or "Verae",
                "lastName": last[:30] or "Tester",
            },
            "relationships": {
                "betaGroups": {
                    "data": [{"type": "betaGroups", "id": group_id}]
                },
                "apps": {
                    "data": [{"type": "apps", "id": app_id}]
                },
            },
        }
    }
    return _request("POST", "/v1/betaTesters", body=body, token=token)


def add_tester_to_group(tester_id: str, token: str) -> tuple[int, Any]:
    group_id = os.environ.get("ASC_BETA_GROUP_ID", DEFAULT_GROUP_ID)
    body = {
        "data": [{"type": "betaTesters", "id": tester_id}]
    }
    return _request(
        "POST",
        f"/v1/betaGroups/{group_id}/relationships/betaTesters",
        body=body,
        token=token,
    )


def resend_invitation(tester_id: str, token: str) -> tuple[int, Any]:
    # App Store Connect: POST /v1/betaTesterInvitations
    app_id = os.environ.get("ASC_APP_ID", DEFAULT_APP_ID)
    body = {
        "data": {
            "type": "betaTesterInvitations",
            "relationships": {
                "betaTester": {"data": {"type": "betaTesters", "id": tester_id}},
                "app": {"data": {"type": "apps", "id": app_id}},
            },
        }
    }
    return _request("POST", "/v1/betaTesterInvitations", body=body, token=token)


def enroll(email: str, note: str = "") -> dict:
    """Validate + create/add external tester + resend invite. Returns public result dict."""
    email = validate_email(email)
    token = make_token()
    existing = find_tester_by_email(email, token)
    created = False
    tester_id = None
    steps: list[str] = []

    if existing:
        tester_id = existing["id"]
        steps.append("found_existing")
        st, _ = add_tester_to_group(tester_id, token)
        # 204 success; 409 often already in group
        if st in (204, 409):
            steps.append("group_ok" if st == 204 else "already_in_group")
        else:
            steps.append(f"group_status_{st}")
    else:
        st, data = create_tester(email, token)
        if st in (201, 200):
            created = True
            tester_id = (data.get("data") or {}).get("id")
            steps.append("created")
        elif st == 409:
            # race / already exists — look up again
            steps.append("create_conflict")
            existing = find_tester_by_email(email, token)
            if not existing:
                return {
                    "ok": False,
                    "error": "tester_conflict",
                    "message": "Apple reported a conflict creating this tester. Try again or email marchon@gmail.com.",
                    "steps": steps,
                }
            tester_id = existing["id"]
            st2, _ = add_tester_to_group(tester_id, token)
            steps.append("group_ok" if st2 in (204, 409) else f"group_status_{st2}")
        else:
            detail = ""
            errs = (data or {}).get("errors") or []
            if errs:
                detail = errs[0].get("detail") or errs[0].get("title") or ""
            return {
                "ok": False,
                "error": "asc_create_failed",
                "message": detail or f"App Store Connect create failed ({st}).",
                "steps": steps,
                "status": st,
            }

    if tester_id:
        rst, _ = resend_invitation(tester_id, token)
        if rst in (201, 200, 204):
            steps.append("invite_resent")
        else:
            steps.append(f"invite_status_{rst}")

    return {
        "ok": True,
        "email": email,
        "created": created,
        "tester_id": tester_id,
        "steps": steps,
        "join_url": "https://testflight.apple.com/join/dsHsyx2Z",
        "beta_review_note": (
            "External builds may still be WAITING_FOR_BETA_REVIEW. "
            "Enrollment succeeded; you may not see the build until Apple approves the beta."
        ),
        "note_received": bool((note or "").strip()),
    }