#!/usr/bin/env python3 """ Persistent cache for the Apple Distribution cert + per-bundle provisioning profiles, stored under the workspace's ``creds/`` directory. The caller supplies a retained identity through encrypted CI inputs or an existing legacy cache. CI reuses it and may refresh per-app profiles, but never issues a replacement distribution identity in its temporary checkout. The layout is: creds/ cert.p12 PKCS12 with cert + private key cert.meta.json {cert_id, not_after, p12_password, created_at} cert.registry.json optional registry ownership marker profiles.manifest.json {cert_id, profiles:[ {bundle_id, name, uuid, filename, expiration}]} profiles/.mobileprovision one per signable bundle Reuse rules (encoded in :func:`load_cached_cert` and :func:`find_reusable_profile`): * Cert is reusable iff cert.p12 exists, decrypts with the stored password, NotAfter is more than 30 days away, AND GET /certificates/{cert_id} returns 200 (Apple hasn't revoked it). * Profile is reusable iff the cert is reusable AND the manifest entry has matching cert_id AND expiration is more than 30 days away AND the .mobileprovision file is on disk. An unusable identity is preserved for reconciliation by its credential owner. New signing files remain ephemeral and are never staged into source. """ from __future__ import annotations import base64 import hashlib import json import os from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from asc_common import request from cryptography import x509 from cryptography.hazmat.primitives.serialization import Encoding, pkcs12 from profile_store import ( ProfileEntry, find_reusable_profile, load_profile_manifest, profile_path, write_cached_profile, write_profile_manifest, ) # Refuse identity reuse inside this window; the credential owner provisions its replacement. RENEW_THRESHOLD_DAYS = 30 # Surface a non-fatal warning when the cert is within this many days of # expiry but not yet inside the refusal window, so its owner can provision a replacement. WARN_THRESHOLD_DAYS = 60 # Compatibility default for existing legacy metadata; supplied identities carry their own password. P12_PASSWORD = "ci" # File-name layout under creds/. Centralised so callers never compute # their own paths and the layout stays a single fact in one place. _CERT_P12 = "cert.p12" _CERT_META = "cert.meta.json" _CERT_REGISTRY = "cert.registry.json" _MANIFEST = "profiles.manifest.json" _PROFILES_SUBDIR = "profiles" REGISTRY_MANAGER = "app-robot" @dataclass class CachedCert: cert_id: str not_after: datetime p12_bytes: bytes password: str certificate_sha256: str # --------------------------------------------------------------------------- # # Internal helpers (atomic IO + datetime parsing) # # --------------------------------------------------------------------------- # def _atomic_write(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_bytes(data) os.replace(tmp, path) def _parse_iso(value: str) -> datetime: """Parse an ISO-8601 datetime; tolerate trailing 'Z' for UTC.""" if value.endswith("Z"): value = value[:-1] + "+00:00" parsed = datetime.fromisoformat(value) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return parsed def _now_utc() -> datetime: return datetime.now(timezone.utc) def registry_managed(creds_dir: Path) -> bool: """Return whether the identity is owned by the app-robot registry. The dedicated marker makes ownership detectable even when the P12 or metadata is missing/corrupt. Ownership is intentionally marker-only: a rollback removes this panel-owned file while keeping the working P12/meta cache intact. Unknown managers fail closed instead of falling into legacy rotation. """ marker = creds_dir / _CERT_REGISTRY return _registry_payload_managed(marker) if marker.exists() else False def _registry_payload_managed(path: Path) -> bool: try: payload = json.loads(path.read_text()) except (OSError, ValueError) as exc: raise SystemExit(f"registry marker is unreadable: {exc}") from exc manager = payload.get("managed_by") if not manager: raise SystemExit("registry marker is missing managed_by") if manager != REGISTRY_MANAGER: raise SystemExit(f"unsupported signing registry manager: {manager}") if payload.get("state") != "ready": raise SystemExit("registry-managed signing bundle is not committed and ready") files = (("p12_sha256", _CERT_P12), ("metadata_sha256", _CERT_META)) for field, filename in files: expected = str(payload.get(field) or "").lower() if len(expected) != 64 or any(char not in "0123456789abcdef" for char in expected): raise SystemExit(f"registry marker has invalid {field}") try: actual = hashlib.sha256((path.parent / filename).read_bytes()).hexdigest() except OSError as exc: raise SystemExit(f"registry signing bundle is incomplete: {filename}") from exc if actual != expected: raise SystemExit(f"registry signing bundle digest mismatch: {filename}") return True # --------------------------------------------------------------------------- # # Cert cache # # --------------------------------------------------------------------------- # def load_cached_cert(creds_dir: Path) -> CachedCert | None: """Load a locally valid cache; the caller still verifies it with Apple.""" p12 = creds_dir / _CERT_P12 meta = creds_dir / _CERT_META if not p12.exists() or not meta.exists(): return None try: meta_raw = json.loads(meta.read_text()) cert_id = meta_raw["cert_id"] not_after = _parse_iso(meta_raw["not_after"]) password = meta_raw.get("p12_password") or P12_PASSWORD except (OSError, ValueError, KeyError) as exc: print(f"::warning::cert.meta.json unreadable ({exc}); identity cannot be reused") return None p12_bytes = p12.read_bytes() try: _, certificate, _ = pkcs12.load_key_and_certificates(p12_bytes, password.encode()) except (ValueError, TypeError) as exc: print(f"::warning::cert.p12 decrypt failed ({exc}); identity cannot be reused") return None if certificate is None: print("::warning::cert.p12 has no certificate; identity cannot be reused") return None certificate_sha256 = hashlib.sha256( certificate.public_bytes(Encoding.DER)).hexdigest() declared_digest = str(meta_raw.get("certificate_sha256") or "").lower() if declared_digest and declared_digest != certificate_sha256: print("::warning::cert.meta.json certificate digest does not match cert.p12") return None deadline = _now_utc() + timedelta(days=RENEW_THRESHOLD_DAYS) if not_after <= deadline: print( f"::warning::Cached cert {cert_id} expires {not_after.isoformat()} " f"(within {RENEW_THRESHOLD_DAYS}d); identity requires replacement" ) return None return CachedCert( cert_id=cert_id, not_after=not_after, p12_bytes=p12_bytes, password=password, certificate_sha256=certificate_sha256, ) def verify_cert_alive(token: str, cert_id: str, expected_certificate_sha256: str | None = None) -> bool: """Return True iff GET /certificates/{cert_id} returns 200. A 404 means Apple revoked it (manually or via the cap-rotation done by another signing client); any other non-200 is treated conservatively as not-alive so we regenerate. """ resp = request( "GET", f"/certificates/{cert_id}", token, allow_status={404}, ) if resp.status_code != 200: return False if expected_certificate_sha256 is None: return True try: encoded = resp.json()["data"]["attributes"]["certificateContent"] certificate_der = base64.b64decode(encoded, validate=True) except (KeyError, TypeError, ValueError): return False return hashlib.sha256(certificate_der).hexdigest() == expected_certificate_sha256 def write_cert_bundle( creds_dir: Path, cert_id: str, p12_bytes: bytes, password: str, not_after: datetime, ) -> None: """Persist the cert bundle atomically (p12 + meta).""" creds_dir.mkdir(parents=True, exist_ok=True) _atomic_write(creds_dir / _CERT_P12, p12_bytes) meta = { "cert_id": cert_id, "not_after": not_after.astimezone(timezone.utc).isoformat(), # Password persisted alongside the cert it unlocks — see DESIGN # DECISION block at P12_PASSWORD definition above for the # rationale (encryption-at-rest is decorative when the .p8 lives # in plaintext alongside in the same private repo). "p12_password": password, "created_at": _now_utc().isoformat(), } _atomic_write(creds_dir / _CERT_META, json.dumps(meta, indent=2).encode()) # --------------------------------------------------------------------------- # # Cache invalidation + cross-module utilities # # --------------------------------------------------------------------------- # def invalidate_cache(creds_dir: Path) -> None: """Remove all cache artifacts under ``creds_dir`` (best-effort).""" for name in (_CERT_P12, _CERT_META, _MANIFEST): (creds_dir / name).unlink(missing_ok=True) pdir = creds_dir / _PROFILES_SUBDIR if pdir.is_dir(): for entry in pdir.iterdir(): if entry.suffix == ".mobileprovision": entry.unlink(missing_ok=True) print(f"Invalidated signing cache under {creds_dir}") def cert_not_after_from_der(cert_der: bytes) -> datetime: """Extract NotAfter from a DER-encoded certificate, normalised to UTC. Prefers ``not_valid_after_utc`` (cryptography >= 42); falls back to the deprecated naive-UTC ``not_valid_after`` for older runtimes. Also used by the prepare_signing entrypoint to decide T-60d warnings on cached certs without re-loading the PKCS12. """ cert = x509.load_der_x509_certificate(cert_der) try: return cert.not_valid_after_utc except AttributeError: return cert.not_valid_after.replace(tzinfo=timezone.utc)