| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| import abc |
| import base64 |
| import time |
| from abc import ABC |
| from datetime import datetime, timedelta, timezone |
| from typing import TYPE_CHECKING, Callable, Dict, Optional, Union |
|
|
| import jwt |
| from requests import utils |
|
|
| from github import Consts |
| from github.InstallationAuthorization import InstallationAuthorization |
| from github.Requester import Requester, WithRequester |
|
|
| if TYPE_CHECKING: |
| from github.GithubIntegration import GithubIntegration |
|
|
| PrivateKeyGenerator = Callable[[], Union[str, bytes]] |
| DictSignFunction = Callable[[dict], Union[str, bytes]] |
|
|
| |
| ACCESS_TOKEN_REFRESH_THRESHOLD_SECONDS = 20 |
| TOKEN_REFRESH_THRESHOLD_TIMEDELTA = timedelta(seconds=ACCESS_TOKEN_REFRESH_THRESHOLD_SECONDS) |
|
|
|
|
| |
| class Auth(abc.ABC): |
| """ |
| This class is the base class of all authentication methods for Requester. |
| """ |
|
|
| @property |
| @abc.abstractmethod |
| def token_type(self) -> str: |
| """ |
| The type of the auth token as used in the HTTP Authorization header, e.g. Bearer or Basic. |
| |
| :return: token type |
| |
| """ |
|
|
| @property |
| @abc.abstractmethod |
| def token(self) -> str: |
| """ |
| The auth token as used in the HTTP Authorization header. |
| |
| :return: token |
| |
| """ |
|
|
| def authentication(self, headers: dict) -> None: |
| """ |
| Add authorization to the headers. |
| """ |
| headers["Authorization"] = f"{self.token_type} {self.token}" |
|
|
| def mask_authentication(self, headers: dict) -> None: |
| """ |
| Mask header, e.g. for logging. |
| """ |
| headers["Authorization"] = self._masked_token |
|
|
| @property |
| def _masked_token(self) -> str: |
| return "(unknown auth removed)" |
|
|
|
|
| class HTTPBasicAuth(Auth, abc.ABC): |
| @property |
| @abc.abstractmethod |
| def username(self) -> str: |
| """ |
| The username. |
| """ |
|
|
| @property |
| @abc.abstractmethod |
| def password(self) -> str: |
| """ |
| The password. |
| """ |
|
|
| @property |
| def token_type(self) -> str: |
| return "Basic" |
|
|
| @property |
| def token(self) -> str: |
| return base64.b64encode(f"{self.username}:{self.password}".encode()).decode("utf-8").replace("\n", "") |
|
|
| @property |
| def _masked_token(self) -> str: |
| return "Basic (login and password removed)" |
|
|
|
|
| class Login(HTTPBasicAuth): |
| """ |
| This class is used to authenticate with login and password. |
| """ |
|
|
| def __init__(self, login: str, password: str): |
| assert isinstance(login, str) |
| assert len(login) > 0 |
| assert isinstance(password, str) |
| assert len(password) > 0 |
|
|
| self._login = login |
| self._password = password |
|
|
| @property |
| def login(self) -> str: |
| return self._login |
|
|
| @property |
| def username(self) -> str: |
| return self.login |
|
|
| @property |
| def password(self) -> str: |
| return self._password |
|
|
|
|
| class Token(Auth): |
| """ |
| This class is used to authenticate with a single constant token. |
| """ |
|
|
| def __init__(self, token: str): |
| assert isinstance(token, str) |
| assert len(token) > 0 |
| self._token = token |
|
|
| @property |
| def token_type(self) -> str: |
| return "token" |
|
|
| @property |
| def token(self) -> str: |
| return self._token |
|
|
| @property |
| def _masked_token(self) -> str: |
| return "token (oauth token removed)" |
|
|
|
|
| class JWT(Auth, ABC): |
| """ |
| This class is the base class to authenticate with a JSON Web Token (JWT). |
| |
| https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-json-web-token-jwt-for-a-github-app |
| |
| """ |
|
|
| @property |
| def token_type(self) -> str: |
| return "Bearer" |
|
|
|
|
| class JwtSigner: |
| def __init__(self, private_key_or_func: Union[str, PrivateKeyGenerator], jwt_algorithm: str): |
| self._private_key_or_func = private_key_or_func |
| self._jwt_algorithm = jwt_algorithm |
|
|
| def jwt_sign(self, payload: dict) -> Union[str, bytes]: |
| if callable(self._private_key_or_func): |
| private_key = self._private_key_or_func() |
| else: |
| private_key = self._private_key_or_func |
| return jwt.encode(payload, key=private_key, algorithm=self._jwt_algorithm) |
|
|
|
|
| class AppAuth(JWT): |
| """ |
| This class is used to authenticate as a GitHub App. |
| |
| https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-as-a-github-app |
| |
| """ |
|
|
| @staticmethod |
| def create_jwt_sign(private_key_or_func: Union[str, PrivateKeyGenerator], jwt_algorithm: str) -> DictSignFunction: |
| return JwtSigner(private_key_or_func, jwt_algorithm).jwt_sign |
|
|
| |
| def __init__( |
| self, |
| app_id: Union[int, str], |
| private_key: Optional[Union[str, PrivateKeyGenerator]] = None, |
| *, |
| sign_func: Optional[DictSignFunction] = None, |
| jwt_expiry: int = Consts.DEFAULT_JWT_EXPIRY, |
| jwt_issued_at: int = Consts.DEFAULT_JWT_ISSUED_AT, |
| ): |
| assert isinstance(app_id, (int, str)), app_id |
| if isinstance(app_id, str): |
| assert len(app_id) > 0, "app_id must not be empty" |
| assert private_key is not None or sign_func is not None, "either private_key or sign_func must be given" |
| assert private_key is None or sign_func is None, "private_key or sign_func cannot both be given" |
| if private_key is not None: |
| assert isinstance(private_key, str) or callable(private_key) |
| if isinstance(private_key, str): |
| assert len(private_key) > 0, "private_key must not be empty" |
| sign_func = AppAuth.create_jwt_sign(private_key, Consts.DEFAULT_JWT_ALGORITHM) |
| assert isinstance(jwt_expiry, int), jwt_expiry |
| assert Consts.MIN_JWT_EXPIRY <= jwt_expiry <= Consts.MAX_JWT_EXPIRY, jwt_expiry |
|
|
| self._app_id = app_id |
| self._sign_func = sign_func |
| self._jwt_expiry = jwt_expiry |
| self._jwt_issued_at = jwt_issued_at |
|
|
| @property |
| def app_id(self) -> Union[int, str]: |
| return self._app_id |
|
|
| @property |
| def token(self) -> str: |
| return self.create_jwt() |
|
|
| def get_installation_auth( |
| self, |
| installation_id: int, |
| token_permissions: Optional[Dict[str, str]] = None, |
| requester: Optional[Requester] = None, |
| ) -> "AppInstallationAuth": |
| """ |
| Creates a github.Auth.AppInstallationAuth instance for an installation. |
| |
| :param installation_id: installation id |
| :param token_permissions: optional permissions |
| :param requester: optional requester with app authentication |
| :return: |
| |
| """ |
| return AppInstallationAuth(self, installation_id, token_permissions, requester) |
|
|
| def create_jwt(self, expiration: Optional[int] = None) -> str: |
| """ |
| Create a signed JWT |
| https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-a-github-app |
| |
| :return string: jwt |
| """ |
| if expiration is not None: |
| assert isinstance(expiration, int), expiration |
| assert Consts.MIN_JWT_EXPIRY <= expiration <= Consts.MAX_JWT_EXPIRY, expiration |
|
|
| now = int(time.time()) |
| payload = { |
| "iat": now + self._jwt_issued_at, |
| "exp": now + (expiration if expiration is not None else self._jwt_expiry), |
| "iss": self._app_id, |
| } |
| assert self._sign_func is not None |
| encrypted = self._sign_func(payload) |
|
|
| if isinstance(encrypted, bytes): |
| return encrypted.decode("utf-8") |
| return encrypted |
|
|
|
|
| class AppAuthToken(JWT): |
| """ |
| This class is used to authenticate as a GitHub App with a single constant JWT. |
| |
| https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-as-a-github-app |
| |
| """ |
|
|
| def __init__(self, token: str): |
| assert isinstance(token, str) |
| assert len(token) > 0 |
| self._token = token |
|
|
| @property |
| def token(self) -> str: |
| return self._token |
|
|
|
|
| class AppInstallationAuth(Auth, WithRequester["AppInstallationAuth"]): |
| """ |
| This class is used to authenticate as a GitHub App Installation. |
| |
| https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-as-a-github-app-installation |
| |
| """ |
|
|
| |
| __integration: Optional["GithubIntegration"] = None |
| __installation_authorization: Optional[InstallationAuthorization] = None |
|
|
| def __init__( |
| self, |
| app_auth: AppAuth, |
| installation_id: int, |
| token_permissions: Optional[Dict[str, str]] = None, |
| requester: Optional[Requester] = None, |
| ): |
| super().__init__() |
|
|
| assert isinstance(app_auth, AppAuth), app_auth |
| assert isinstance(installation_id, int), installation_id |
| assert token_permissions is None or isinstance(token_permissions, dict), token_permissions |
| assert requester is None or isinstance(requester, Requester), requester |
|
|
| self._app_auth = app_auth |
| self._installation_id = installation_id |
| self._token_permissions = token_permissions |
|
|
| if requester is not None: |
| self.withRequester(requester) |
|
|
| def withRequester(self, requester: Requester) -> "AppInstallationAuth": |
| assert isinstance(requester, Requester), requester |
| super().withRequester(requester.withAuth(self._app_auth)) |
|
|
| |
| from github.GithubIntegration import GithubIntegration |
|
|
| self.__integration = GithubIntegration(**self.requester.kwargs) |
|
|
| return self |
|
|
| @property |
| def app_id(self) -> Union[int, str]: |
| return self._app_auth.app_id |
|
|
| @property |
| def installation_id(self) -> int: |
| return self._installation_id |
|
|
| @property |
| def token_permissions(self) -> Optional[Dict[str, str]]: |
| return self._token_permissions |
|
|
| @property |
| def token_type(self) -> str: |
| return "token" |
|
|
| @property |
| def token(self) -> str: |
| if self.__installation_authorization is None or self._is_expired: |
| self.__installation_authorization = self._get_installation_authorization() |
| return self.__installation_authorization.token |
|
|
| @property |
| def _is_expired(self) -> bool: |
| assert self.__installation_authorization is not None |
| token_expires_at = self.__installation_authorization.expires_at - TOKEN_REFRESH_THRESHOLD_TIMEDELTA |
| return token_expires_at < datetime.now(timezone.utc) |
|
|
| def _get_installation_authorization(self) -> InstallationAuthorization: |
| assert self.__integration is not None, "Method withRequester(Requester) must be called first" |
| return self.__integration.get_access_token( |
| self._installation_id, |
| permissions=self._token_permissions, |
| ) |
|
|
| @property |
| def _masked_token(self) -> str: |
| return "token (oauth token removed)" |
|
|
|
|
| class AppUserAuth(Auth, WithRequester["AppUserAuth"]): |
| """ |
| This class is used to authenticate as a GitHub App on behalf of a user. |
| |
| https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-with-a-github-app-on-behalf-of-a-user |
| |
| """ |
|
|
| _client_id: str |
| _client_secret: str |
| _token: str |
| _type: str |
| _scope: Optional[str] |
| _expires_at: Optional[datetime] |
| _refresh_token: Optional[str] |
| _refresh_expires_at: Optional[datetime] |
|
|
| |
| from github.ApplicationOAuth import ApplicationOAuth |
|
|
| __app: ApplicationOAuth |
|
|
| def __init__( |
| self, |
| client_id: str, |
| client_secret: str, |
| token: str, |
| token_type: Optional[str] = None, |
| expires_at: Optional[datetime] = None, |
| refresh_token: Optional[str] = None, |
| refresh_expires_at: Optional[datetime] = None, |
| requester: Optional[Requester] = None, |
| ) -> None: |
| super().__init__() |
|
|
| assert isinstance(client_id, str) and len(client_id) > 0 |
| assert isinstance(client_secret, str) and len(client_secret) > 0 |
| assert isinstance(token, str) and len(token) > 0 |
| assert token_type is None or isinstance(token_type, str) and len(token_type) > 0, token_type |
| assert expires_at is None or isinstance(expires_at, datetime), expires_at |
| assert refresh_token is None or isinstance(refresh_token, str) and len(refresh_token) > 0 |
| assert refresh_expires_at is None or isinstance(refresh_expires_at, datetime), refresh_expires_at |
| assert requester is None or isinstance(requester, Requester), requester |
|
|
| self._client_id = client_id |
| self._client_secret = client_secret |
| self._token = token |
| self._type = token_type or "bearer" |
| self._expires_at = expires_at |
| self._refresh_token = refresh_token |
| self._refresh_expires_at = refresh_expires_at |
|
|
| if requester is not None: |
| self.withRequester(requester) |
|
|
| @property |
| def token_type(self) -> str: |
| return self._type |
|
|
| @property |
| def token(self) -> str: |
| if self._is_expired: |
| self._refresh() |
| return self._token |
|
|
| def withRequester(self, requester: Requester) -> "AppUserAuth": |
| assert isinstance(requester, Requester), requester |
| super().withRequester(requester.withAuth(None)) |
|
|
| |
| from github.ApplicationOAuth import ApplicationOAuth |
|
|
| self.__app = ApplicationOAuth( |
| |
| super().requester, |
| headers={}, |
| attributes={ |
| "client_id": self._client_id, |
| "client_secret": self._client_secret, |
| }, |
| ) |
|
|
| return self |
|
|
| @property |
| def _is_expired(self) -> bool: |
| return self._expires_at is not None and self._expires_at < datetime.now(timezone.utc) |
|
|
| def _refresh(self) -> None: |
| if self._refresh_token is None: |
| raise RuntimeError("Cannot refresh expired token because no refresh token has been provided") |
| if self._refresh_expires_at is not None and self._refresh_expires_at < datetime.now(timezone.utc): |
| raise RuntimeError("Cannot refresh expired token because refresh token also expired") |
|
|
| |
| token = self.__app.refresh_access_token(self._refresh_token) |
|
|
| |
| self._token = token.token |
| self._type = token.type |
| self._scope = token.scope |
| self._expires_at = token.expires_at |
| self._refresh_token = token.refresh_token |
| self._refresh_expires_at = token.refresh_expires_at |
|
|
| @property |
| def expires_at(self) -> Optional[datetime]: |
| return self._expires_at |
|
|
| @property |
| def refresh_token(self) -> Optional[str]: |
| return self._refresh_token |
|
|
| @property |
| def refresh_expires_at(self) -> Optional[datetime]: |
| return self._refresh_expires_at |
|
|
| @property |
| def _masked_token(self) -> str: |
| return "Bearer (jwt removed)" |
|
|
|
|
| class NetrcAuth(HTTPBasicAuth, WithRequester["NetrcAuth"]): |
| """ |
| This class is used to authenticate via .netrc. |
| """ |
|
|
| def __init__(self) -> None: |
| super().__init__() |
|
|
| self._login: Optional[str] = None |
| self._password: Optional[str] = None |
|
|
| @property |
| def username(self) -> str: |
| return self.login |
|
|
| @property |
| def login(self) -> str: |
| assert self._login is not None, "Method withRequester(Requester) must be called first" |
| return self._login |
|
|
| @property |
| def password(self) -> str: |
| assert self._password is not None, "Method withRequester(Requester) must be called first" |
| return self._password |
|
|
| def withRequester(self, requester: Requester) -> "NetrcAuth": |
| assert isinstance(requester, Requester), requester |
| super().withRequester(requester) |
|
|
| auth = utils.get_netrc_auth(requester.base_url, raise_errors=True) |
| if auth is None: |
| raise RuntimeError(f"Could not get credentials from netrc for host {requester.hostname}") |
|
|
| self._login, self._password = auth |
|
|
| return self |
|
|
