| 10 | |
| 11 | |
| 12 | class GitHubAppAuth: |
| 13 | def __init__(self, app_id: str, private_key: str, installation_id: str): |
| 14 | self.app_id = app_id |
| 15 | self.private_key = private_key |
| 16 | self.installation_id = installation_id |
| 17 | self._token: str | None = None |
| 18 | self._token_expires_at: float = 0 |
| 19 | self._session: aiohttp.ClientSession | None = None |
| 20 | |
| 21 | def _generate_jwt(self) -> str: |
| 22 | now = int(time.time()) |
| 23 | logger.debug(f"Generating JWT at timestamp: {now}") |
| 24 | payload = {"iat": now - 30, "exp": now + 60, "iss": self.app_id} |
| 25 | return jwt.encode(payload, self.private_key, algorithm="RS256") |
| 26 | |
| 27 | async def _get_session(self) -> aiohttp.ClientSession: |
| 28 | if self._session is None or self._session.closed: |
| 29 | self._session = aiohttp.ClientSession() |
| 30 | return self._session |
| 31 | |
| 32 | async def close(self): |
| 33 | if self._session and not self._session.closed: |
| 34 | await self._session.close() |
| 35 | self._session = None |
| 36 | |
| 37 | async def get_token(self) -> str | None: |
| 38 | if self._token and time.time() < (self._token_expires_at - TOKEN_REFRESH_BUFFER): |
| 39 | return self._token |
| 40 | |
| 41 | logger.info("Fetching new GitHub App installation token...") |
| 42 | |
| 43 | try: |
| 44 | jwt_token = self._generate_jwt() |
| 45 | session = await self._get_session() |
| 46 | |
| 47 | url = f"https://api.github.com/app/installations/{self.installation_id}/access_tokens" |
| 48 | headers = { |
| 49 | "Authorization": f"Bearer {jwt_token}", |
| 50 | "Accept": "application/vnd.github+json", |
| 51 | "X-GitHub-Api-Version": "2022-11-28", |
| 52 | } |
| 53 | |
| 54 | async with session.post(url, headers=headers) as response: |
| 55 | if response.status == 201: |
| 56 | data = await response.json() |
| 57 | self._token = data["token"] |
| 58 | self._token_expires_at = time.time() + 3600 |
| 59 | logger.info("GitHub App token refreshed successfully") |
| 60 | return self._token |
| 61 | else: |
| 62 | error = await response.text() |
| 63 | logger.error(f"Failed to get installation token: {response.status} - {error}") |
| 64 | return None |
| 65 | |
| 66 | except Exception as e: |
| 67 | logger.error(f"Error getting GitHub App token: {e}") |
| 68 | return None |
| 69 | |