""" Registry for GitHub event handlers. """ import dataclasses import logging import sys import threading import time import typing as t from urllib.parse import parse_qs, urlencode import deprecated import requests import urllib3 from . import __version__ from .token import InstallationTokenSupplier, JwtSupplier, TokenInfo from .utils.functions import coalesce T = t.TypeVar("T") logger = logging.getLogger(__name__) user_agent = f"python/{sys.version.split()[0]} github-bot-api/{__version__}" if t.TYPE_CHECKING: import github @dataclasses.dataclass class GithubClientSettings: """ Settings for constructing a #github.Github client object. """ base_url: t.Optional[str] = None user_agent: t.Optional[str] = None timeout: t.Optional[int] = None per_page: t.Optional[int] = None verify: t.Optional[bool] = None retry: t.Optional[urllib3.Retry] = None def update(self, other: "GithubClientSettings") -> "GithubClientSettings": result = GithubClientSettings() for field in dataclasses.fields(self): value = getattr(other, field.name) if value is None: value = getattr(self, field.name) setattr(result, field.name, value) return result def make_client(self, login_or_token: t.Optional[str] = None, jwt: t.Optional[str] = None) -> "github.Github": import github import github.Consts return github.Github( login_or_token=login_or_token, jwt=jwt, base_url=self.base_url or github.Consts.DEFAULT_BASE_URL, user_agent=self.user_agent or "PyGithub/Python", timeout=coalesce(self.timeout, github.Consts.DEFAULT_TIMEOUT), per_page=coalesce(self.per_page, github.Consts.DEFAULT_PER_PAGE), verify=coalesce(self.verify, True), retry=self.retry, ) @dataclasses.dataclass class GithubApp: """ Represents a GitHub application and all the required details. """ PUBLIC_GITHUB_V3_API_URL = "https://api.github.com" user_agent: str """User agent of the application. This will be respected in #get_user_agent().""" app_id: int """GitHub Application ID.""" private_key: str = dataclasses.field(repr=False) """RSA private key to sign the JWT with.""" client_id: t.Optional[str] = None """The GitHub App's OAuth client ID. This is required for OAuth2 authorization URL generation. Can be omitted if the app does not use OAuth2.""" client_secret: t.Optional[str] = dataclasses.field(default=None, repr=False) """The GitHub App's OAuth client secret. This is required for OAuth2 authorization URL generation. Can be omitted if the app does not use OAuth2. Note that this must be specified if #client_id is specified.""" redirect_uri: t.Optional[str] = None """The GitHub App's OAuth redirect URI. This is required for OAuth2 authorization URL generation. Can be omitted if the app does not use OAuth2. This field is optional, but required for the web authorization flow with #oauth2_web_application_flow_url().""" v3_api_url: str = PUBLIC_GITHUB_V3_API_URL """GitHub API base URL. Defaults to the public GitHub API.""" def __post_init__(self): self._jwt_supplier = JwtSupplier(self.app_id, self.private_key) self._lock = threading.Lock() self._installation_tokens: t.Dict[int, InstallationTokenSupplier] = {} if self.client_id is not None: if self.client_secret is None: raise ValueError("client_secret must be specified if client_id is specified.") if self.redirect_uri is not None: if self.client_id is None: raise ValueError("redirect_uri does not make sense without client_id.") def _get_base_github_client_settings(self) -> GithubClientSettings: return GithubClientSettings(self.v3_api_url, self.get_user_agent()) def get_user_agent(self, installation_id: t.Optional[int] = None) -> str: """ Create a user agent string for the PyGithub client, including the installation if specified. """ user_agent = f"{self.user_agent} PyGithub/python (app_id={self.app_id}" if installation_id: user_agent += f", installation_id={installation_id})" return user_agent @property def jwt(self) -> TokenInfo: """ Returns the JWT for your GitHub application. The JWT is the token to use with GitHub application APIs. """ return self._jwt_supplier() @property def jwt_supplier(self) -> JwtSupplier: """ Returns a new #JwtSupplier that is used for generating JWT tokens for your GitHub application. """ return JwtSupplier(self.app_id, self.private_key) def app_client(self, settings: t.Union[GithubClientSettings, t.Dict[str, t.Any], None] = None) -> "github.Github": """ Returns a PyGithub client for your GitHub application. Note that the client's token will expire after 10 minutes and you will have to create a new client or update the client's token with the value returned by #jwt. It is recommended that you create a new client for each atomic operation you perform. This requires you to install `PyGithub>=1.58`. """ if isinstance(settings, dict): settings = GithubClientSettings(**settings) elif settings is None: settings = GithubClientSettings() settings = self._get_base_github_client_settings().update(settings) return settings.make_client(jwt=self.jwt.value) def __requestor(self, auth_header: str, installation_id: int) -> t.Dict[str, str]: return requests.post( self.v3_api_url.rstrip("/") + f"/app/installations/{installation_id}/access_tokens", headers={"Authorization": auth_header, "User-Agent": user_agent}, ).json() @deprecated.deprecated(reason="Use .installation_token_supplier() instead.", version="0.8.0") def get_installation_token_supplier(self, installation_id: int) -> InstallationTokenSupplier: return self.installation_token_supplier(installation_id) def installation_token_supplier(self, installation_id: int) -> InstallationTokenSupplier: """ Create an #InstallationTokenSupplier for your GitHub application to act within the scope of the given *installation_id*. """ with self._lock: return self._installation_tokens.setdefault( installation_id, InstallationTokenSupplier( self._jwt_supplier, installation_id, self.__requestor, ), ) def installation_token(self, installation_id: int) -> TokenInfo: """ A short-hand to retrieve a new installation token for the given *installation_id*. """ return self.get_installation_token_supplier(installation_id)() def installation_client( self, installation_id: int, settings: t.Union[GithubClientSettings, t.Dict[str, t.Any], None] = None, ) -> "github.Github": """ Returns a PyGithub client for your GitHub application to act in the scope of the given *installation_id*. Note that the client's token will expire after 10 minutes and you will have to create a new client or update the client's token with the value returned by #jwt. It is recommended that you create a new client for each atomic operation you perform. This requires you to install `PyGithub>=1.58`. """ if isinstance(settings, dict): settings = GithubClientSettings(**settings) elif settings is None: settings = GithubClientSettings() token = self.installation_token(installation_id).value settings = self._get_base_github_client_settings().update(settings) return settings.make_client(login_or_token=token) def oauth2_web_application_flow_url(self, state: t.Optional[str] = None) -> str: """ Returns the URL for a user to begin the OAuth2 web authorization flow. Preconditions: - You must have provided a `client_id` when constructing the #GithubApp instance. Documentation: https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app#using-the-web-application-flow-to-generate-a-user-access-token """ if self.client_id is None: raise ValueError("client_id must be specified to generate OAuth2 authorization URL.") if self.client_secret is None: raise ValueError("client_secret must be specified to generate OAuth2 authorization URL.") if not self.redirect_uri: raise ValueError("redirect_uri must be specified to generate OAuth2 authorization URL.") params = { "client_id": self.client_id, "redirect_uri": self.redirect_uri, } if state is not None: params["state"] = state url = self.v3_api_url.replace("api.", "").replace("/api/v3", "").rstrip("/") return f"{url}/login/oauth/authorize?" + urlencode(params) def oauth2_device_flow(self) -> "OAuth2DeviceCodeFlow": """ Makes a request to GitHub to request a device code for the OAuth2 device flow. Prerequisites: - "Enable Device Flow" must be checked in your GitHub app's settings. - You must have provided a `client_id` when constructing the #GithubApp instance. Documentation: https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app#using-the-device-flow-to-generate-a-user-access-token """ params = {"client_id": self.client_id} url = self.v3_api_url.replace("api.", "").replace("/api/v3", "").rstrip("/") response = requests.post(f"{url}/login/device/code", params=params) response.raise_for_status() payload = {k: v[0] for k, v in parse_qs(response.text).items()} return OAuth2DeviceCodeFlow( device_code=payload["device_code"], user_code=payload["user_code"], verification_uri=payload["verification_uri"], expires_in=int(payload["expires_in"]), interval=int(payload["interval"]), app=self, ) def oauth2_access_token( self, *, code: t.Optional[str] = None, device_code: t.Optional[str] = None ) -> t.Optional["OAuth2TokenInfo"]: """ Makes a request to GitHub to exchange an OAuth2 code for an access token. Important: You must provide the correct `code` or `device_code` parameter, but not both. Documentation: https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app#generating-a-user-access-token-when-a-user-installs-your-app Returns None if the token is not yet available (in the case of the device flow). Otherwise, returns a #TokenInfo object. If an error occurs, an #AccessTokenError is raised. """ params = {"client_id": self.client_id, "client_secret": self.client_secret} if code is not None: params["code"] = code elif device_code is not None: params["device_code"] = device_code params["grant_type"] = "urn:ietf:params:oauth:grant-type:device_code" else: raise ValueError("You must provide either a code or a device_code.") url = self.v3_api_url.replace("api.", "").replace("/api/v3", "").rstrip("/") response = requests.post(f"{url}/login/oauth/access_token", params=params) payload = {k: v[0] for k, v in parse_qs(response.text).items()} if (error := payload.get("error")) == "authorization_pending": return None elif error is not None: return OAuth2TokenInfo( access_token=payload["access_token"], expires_in=int(payload["expires_in"]), scope=payload.get("scope"), token_type=payload["token_type"], refresh_token=payload.get("refresh_token"), refresh_token_expires_in=int(payload["refresh_token_expires_in"]) if "refresh_token_expires_in" in payload else None, ) else: raise AccessTokenError(error) # type: ignore[arg-type] @dataclasses.dataclass class OAuth2DeviceCodeFlow: device_code: str user_code: str verification_uri: str expires_in: int interval: int # If you want to use this object to poll for the token, you need the GitHub App. app: t.Optional[GithubApp] = None def wait_for_token( self, aborted: t.Optional[threading.Event] = None, max_duration: t.Optional[float] = None ) -> "OAuth2TokenInfo": """ Polls GitHub for the access token until it is available. If you pass in a threading.Event object, it will be used to abort the polling when the event is set. If *max_duration* is specified, the polling will stop after that many seconds and raise a #TimeoutError if the token is not available by then. """ assert self.app is not None, "You must set the 'app' attribute to use this method." tstart = time.perf_counter() while True: if max_duration is not None and time.perf_counter() - tstart > max_duration: raise TimeoutError("Timed out waiting for token.") if aborted and aborted.is_set(): raise RuntimeError("Polling for token was aborted.") try: if token := self.app.oauth2_access_token(device_code=self.device_code): return token except AccessTokenError as e: if e.error == "slow_down": time.sleep(5) else: raise time.sleep(self.interval) @dataclasses.dataclass class OAuth2TokenInfo: access_token: str expires_in: int token_type: str scope: t.Optional[str] = None refresh_token: t.Optional[str] = None refresh_token_expires_in: t.Optional[int] = None @property def auth_header(self) -> str: return f"{self.token_type} {self.access_token}" @dataclasses.dataclass class AccessTokenError(Exception): error: t.Literal[ "slow_down", "expired_token", "unsupported_grant_type", "incorrect_client_credentials", "incorrect_device_code", "access_denied", "device_flow_disabled", ]