|
| 1 | +# ***************************************************************************** |
| 2 | +# Copyright (c) 2026 IBM Corporation and other Contributors. |
| 3 | +# |
| 4 | +# All rights reserved. This program and the accompanying materials |
| 5 | +# are made available under the terms of the Eclipse Public License v1.0 |
| 6 | +# which accompanies this distribution, and is available at |
| 7 | +# http://www.eclipse.org/legal/epl-v10.html |
| 8 | +# |
| 9 | +# ***************************************************************************** |
| 10 | + |
| 11 | +"""GitHub Checks API helpers for FVT result reporting. |
| 12 | +
|
| 13 | +Provides createCheckRun() and updateCheckRun() to post check run status |
| 14 | +against specific commits on github.ibm.com, using a GitHub App for auth. |
| 15 | +
|
| 16 | +Authentication is handled automatically. Set GITHUB_APP_PRIVATE_KEY in the |
| 17 | +environment; an installation token is fetched, cached, and refreshed |
| 18 | +transparently. |
| 19 | +
|
| 20 | +Environment variables: |
| 21 | + GITHUB_APP_PRIVATE_KEY (required) PEM-encoded RSA private key. |
| 22 | + GITHUB_API_BASE (optional) Override API base URL. |
| 23 | + Defaults to https://github.ibm.com/api/v3 |
| 24 | + GITHUB_APP_ID (optional) Override app ID. Defaults to 6035. |
| 25 | + GITHUB_APP_INSTALLATION_ID (optional) Skip installation lookup. |
| 26 | +""" |
| 27 | + |
| 28 | +import base64 |
| 29 | +import json |
| 30 | +import logging |
| 31 | +import os |
| 32 | +import time |
| 33 | +import urllib.error |
| 34 | +import urllib.request |
| 35 | +from dataclasses import dataclass, field |
| 36 | +from datetime import datetime, timezone |
| 37 | + |
| 38 | +from cryptography.hazmat.primitives import hashes, serialization |
| 39 | +from cryptography.hazmat.primitives.asymmetric import padding |
| 40 | + |
| 41 | +logger = logging.getLogger(__name__) |
| 42 | + |
| 43 | +_GITHUB_API_BASE = os.environ.get("GITHUB_API_BASE", "https://github.ibm.com/api/v3") |
| 44 | +_GITHUB_APP_ID = os.environ.get("GITHUB_APP_ID", "6035") |
| 45 | + |
| 46 | + |
| 47 | +# --------------------------------------------------------------------------- |
| 48 | +# Token cache |
| 49 | +# --------------------------------------------------------------------------- |
| 50 | +@dataclass |
| 51 | +class _TokenCache: |
| 52 | + token: str = "" |
| 53 | + expiresAt: float = field(default_factory=float) |
| 54 | + |
| 55 | + |
| 56 | +_tokenCache: dict[str, _TokenCache] = {} |
| 57 | + |
| 58 | + |
| 59 | +# --------------------------------------------------------------------------- |
| 60 | +# Internal helpers |
| 61 | +# --------------------------------------------------------------------------- |
| 62 | +def _buildJwt(privateKeyPem: str) -> str: |
| 63 | + """Build a signed RS256 JWT to authenticate as the GitHub App.""" |
| 64 | + key = serialization.load_pem_private_key(privateKeyPem.encode(), password=None) |
| 65 | + now = int(time.time()) |
| 66 | + header = base64.urlsafe_b64encode(json.dumps({"alg": "RS256", "typ": "JWT"}).encode()).rstrip(b"=") |
| 67 | + payload = base64.urlsafe_b64encode(json.dumps({"iat": now - 60, "exp": now + 540, "iss": _GITHUB_APP_ID}).encode()).rstrip(b"=") |
| 68 | + message = header + b"." + payload |
| 69 | + sig = base64.urlsafe_b64encode(key.sign(message, padding.PKCS1v15(), hashes.SHA256())).rstrip(b"=") |
| 70 | + return (message + b"." + sig).decode() |
| 71 | + |
| 72 | + |
| 73 | +def _apiRequest(method: str, url: str, token: str, payload: dict | None = None, isJwt: bool = False) -> dict: |
| 74 | + """Send an authenticated request to the GitHub API.""" |
| 75 | + data = json.dumps(payload).encode() if payload else None |
| 76 | + authScheme = "Bearer" if isJwt else "token" |
| 77 | + req = urllib.request.Request( |
| 78 | + url, |
| 79 | + data=data, |
| 80 | + method=method, |
| 81 | + headers={ |
| 82 | + "Accept": "application/vnd.github+json", |
| 83 | + "Authorization": f"{authScheme} {token}", |
| 84 | + "X-GitHub-Api-Version": "2022-11-28", |
| 85 | + "Content-Type": "application/json", |
| 86 | + }, |
| 87 | + ) |
| 88 | + try: |
| 89 | + with urllib.request.urlopen(req) as resp: |
| 90 | + return json.loads(resp.read().decode()) |
| 91 | + except urllib.error.HTTPError as e: |
| 92 | + raise RuntimeError(f"GHE API {method} {url} → {e.code}: {e.read().decode(errors='replace')}") from e |
| 93 | + except urllib.error.URLError as e: |
| 94 | + raise RuntimeError(f"GHE API request failed: {e.reason}") from e |
| 95 | + |
| 96 | + |
| 97 | +def _getInstallationToken(org: str) -> str: |
| 98 | + """Return a valid installation token for *org*, refreshing when near expiry.""" |
| 99 | + cache = _tokenCache.get(org) |
| 100 | + if cache and cache.token and time.time() < cache.expiresAt - 60: |
| 101 | + return cache.token |
| 102 | + |
| 103 | + privateKeyPem = os.environ.get("GITHUB_APP_PRIVATE_KEY") |
| 104 | + if not privateKeyPem: |
| 105 | + raise RuntimeError("GITHUB_APP_PRIVATE_KEY environment variable is not set") |
| 106 | + |
| 107 | + jwt = _buildJwt(privateKeyPem) |
| 108 | + |
| 109 | + # Resolve installation ID for this org (or use explicit override) |
| 110 | + installationId = os.environ.get("GITHUB_APP_INSTALLATION_ID") |
| 111 | + if not installationId: |
| 112 | + installations = _apiRequest("GET", f"{_GITHUB_API_BASE}/app/installations", jwt, isJwt=True) |
| 113 | + for inst in installations: |
| 114 | + if inst.get("account", {}).get("login", "").lower() == org.lower(): |
| 115 | + installationId = str(inst["id"]) |
| 116 | + break |
| 117 | + if not installationId: |
| 118 | + raise RuntimeError(f"No GHE App installation found for org '{org}'") |
| 119 | + |
| 120 | + body = _apiRequest("POST", f"{_GITHUB_API_BASE}/app/installations/{installationId}/access_tokens", jwt, payload={}, isJwt=True) |
| 121 | + token = body.get("token") |
| 122 | + if not token: |
| 123 | + raise RuntimeError(f"No token in GHE access_tokens response: {body}") |
| 124 | + |
| 125 | + _tokenCache[org] = _TokenCache(token=token, expiresAt=time.time() + 3600) |
| 126 | + logger.debug("Fetched GHE installation token for org=%s", org) |
| 127 | + return token |
| 128 | + |
| 129 | + |
| 130 | +# --------------------------------------------------------------------------- |
| 131 | +# Public API |
| 132 | +# --------------------------------------------------------------------------- |
| 133 | +def createCheckRun(name: str, repoSlug: str, commitSha: str, detailsUrl: str = "") -> int: |
| 134 | + """Create a GitHub Check Run in 'in_progress' state. Returns the check run ID.""" |
| 135 | + org = repoSlug.split("/")[0] |
| 136 | + appToken = _getInstallationToken(org) |
| 137 | + isoNow = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") |
| 138 | + payload: dict = {"name": name, "head_sha": commitSha, "status": "in_progress", "started_at": isoNow} |
| 139 | + if detailsUrl: |
| 140 | + payload["details_url"] = detailsUrl |
| 141 | + response = _apiRequest("POST", f"{_GITHUB_API_BASE}/repos/{repoSlug}/check-runs", appToken, payload) |
| 142 | + checkRunId = response.get("id") |
| 143 | + if not checkRunId: |
| 144 | + raise RuntimeError(f"Failed to create check run '{name}': {response}") |
| 145 | + logger.debug("Created check run id=%s for %s@%s", checkRunId, repoSlug, commitSha[:8]) |
| 146 | + return checkRunId |
| 147 | + |
| 148 | + |
| 149 | +def updateCheckRun(checkRunId: int, repoSlug: str, conclusion: str, detailsUrl: str = "", outputTitle: str = "", outputSummary: str = "") -> None: |
| 150 | + """Complete an existing GitHub Check Run with the given conclusion.""" |
| 151 | + org = repoSlug.split("/")[0] |
| 152 | + appToken = _getInstallationToken(org) |
| 153 | + isoNow = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") |
| 154 | + payload: dict = {"status": "completed", "conclusion": conclusion, "completed_at": isoNow} |
| 155 | + if detailsUrl: |
| 156 | + payload["details_url"] = detailsUrl |
| 157 | + if outputTitle and outputSummary: |
| 158 | + payload["output"] = {"title": outputTitle, "summary": outputSummary} |
| 159 | + _apiRequest("PATCH", f"{_GITHUB_API_BASE}/repos/{repoSlug}/check-runs/{checkRunId}", appToken, payload) |
| 160 | + logger.debug("Updated check run id=%s conclusion=%s", checkRunId, conclusion) |
| 161 | + |
| 162 | + |
| 163 | +def findCheckRun(name: str, repoSlug: str, commitSha: str) -> int | None: |
| 164 | + """Return the ID of the most recent existing check run matching name+commit, or None. |
| 165 | +
|
| 166 | + Used to upsert — update an existing check run rather than creating a duplicate. |
| 167 | + """ |
| 168 | + org = repoSlug.split("/")[0] |
| 169 | + appToken = _getInstallationToken(org) |
| 170 | + url = f"{_GITHUB_API_BASE}/repos/{repoSlug}/commits/{commitSha}/check-runs" |
| 171 | + response = _apiRequest("GET", url, appToken) |
| 172 | + for run in response.get("check_runs", []): |
| 173 | + if run.get("name") == name: |
| 174 | + logger.debug("Found existing check run id=%s name=%s", run["id"], name) |
| 175 | + return run["id"] |
| 176 | + return None |
0 commit comments