Coverage for extra_codeowners/github.py: 98%
336 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-13 09:46 +0000
1"""Least-privilege asynchronous GitHub REST API client."""
3from __future__ import annotations
5import asyncio
6import math
7from dataclasses import dataclass
8from datetime import UTC, datetime, timedelta
9from email.utils import parsedate_to_datetime
10from typing import Any, Final, NoReturn
11from urllib.parse import quote
13import httpx
14import jwt
15from cryptography.hazmat.primitives import serialization
16from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
18MAX_PULL_FILES: Final = 3000
19MAX_PULL_REVIEWS: Final = 1000
20MAX_CONFIG_BYTES: Final = 1_000_000
21MAX_CODEOWNERS_BYTES: Final = 3 * 1024 * 1024
24class GitHubError(RuntimeError):
25 """Base class for GitHub API failures."""
28class GitHubAPIError(GitHubError):
29 """A non-success GitHub API response."""
31 def __init__(self, status_code: int, method: str, path: str, message: str) -> None:
32 super().__init__(f"GitHub API {method} {path} returned {status_code}: {message}")
33 self.status_code = status_code
34 self.method = method
35 self.path = path
38class GitHubRateLimitError(GitHubAPIError):
39 """GitHub asked the caller to wait before retrying."""
41 def __init__(
42 self,
43 status_code: int,
44 method: str,
45 path: str,
46 message: str,
47 retry_after_seconds: int,
48 ) -> None:
49 super().__init__(status_code, method, path, message)
50 self.retry_after_seconds = retry_after_seconds
53class PullRequestTooLargeError(GitHubError):
54 """A pull request exceeds GitHub's files API visibility limit."""
57@dataclass(frozen=True, slots=True)
58class InstallationToken:
59 """Cached installation token and conservative expiry."""
61 value: str
62 expires_at: datetime
65class GitHubClient:
66 """GitHub App and installation REST API client.
68 The caller supplies only an App ID and PEM key. Installation tokens are
69 cached in memory and never logged or persisted.
70 """
72 def __init__(
73 self,
74 app_id: int,
75 private_key: str,
76 *,
77 api_url: str = "https://api.github.com",
78 api_version: str = "2026-03-10",
79 timeout_seconds: float = 20,
80 transport: httpx.AsyncBaseTransport | None = None,
81 ) -> None:
82 self.app_id = app_id
83 try:
84 loaded_key = serialization.load_pem_private_key(private_key.encode(), password=None)
85 except (TypeError, ValueError) as error:
86 msg = "GitHub App private key is not a valid unencrypted PEM private key"
87 raise ValueError(msg) from error
88 if not isinstance(loaded_key, RSAPrivateKey):
89 msg = "GitHub App private key must be an RSA private key"
90 raise ValueError(msg)
91 self._private_key = loaded_key
92 self._tokens: dict[int, InstallationToken] = {}
93 self._token_locks: dict[int, asyncio.Lock] = {}
94 self._http = httpx.AsyncClient(
95 base_url=api_url.rstrip("/"),
96 timeout=httpx.Timeout(timeout_seconds),
97 transport=transport,
98 headers={
99 "Accept": "application/vnd.github+json",
100 "User-Agent": "extra-codeowners",
101 "X-GitHub-Api-Version": api_version,
102 },
103 )
105 async def close(self) -> None:
106 """Close pooled HTTP connections."""
107 await self._http.aclose()
109 def _app_jwt(self) -> str:
110 now = datetime.now(UTC)
111 payload = {
112 "iat": int((now - timedelta(seconds=60)).timestamp()),
113 "exp": int((now + timedelta(minutes=9)).timestamp()),
114 "iss": str(self.app_id),
115 }
116 return str(jwt.encode(payload, self._private_key, algorithm="RS256"))
118 async def _installation_token(self, installation_id: int) -> str:
119 cached = self._tokens.get(installation_id)
120 now = datetime.now(UTC)
121 if cached is not None and cached.expires_at > now + timedelta(minutes=5):
122 return cached.value
124 lock = self._token_locks.setdefault(installation_id, asyncio.Lock())
125 async with lock:
126 cached = self._tokens.get(installation_id)
127 if cached is not None and cached.expires_at > now + timedelta(minutes=5):
128 return cached.value
129 response = await self._request(
130 "POST",
131 f"/app/installations/{installation_id}/access_tokens",
132 app_authenticated=True,
133 json={
134 "permissions": {
135 "checks": "write",
136 "contents": "read",
137 "members": "read",
138 "pull_requests": "read",
139 }
140 },
141 )
142 token = response.get("token")
143 expires_at = response.get("expires_at")
144 if not isinstance(token, str) or not isinstance(expires_at, str):
145 msg = "installation-token response omitted token or expiry"
146 raise GitHubError(msg)
147 parsed_expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
148 self._tokens[installation_id] = InstallationToken(token, parsed_expiry)
149 return token
151 async def _request(
152 self,
153 method: str,
154 path: str,
155 *,
156 installation_id: int | None = None,
157 app_authenticated: bool = False,
158 params: dict[str, Any] | None = None,
159 json: dict[str, Any] | None = None,
160 headers: dict[str, str] | None = None,
161 allow_not_found: bool = False,
162 ) -> dict[str, Any]:
163 if app_authenticated == (installation_id is not None):
164 msg = "request must use exactly one authentication mode"
165 raise ValueError(msg)
166 response = await self._authenticated_response(
167 method,
168 path,
169 installation_id=installation_id,
170 app_authenticated=app_authenticated,
171 params=params,
172 json=json,
173 headers=headers,
174 )
175 if allow_not_found and response.status_code == 404:
176 return {}
177 if response.is_success:
178 if response.status_code == 204 or not response.content:
179 return {}
180 value = response.json()
181 if not isinstance(value, dict):
182 msg = f"expected object response from {method} {path}"
183 raise GitHubError(msg)
184 return value
185 self._raise_api_error(response, method, path)
186 raise AssertionError("unreachable") # pragma: no cover
188 async def _authenticated_response(
189 self,
190 method: str,
191 path: str,
192 *,
193 installation_id: int | None = None,
194 app_authenticated: bool = False,
195 params: dict[str, Any] | None = None,
196 json: dict[str, Any] | None = None,
197 headers: dict[str, str] | None = None,
198 ) -> httpx.Response:
199 """Send an authenticated request, refreshing a rejected installation token once."""
200 attempts = 1 if app_authenticated else 2
201 for attempt in range(attempts):
202 if app_authenticated:
203 token = self._app_jwt()
204 else:
205 assert installation_id is not None
206 token = await self._installation_token(installation_id)
207 response = await self._http.request(
208 method,
209 path,
210 params=params,
211 json=json,
212 headers={**(headers or {}), "Authorization": f"Bearer {token}"},
213 )
214 if response.status_code != 401 or app_authenticated or attempt > 0:
215 return response
216 assert installation_id is not None
217 cached = self._tokens.get(installation_id)
218 if cached is not None and cached.value == token: 218 ↛ 201line 218 didn't jump to line 201 because the condition on line 218 was always true
219 self._tokens.pop(installation_id, None)
220 raise AssertionError("unreachable") # pragma: no cover
222 async def _authenticated_streaming_response(
223 self,
224 path: str,
225 installation_id: int,
226 *,
227 params: dict[str, Any] | None = None,
228 headers: dict[str, str] | None = None,
229 ) -> httpx.Response:
230 """Return an open streamed response, refreshing a rejected token once.
232 The caller owns the returned response and must close it. A rejected
233 response is closed before its cached token is evicted and retried.
234 """
235 for attempt in range(2):
236 token = await self._installation_token(installation_id)
237 request = self._http.build_request(
238 "GET",
239 path,
240 params=params,
241 headers={**(headers or {}), "Authorization": f"Bearer {token}"},
242 )
243 response = await self._http.send(request, stream=True)
244 if response.status_code != 401 or attempt > 0:
245 return response
246 await response.aclose()
247 cached = self._tokens.get(installation_id)
248 if cached is not None and cached.value == token: 248 ↛ 235line 248 didn't jump to line 235 because the condition on line 248 was always true
249 self._tokens.pop(installation_id, None)
250 raise AssertionError("unreachable") # pragma: no cover
252 @staticmethod
253 async def _bounded_error_response(response: httpx.Response, max_bytes: int) -> httpx.Response:
254 """Read a bounded error prefix into a response safe for error parsing."""
255 limit = max(0, min(max_bytes, 1000))
256 content = bytearray()
257 if limit: 257 ↛ 265line 257 didn't jump to line 265 because the condition on line 257 was always true
258 async for chunk in response.aiter_bytes(chunk_size=min(limit, 64 * 1024)):
259 remaining = limit - len(content)
260 content.extend(chunk[:remaining])
261 if len(content) == limit:
262 break
263 # aiter_bytes() has already decoded transfer/content encodings. Do not
264 # make the synthetic response decode the bounded prefix a second time.
265 headers = [
266 (name, value)
267 for name, value in response.headers.multi_items()
268 if name.lower() not in {"content-encoding", "content-length", "transfer-encoding"}
269 ]
270 return httpx.Response(
271 response.status_code,
272 headers=headers,
273 content=bytes(content),
274 request=response.request,
275 )
277 @staticmethod
278 def _response_message(response: httpx.Response) -> str:
279 message = response.text[:1000]
280 try:
281 body = response.json()
282 if isinstance(body, dict) and isinstance(body.get("message"), str): 282 ↛ 286line 282 didn't jump to line 286 because the condition on line 282 was always true
283 message = body["message"]
284 except ValueError:
285 return message
286 return message
288 @staticmethod
289 def _retry_delay(response: httpx.Response, message: str) -> int | None:
290 retry_after = response.headers.get("retry-after")
291 remaining = response.headers.get("x-ratelimit-remaining")
292 is_limited = response.status_code == 429 or (
293 response.status_code == 403
294 and (retry_after is not None or remaining == "0" or "rate limit" in message.lower())
295 )
296 if not is_limited:
297 return None
299 delay: float | None = None
300 if retry_after is not None:
301 try:
302 delay = float(retry_after)
303 except ValueError:
304 try:
305 parsed = parsedate_to_datetime(retry_after)
306 delay = (parsed - datetime.now(UTC)).total_seconds()
307 except (TypeError, ValueError, OverflowError):
308 delay = None
309 if delay is None:
310 reset = response.headers.get("x-ratelimit-reset")
311 if reset is not None:
312 try:
313 delay = float(reset) - datetime.now(UTC).timestamp()
314 except ValueError:
315 delay = None
316 return max(1, min(math.ceil(delay if delay is not None else 60), 86_400))
318 @classmethod
319 def _raise_api_error(cls, response: httpx.Response, method: str, path: str) -> NoReturn:
320 message = cls._response_message(response)
321 retry_after = cls._retry_delay(response, message)
322 if retry_after is not None:
323 raise GitHubRateLimitError(
324 response.status_code,
325 method,
326 path,
327 message,
328 retry_after,
329 )
330 raise GitHubAPIError(response.status_code, method, path, message)
332 async def _get_list(
333 self,
334 path: str,
335 installation_id: int,
336 *,
337 params: dict[str, Any] | None = None,
338 max_items: int | None = None,
339 ) -> list[dict[str, Any]]:
340 query = {**(params or {}), "per_page": 100}
341 items: list[dict[str, Any]] = []
342 page = 1
343 while True:
344 query["page"] = page
345 response = await self._authenticated_response(
346 "GET",
347 path,
348 installation_id=installation_id,
349 params=query,
350 )
351 if not response.is_success:
352 self._raise_api_error(response, "GET", path)
353 page_items = response.json()
354 if not isinstance(page_items, list):
355 msg = f"expected list response from GET {path}"
356 raise GitHubError(msg)
357 for item in page_items:
358 if not isinstance(item, dict):
359 msg = f"expected object items from GET {path}"
360 raise GitHubError(msg)
361 items.append(item)
362 if max_items is not None and len(items) > max_items:
363 msg = f"GET {path} exceeded the supported {max_items}-item limit"
364 raise PullRequestTooLargeError(msg)
365 if len(page_items) < 100:
366 break
367 page += 1
368 return items
370 async def get_pull(self, installation_id: int, repository: str, number: int) -> dict[str, Any]:
371 """Fetch current pull request metadata."""
372 return await self._request(
373 "GET", f"/repos/{repository}/pulls/{number}", installation_id=installation_id
374 )
376 async def get_pull_files(
377 self, installation_id: int, repository: str, number: int
378 ) -> list[dict[str, Any]]:
379 """Fetch every visible changed file, failing at GitHub's 3,000-file cap."""
380 items = await self._get_list(
381 f"/repos/{repository}/pulls/{number}/files",
382 installation_id,
383 max_items=MAX_PULL_FILES,
384 )
385 if len(items) >= MAX_PULL_FILES: 385 ↛ 389line 385 didn't jump to line 389 because the condition on line 385 was always true
386 # GitHub truncates at 3,000, so exactly 3,000 cannot prove completeness.
387 msg = f"pull request has at least GitHub's {MAX_PULL_FILES:,}-file API limit"
388 raise PullRequestTooLargeError(msg)
389 return items
391 async def get_reviews(
392 self, installation_id: int, repository: str, number: int
393 ) -> list[dict[str, Any]]:
394 """Fetch all submitted pull request reviews."""
395 return await self._get_list(
396 f"/repos/{repository}/pulls/{number}/reviews",
397 installation_id,
398 max_items=MAX_PULL_REVIEWS,
399 )
401 async def get_codeowners_errors(
402 self, installation_id: int, repository: str, ref: str
403 ) -> list[dict[str, Any]]:
404 """Ask GitHub to validate CODEOWNERS at the exact base revision."""
405 result = await self._request(
406 "GET",
407 f"/repos/{repository}/codeowners/errors",
408 installation_id=installation_id,
409 params={"ref": ref},
410 )
411 errors = result.get("errors")
412 if not isinstance(errors, list):
413 msg = "CODEOWNERS errors response omitted its errors list"
414 raise GitHubError(msg)
415 if any(not isinstance(item, dict) for item in errors):
416 msg = "CODEOWNERS errors response contained a malformed error"
417 raise GitHubError(msg)
418 return errors
420 async def get_file_text(
421 self,
422 installation_id: int,
423 repository: str,
424 path: str,
425 *,
426 ref: str | None = None,
427 max_bytes: int = MAX_CONFIG_BYTES,
428 ) -> str | None:
429 """Read bounded raw UTF-8 repository content, returning ``None`` for 404."""
430 params = {"ref": ref} if ref is not None else None
431 encoded_path = quote(path, safe="/")
432 endpoint = f"/repos/{repository}/contents/{encoded_path}"
433 response = await self._authenticated_streaming_response(
434 endpoint,
435 installation_id,
436 params=params,
437 headers={
438 "Accept": "application/vnd.github.raw+json",
439 },
440 )
441 try:
442 if response.status_code == 404:
443 return None
444 if not response.is_success:
445 error_response = await self._bounded_error_response(response, max_bytes)
446 self._raise_api_error(error_response, "GET", endpoint)
448 content_length = response.headers.get("content-length")
449 normalized_length = content_length.strip() if content_length is not None else ""
450 declared_length = (
451 int(normalized_length)
452 if normalized_length.isascii() and normalized_length.isdecimal()
453 else None
454 )
455 if declared_length is not None and declared_length > max_bytes:
456 msg = f"{repository}:{path} exceeds the {max_bytes}-byte file limit"
457 raise GitHubError(msg)
459 content = bytearray()
460 chunk_size = max(1, min(max_bytes, 64 * 1024))
461 async for chunk in response.aiter_bytes(chunk_size=chunk_size):
462 if len(chunk) > max_bytes - len(content):
463 msg = f"{repository}:{path} exceeds the {max_bytes}-byte file limit"
464 raise GitHubError(msg)
465 content.extend(chunk)
466 try:
467 return content.decode("utf-8")
468 except UnicodeDecodeError as error:
469 msg = f"{repository}:{path} is not valid UTF-8"
470 raise GitHubError(msg) from error
471 finally:
472 await response.aclose()
474 async def team_member(
475 self,
476 installation_id: int,
477 organization: str,
478 team_slug: str,
479 username: str,
480 ) -> bool:
481 """Return whether a user is an active member of an organization team."""
482 result = await self._request(
483 "GET",
484 f"/orgs/{organization}/teams/{team_slug}/memberships/{username}",
485 installation_id=installation_id,
486 allow_not_found=True,
487 )
488 return result.get("state") == "active"
490 async def user_can_own_repository(
491 self,
492 installation_id: int,
493 repository: str,
494 username: str,
495 ) -> bool:
496 """Return whether a direct CODEOWNER candidate currently has write access."""
497 result = await self._request(
498 "GET",
499 f"/repos/{repository}/collaborators/{username}/permission",
500 installation_id=installation_id,
501 allow_not_found=True,
502 )
503 return result.get("permission") in {"write", "maintain", "admin"}
505 async def team_can_own_repository(
506 self,
507 installation_id: int,
508 organization: str,
509 team_slug: str,
510 repository: str,
511 ) -> bool:
512 """Return whether a visible team explicitly grants write access."""
513 team = await self._request(
514 "GET",
515 f"/orgs/{organization}/teams/{team_slug}",
516 installation_id=installation_id,
517 allow_not_found=True,
518 )
519 if team.get("privacy") != "closed":
520 return False
521 repository_metadata = await self._request(
522 "GET",
523 f"/orgs/{organization}/teams/{team_slug}/repos/{repository}",
524 installation_id=installation_id,
525 headers={"Accept": "application/vnd.github.v3.repository+json"},
526 allow_not_found=True,
527 )
528 permissions = repository_metadata.get("permissions")
529 if not isinstance(permissions, dict):
530 return False
531 return any(permissions.get(level) is True for level in ("push", "maintain", "admin"))
533 async def get_app(self, installation_id: int, slug: str) -> dict[str, Any]:
534 """Fetch independently observed public identity metadata for an allowed App."""
535 return await self._request(
536 "GET",
537 f"/apps/{slug}",
538 installation_id=installation_id,
539 )
541 async def _latest_check_run_id(
542 self,
543 installation_id: int,
544 repository: str,
545 head_sha: str,
546 check_name: str,
547 ) -> int | None:
548 existing = await self._request(
549 "GET",
550 f"/repos/{repository}/commits/{head_sha}/check-runs",
551 installation_id=installation_id,
552 params={
553 "check_name": check_name,
554 "filter": "latest",
555 "app_id": self.app_id,
556 "per_page": 100,
557 },
558 )
559 runs = existing.get("check_runs", [])
560 if not isinstance(runs, list):
561 msg = "check-runs response omitted its list"
562 raise GitHubError(msg)
563 for run in runs:
564 if not isinstance(run, dict) or run.get("name") != check_name:
565 continue
566 app = run.get("app")
567 if isinstance(app, dict) and app.get("id") == self.app_id:
568 candidate = run.get("id")
569 if isinstance(candidate, int):
570 return candidate
571 return None
573 async def has_check_run(
574 self,
575 installation_id: int,
576 repository: str,
577 head_sha: str,
578 check_name: str,
579 ) -> bool:
580 """Return whether this App already manages the named check on a commit."""
581 return (
582 await self._latest_check_run_id(
583 installation_id,
584 repository,
585 head_sha,
586 check_name,
587 )
588 is not None
589 )
591 async def upsert_check_run(
592 self,
593 installation_id: int,
594 repository: str,
595 head_sha: str,
596 check_name: str,
597 *,
598 status: str = "completed",
599 conclusion: str | None = None,
600 title: str,
601 summary: str,
602 text: str = "",
603 details_url: str | None = None,
604 external_id: str | None = None,
605 ) -> int:
606 """Create or update this App's latest named check on a commit."""
607 if status not in {"queued", "in_progress", "completed"}:
608 raise ValueError("unsupported check-run status")
609 if status == "completed" and conclusion is None:
610 raise ValueError("completed check runs require a conclusion")
611 if status != "completed" and conclusion is not None:
612 raise ValueError("non-completed check runs cannot have a conclusion")
613 check_id = await self._latest_check_run_id(
614 installation_id,
615 repository,
616 head_sha,
617 check_name,
618 )
620 payload: dict[str, Any] = {
621 "name": check_name,
622 "status": status,
623 "output": {"title": title[:255], "summary": summary[:65535], "text": text[:65535]},
624 }
625 if status == "completed":
626 payload["conclusion"] = conclusion
627 payload["completed_at"] = datetime.now(UTC).isoformat().replace("+00:00", "Z")
628 elif status == "in_progress": 628 ↛ 630line 628 didn't jump to line 630 because the condition on line 628 was always true
629 payload["started_at"] = datetime.now(UTC).isoformat().replace("+00:00", "Z")
630 if details_url is not None: 630 ↛ 631line 630 didn't jump to line 631 because the condition on line 630 was never true
631 payload["details_url"] = details_url
632 if external_id is not None:
633 payload["external_id"] = external_id[:255]
634 if check_id is None:
635 payload["head_sha"] = head_sha
636 result = await self._request(
637 "POST",
638 f"/repos/{repository}/check-runs",
639 installation_id=installation_id,
640 json=payload,
641 )
642 else:
643 payload.pop("name")
644 result = await self._request(
645 "PATCH",
646 f"/repos/{repository}/check-runs/{check_id}",
647 installation_id=installation_id,
648 json=payload,
649 )
650 result_id = result.get("id")
651 if not isinstance(result_id, int):
652 msg = "check-run response omitted its integer ID"
653 raise GitHubError(msg)
654 return result_id
656 async def list_installations(self) -> list[dict[str, Any]]:
657 """List every installation visible to the App JWT."""
658 items: list[dict[str, Any]] = []
659 page = 1
660 while True:
661 response = await self._authenticated_response(
662 "GET",
663 "/app/installations",
664 app_authenticated=True,
665 params={"per_page": 100, "page": page},
666 )
667 if not response.is_success:
668 self._raise_api_error(response, "GET", "/app/installations")
669 values = response.json()
670 if not isinstance(values, list):
671 msg = "expected list response from GET /app/installations"
672 raise GitHubError(msg)
673 items.extend(item for item in values if isinstance(item, dict))
674 if len(values) < 100:
675 return items
676 page += 1
678 async def list_installation_repositories(self, installation_id: int) -> list[dict[str, Any]]:
679 """List repositories granted to one installation."""
680 result: list[dict[str, Any]] = []
681 page = 1
682 while True:
683 response = await self._request(
684 "GET",
685 "/installation/repositories",
686 installation_id=installation_id,
687 params={"per_page": 100, "page": page},
688 )
689 repositories = response.get("repositories")
690 if not isinstance(repositories, list):
691 msg = "installation repositories response omitted repositories"
692 raise GitHubError(msg)
693 result.extend(item for item in repositories if isinstance(item, dict))
694 if len(repositories) < 100:
695 return result
696 page += 1
698 async def list_open_pulls(self, installation_id: int, repository: str) -> list[dict[str, Any]]:
699 """List open pull requests for reconciliation."""
700 return await self._get_list(
701 f"/repos/{repository}/pulls",
702 installation_id,
703 params={"state": "open", "sort": "updated", "direction": "desc"},
704 )
706 async def list_commit_pulls(
707 self, installation_id: int, repository: str, head_sha: str
708 ) -> list[dict[str, Any]]:
709 """List pull requests associated with a commit for shared-head detection."""
710 return await self._get_list(
711 f"/repos/{repository}/commits/{head_sha}/pulls",
712 installation_id,
713 max_items=100,
714 )