mirror of
https://github.com/status-im/PyGithub.git
synced 2026-09-02 11:51:10 +00:00
Support full GitHub app authentication (#1986)
* Support full GitHub app authentication Refactor GithubIntegration class and add test case for app authentication Add permissions and repository properties in InstallationAuthorization Set JWT_EXPIRY=60 by default in GithubIntegration constructor * Modify existing testcases for GithubIntegration as per the framework and add missing tests * Provide installation ID for creating the access token instead of getting the first installation * Add optional permissions support for installation access token * Add lock around app authentication * Keep compatibility for importing GithubIntegration from MainClass * Group app authentication parameters in a class Co-authored-by: Malik Ammar Akbar <malikammar.akbar@pfizer.com> Co-authored-by: Enrico Minack <github@enrico.minack.dev>
This commit is contained in:
co-authored by
Malik Ammar Akbar
Enrico Minack
parent
7cf3dfc18e
commit
5e27c10a31
@@ -0,0 +1,41 @@
|
||||
############################ Copyrights and license ############################
|
||||
# #
|
||||
# Copyright 2023 Denis Blanchette <denisblanchette@gmail.com> #
|
||||
# #
|
||||
# This file is part of PyGithub. #
|
||||
# http://pygithub.readthedocs.io/ #
|
||||
# #
|
||||
# PyGithub is free software: you can redistribute it and/or modify it under #
|
||||
# the terms of the GNU Lesser General Public License as published by the Free #
|
||||
# Software Foundation, either version 3 of the License, or (at your option) #
|
||||
# any later version. #
|
||||
# #
|
||||
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
|
||||
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
|
||||
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
|
||||
# details. #
|
||||
# #
|
||||
# You should have received a copy of the GNU Lesser General Public License #
|
||||
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
|
||||
# #
|
||||
################################################################################
|
||||
|
||||
|
||||
class AppAuthentication:
|
||||
def __init__(
|
||||
self,
|
||||
app_id,
|
||||
private_key,
|
||||
installation_id,
|
||||
token_permissions=None,
|
||||
):
|
||||
assert isinstance(app_id, (int, str)), app_id
|
||||
assert isinstance(private_key, str)
|
||||
assert isinstance(installation_id, int), installation_id
|
||||
assert token_permissions is None or isinstance(
|
||||
token_permissions, dict
|
||||
), token_permissions
|
||||
self.app_id = app_id
|
||||
self.private_key = private_key
|
||||
self.installation_id = installation_id
|
||||
self.token_permissions = token_permissions
|
||||
@@ -0,0 +1,10 @@
|
||||
from typing import Optional, Dict, Union
|
||||
|
||||
class AppAuthentication:
|
||||
def __init__(
|
||||
self,
|
||||
app_id: Union[int, str],
|
||||
private_key: str,
|
||||
installation_id: int,
|
||||
token_permissions: Optional[Dict[str, str]] = ...,
|
||||
): ...
|
||||
@@ -131,3 +131,20 @@ deploymentStatusEnhancementsPreview = "application/vnd.github.flash-preview+json
|
||||
|
||||
# https://developer.github.com/changes/2019-12-03-internal-visibility-changes/
|
||||
repoVisibilityPreview = "application/vnd.github.nebula-preview+json"
|
||||
|
||||
DEFAULT_BASE_URL = "https://api.github.com"
|
||||
DEFAULT_STATUS_URL = "https://status.github.com"
|
||||
# As of 2018-05-17, Github imposes a 10s limit for completion of API requests.
|
||||
# Thus, the timeout should be slightly > 10s to account for network/front-end
|
||||
# latency.
|
||||
DEFAULT_TIMEOUT = 15
|
||||
DEFAULT_PER_PAGE = 30
|
||||
|
||||
# JWT expiry in seconds. Could be set for max 600 seconds (10 minutes).
|
||||
# https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-a-github-app
|
||||
DEFAULT_JWT_EXPIRY = 300
|
||||
MIN_JWT_EXPIRY = 15
|
||||
MAX_JWT_EXPIRY = 600
|
||||
# https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#generating-a-json-web-token-jwt
|
||||
# "The time the JWT was created. To protect against clock drift, we recommend you set this 60 seconds in the past."
|
||||
DEFAULT_JWT_ISSUED_AT = -60
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
import time
|
||||
|
||||
import deprecated
|
||||
import jwt
|
||||
|
||||
from github import Consts
|
||||
from github.GithubException import GithubException
|
||||
from github.Installation import Installation
|
||||
from github.InstallationAuthorization import InstallationAuthorization
|
||||
from github.PaginatedList import PaginatedList
|
||||
from github.Requester import Requester
|
||||
|
||||
|
||||
class GithubIntegration:
|
||||
"""
|
||||
Main class to obtain tokens for a GitHub integration.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
integration_id,
|
||||
private_key,
|
||||
base_url=Consts.DEFAULT_BASE_URL,
|
||||
jwt_expiry=Consts.DEFAULT_JWT_EXPIRY,
|
||||
jwt_issued_at=Consts.DEFAULT_JWT_ISSUED_AT,
|
||||
):
|
||||
"""
|
||||
:param integration_id: int
|
||||
:param private_key: string
|
||||
:param base_url: string
|
||||
:param jwt_expiry: int. Expiry of the JWT used to get the information about this integration.
|
||||
The default expiration is in 5 minutes and is capped at 10 minutes according to GitHub documentation
|
||||
https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#generating-a-json-web-token-jwt
|
||||
:param jwt_issued_at: int. Number of seconds, relative to now, to set for the "iat" (issued at) parameter.
|
||||
The default value is -60 to protect against clock drift
|
||||
"""
|
||||
assert isinstance(integration_id, (int, str)), integration_id
|
||||
assert isinstance(private_key, str), "supplied private key should be a string"
|
||||
assert isinstance(base_url, str), base_url
|
||||
assert isinstance(jwt_expiry, int), jwt_expiry
|
||||
assert Consts.MIN_JWT_EXPIRY <= jwt_expiry <= Consts.MAX_JWT_EXPIRY, jwt_expiry
|
||||
assert isinstance(jwt_issued_at, int)
|
||||
|
||||
self.base_url = base_url
|
||||
self.integration_id = integration_id
|
||||
self.private_key = private_key
|
||||
self.jwt_expiry = jwt_expiry
|
||||
self.jwt_issued_at = jwt_issued_at
|
||||
self.__requester = Requester(
|
||||
login_or_token=None,
|
||||
password=None,
|
||||
jwt=self.create_jwt(),
|
||||
app_auth=None,
|
||||
base_url=self.base_url,
|
||||
timeout=Consts.DEFAULT_TIMEOUT,
|
||||
user_agent="PyGithub/Python",
|
||||
per_page=Consts.DEFAULT_PER_PAGE,
|
||||
verify=True,
|
||||
retry=None,
|
||||
pool_size=None,
|
||||
)
|
||||
|
||||
def _get_headers(self):
|
||||
"""
|
||||
Get headers for the requests.
|
||||
|
||||
:return: dict
|
||||
"""
|
||||
return {
|
||||
"Authorization": f"Bearer {self.create_jwt()}",
|
||||
"Accept": Consts.mediaTypeIntegrationPreview,
|
||||
"User-Agent": "PyGithub/Python",
|
||||
}
|
||||
|
||||
def _get_installed_app(self, url):
|
||||
"""
|
||||
Get installation for the given URL.
|
||||
|
||||
:param url: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
headers, response = self.__requester.requestJsonAndCheck(
|
||||
"GET", url, headers=self._get_headers()
|
||||
)
|
||||
|
||||
return Installation(
|
||||
requester=self.__requester,
|
||||
headers=headers,
|
||||
attributes=response,
|
||||
completed=True,
|
||||
)
|
||||
|
||||
def create_jwt(self):
|
||||
"""
|
||||
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:
|
||||
"""
|
||||
now = int(time.time())
|
||||
payload = {
|
||||
"iat": now + self.jwt_issued_at,
|
||||
"exp": now + self.jwt_expiry,
|
||||
"iss": self.integration_id,
|
||||
}
|
||||
encrypted = jwt.encode(payload, key=self.private_key, algorithm="RS256")
|
||||
|
||||
if isinstance(encrypted, bytes):
|
||||
encrypted = encrypted.decode("utf-8")
|
||||
|
||||
return encrypted
|
||||
|
||||
def get_access_token(self, installation_id, permissions=None):
|
||||
"""
|
||||
:calls: `POST /app/installations/{installation_id}/access_tokens <https://docs.github.com/en/rest/apps/apps#create-an-installation-access-token-for-an-app>`
|
||||
:param installation_id: int
|
||||
:param permissions: dict
|
||||
:return: :class:`github.InstallationAuthorization.InstallationAuthorization`
|
||||
"""
|
||||
if permissions is None:
|
||||
permissions = {}
|
||||
|
||||
if not isinstance(permissions, dict):
|
||||
raise GithubException(
|
||||
status=400, data={"message": "Invalid permissions"}, headers=None
|
||||
)
|
||||
|
||||
body = {"permissions": permissions}
|
||||
headers, response = self.__requester.requestJsonAndCheck(
|
||||
"POST",
|
||||
f"/app/installations/{installation_id}/access_tokens",
|
||||
input=body,
|
||||
)
|
||||
|
||||
return InstallationAuthorization(
|
||||
requester=self.__requester,
|
||||
headers=headers,
|
||||
attributes=response,
|
||||
completed=True,
|
||||
)
|
||||
|
||||
@deprecated.deprecated("Use get_repo_installation")
|
||||
def get_installation(self, owner, repo):
|
||||
"""
|
||||
Deprecated by get_repo_installation
|
||||
|
||||
:calls: `GET /repos/{owner}/{repo}/installation <https://docs.github.com/en/rest/reference/apps#get-a-repository-installation-for-the-authenticated-app>`
|
||||
:param owner: str
|
||||
:param repo: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
return self._get_installed_app(url=f"/repos/{owner}/{repo}/installation")
|
||||
|
||||
def get_installations(self):
|
||||
"""
|
||||
:calls: GET /app/installations <https://docs.github.com/en/rest/reference/apps#list-installations-for-the-authenticated-app>
|
||||
:rtype: :class:`github.PaginatedList.PaginatedList[github.Installation.Installation]`
|
||||
"""
|
||||
return PaginatedList(
|
||||
contentClass=Installation,
|
||||
requester=self.__requester,
|
||||
firstUrl="/app/installations",
|
||||
firstParams=None,
|
||||
headers=self._get_headers(),
|
||||
list_item="installations",
|
||||
)
|
||||
|
||||
def get_org_installation(self, org):
|
||||
"""
|
||||
:calls: `GET /orgs/{org}/installation <https://docs.github.com/en/rest/apps/apps#get-an-organization-installation-for-the-authenticated-app>`
|
||||
:param org: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
return self._get_installed_app(url=f"/orgs/{org}/installation")
|
||||
|
||||
def get_repo_installation(self, owner, repo):
|
||||
"""
|
||||
:calls: `GET /repos/{owner}/{repo}/installation <https://docs.github.com/en/rest/reference/apps#get-a-repository-installation-for-the-authenticated-app>`
|
||||
:param owner: str
|
||||
:param repo: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
return self._get_installed_app(url=f"/repos/{owner}/{repo}/installation")
|
||||
|
||||
def get_user_installation(self, username):
|
||||
"""
|
||||
:calls: `GET /users/{username}/installation <https://docs.github.com/en/rest/apps/apps#get-a-user-installation-for-the-authenticated-app>`
|
||||
:param username: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
return self._get_installed_app(url=f"/users/{username}/installation")
|
||||
|
||||
def get_app_installation(self, installation_id):
|
||||
"""
|
||||
:calls: `GET /app/installations/{installation_id} <https://docs.github.com/en/rest/apps/apps#get-an-installation-for-the-authenticated-app>`
|
||||
:param installation_id: int
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
return self._get_installed_app(url=f"/app/installations/{installation_id}")
|
||||
@@ -0,0 +1,34 @@
|
||||
from typing import Union, Optional, Dict
|
||||
|
||||
from github.Installation import Installation
|
||||
from github.InstallationAuthorization import InstallationAuthorization
|
||||
from github.PaginatedList import PaginatedList
|
||||
from github.Requester import Requester
|
||||
|
||||
class GithubIntegration:
|
||||
integration_id: Union[int, str] = ...
|
||||
private_key: str = ...
|
||||
base_url: str = ...
|
||||
jwt_expiry: int = ...
|
||||
jwt_issued_at: int = ...
|
||||
__requester: Requester = ...
|
||||
def __init__(
|
||||
self,
|
||||
integration_id: Union[int, str],
|
||||
private_key: str,
|
||||
base_url: str = ...,
|
||||
jwt_expiry: int = ...,
|
||||
jwt_issued_at: int = ...,
|
||||
) -> None: ...
|
||||
def _get_installed_app(self, url: str) -> Installation: ...
|
||||
def _get_headers(self) -> Dict[str, str]: ...
|
||||
def create_jwt(self, expiration: int = ...) -> str: ...
|
||||
def get_access_token(
|
||||
self, installation_id: int, permissions: Optional[Dict[str, str]] = ...
|
||||
) -> InstallationAuthorization: ...
|
||||
def get_app_installation(self, installation_id: int) -> Installation: ...
|
||||
def get_installation(self, owner: str, repo: str) -> Installation: ...
|
||||
def get_installations(self) -> PaginatedList[Installation]: ...
|
||||
def get_org_installation(self, org: str) -> Installation: ...
|
||||
def get_repo_installation(self, owner: str, repo: str) -> Installation: ...
|
||||
def get_user_installation(self, username: str) -> Installation: ...
|
||||
@@ -57,10 +57,26 @@ class InstallationAuthorization(github.GithubObject.NonCompletableGithubObject):
|
||||
"""
|
||||
return self._on_behalf_of.value
|
||||
|
||||
@property
|
||||
def permissions(self):
|
||||
"""
|
||||
:type: dict
|
||||
"""
|
||||
return self._permissions.value
|
||||
|
||||
@property
|
||||
def repository_selection(self):
|
||||
"""
|
||||
:type: string
|
||||
"""
|
||||
return self._repository_selection.value
|
||||
|
||||
def _initAttributes(self):
|
||||
self._token = github.GithubObject.NotSet
|
||||
self._expires_at = github.GithubObject.NotSet
|
||||
self._on_behalf_of = github.GithubObject.NotSet
|
||||
self._permissions = github.GithubObject.NotSet
|
||||
self._repository_selection = github.GithubObject.NotSet
|
||||
|
||||
def _useAttributes(self, attributes):
|
||||
if "token" in attributes: # pragma no branch
|
||||
@@ -71,3 +87,9 @@ class InstallationAuthorization(github.GithubObject.NonCompletableGithubObject):
|
||||
self._on_behalf_of = self._makeClassAttribute(
|
||||
github.NamedUser.NamedUser, attributes["on_behalf_of"]
|
||||
)
|
||||
if "permissions" in attributes: # pragma no branch
|
||||
self._permissions = self._makeDictAttribute(attributes["permissions"])
|
||||
if "repository_selection" in attributes: # pragma no branch
|
||||
self._repository_selection = self._makeStringAttribute(
|
||||
attributes["repository_selection"]
|
||||
)
|
||||
|
||||
@@ -14,3 +14,7 @@ class InstallationAuthorization(NonCompletableGithubObject):
|
||||
def on_behalf_of(self) -> NamedUser: ...
|
||||
@property
|
||||
def token(self) -> str: ...
|
||||
@property
|
||||
def permissions(self) -> dict: ...
|
||||
@property
|
||||
def repository_selection(self) -> str: ...
|
||||
|
||||
+14
-113
@@ -49,10 +49,7 @@
|
||||
|
||||
import datetime
|
||||
import pickle
|
||||
import time
|
||||
|
||||
import jwt
|
||||
import requests
|
||||
import urllib3
|
||||
|
||||
import github.ApplicationOAuth
|
||||
@@ -68,24 +65,13 @@ from . import (
|
||||
AuthenticatedUser,
|
||||
Consts,
|
||||
GithubApp,
|
||||
GithubException,
|
||||
GitignoreTemplate,
|
||||
HookDescription,
|
||||
Installation,
|
||||
InstallationAuthorization,
|
||||
RateLimit,
|
||||
Repository,
|
||||
)
|
||||
from .Requester import Requester
|
||||
|
||||
DEFAULT_BASE_URL = "https://api.github.com"
|
||||
DEFAULT_STATUS_URL = "https://status.github.com"
|
||||
# As of 2018-05-17, Github imposes a 10s limit for completion of API requests.
|
||||
# Thus, the timeout should be slightly > 10s to account for network/front-end
|
||||
# latency.
|
||||
DEFAULT_TIMEOUT = 15
|
||||
DEFAULT_PER_PAGE = 30
|
||||
|
||||
|
||||
class Github:
|
||||
"""
|
||||
@@ -97,10 +83,11 @@ class Github:
|
||||
login_or_token=None,
|
||||
password=None,
|
||||
jwt=None,
|
||||
base_url=DEFAULT_BASE_URL,
|
||||
timeout=DEFAULT_TIMEOUT,
|
||||
app_auth=None,
|
||||
base_url=Consts.DEFAULT_BASE_URL,
|
||||
timeout=Consts.DEFAULT_TIMEOUT,
|
||||
user_agent="PyGithub/Python",
|
||||
per_page=DEFAULT_PER_PAGE,
|
||||
per_page=Consts.DEFAULT_PER_PAGE,
|
||||
verify=True,
|
||||
retry=None,
|
||||
pool_size=None,
|
||||
@@ -108,6 +95,8 @@ class Github:
|
||||
"""
|
||||
:param login_or_token: string
|
||||
:param password: string
|
||||
:param jwt: string
|
||||
:param app_auth: github.AppAuthentication
|
||||
:param base_url: string
|
||||
:param timeout: integer
|
||||
:param user_agent: string
|
||||
@@ -125,14 +114,16 @@ class Github:
|
||||
assert user_agent is None or isinstance(user_agent, str), user_agent
|
||||
assert (
|
||||
retry is None
|
||||
or isinstance(retry, (int))
|
||||
or isinstance(retry, (urllib3.util.Retry))
|
||||
)
|
||||
assert pool_size is None or isinstance(pool_size, (int)), pool_size
|
||||
or isinstance(retry, int)
|
||||
or isinstance(retry, urllib3.util.Retry)
|
||||
), retry
|
||||
assert pool_size is None or isinstance(pool_size, int), pool_size
|
||||
|
||||
self.__requester = Requester(
|
||||
login_or_token,
|
||||
password,
|
||||
jwt,
|
||||
app_auth,
|
||||
base_url,
|
||||
timeout,
|
||||
user_agent,
|
||||
@@ -786,95 +777,5 @@ class Github:
|
||||
return GithubApp.GithubApp(self.__requester, headers, data, completed=True)
|
||||
|
||||
|
||||
class GithubIntegration:
|
||||
"""
|
||||
Main class to obtain tokens for a GitHub integration.
|
||||
"""
|
||||
|
||||
def __init__(self, integration_id, private_key, base_url=DEFAULT_BASE_URL):
|
||||
"""
|
||||
:param base_url: string
|
||||
:param integration_id: int
|
||||
:param private_key: string
|
||||
"""
|
||||
self.base_url = base_url
|
||||
self.integration_id = integration_id
|
||||
self.private_key = private_key
|
||||
assert isinstance(base_url, str), base_url
|
||||
|
||||
def create_jwt(self, expiration=60):
|
||||
"""
|
||||
Creates a signed JWT, valid for 60 seconds by default.
|
||||
The expiration can be extended beyond this, to a maximum of 600 seconds.
|
||||
|
||||
:param expiration: int
|
||||
:return string:
|
||||
"""
|
||||
now = int(time.time())
|
||||
payload = {"iat": now, "exp": now + expiration, "iss": self.integration_id}
|
||||
encrypted = jwt.encode(payload, key=self.private_key, algorithm="RS256")
|
||||
|
||||
if isinstance(encrypted, bytes):
|
||||
encrypted = encrypted.decode("utf-8")
|
||||
|
||||
return encrypted
|
||||
|
||||
def get_access_token(self, installation_id, user_id=None):
|
||||
"""
|
||||
Get an access token for the given installation id.
|
||||
POSTs https://api.github.com/app/installations/<installation_id>/access_tokens
|
||||
:param user_id: int
|
||||
:param installation_id: int
|
||||
:return: :class:`github.InstallationAuthorization.InstallationAuthorization`
|
||||
"""
|
||||
body = {}
|
||||
if user_id:
|
||||
body = {"user_id": user_id}
|
||||
response = requests.post(
|
||||
f"{self.base_url}/app/installations/{installation_id}/access_tokens",
|
||||
headers={
|
||||
"Authorization": f"Bearer {self.create_jwt()}",
|
||||
"Accept": Consts.mediaTypeIntegrationPreview,
|
||||
"User-Agent": "PyGithub/Python",
|
||||
},
|
||||
json=body,
|
||||
)
|
||||
|
||||
if response.status_code == 201:
|
||||
return InstallationAuthorization.InstallationAuthorization(
|
||||
requester=None, # not required, this is a NonCompletableGithubObject
|
||||
headers={}, # not required, this is a NonCompletableGithubObject
|
||||
attributes=response.json(),
|
||||
completed=True,
|
||||
)
|
||||
elif response.status_code == 403:
|
||||
raise GithubException.BadCredentialsException(
|
||||
status=response.status_code, data=response.text
|
||||
)
|
||||
elif response.status_code == 404:
|
||||
raise GithubException.UnknownObjectException(
|
||||
status=response.status_code, data=response.text
|
||||
)
|
||||
raise GithubException.GithubException(
|
||||
status=response.status_code, data=response.text
|
||||
)
|
||||
|
||||
def get_installation(self, owner, repo):
|
||||
"""
|
||||
:calls: `GET /repos/{owner}/{repo}/installation <https://docs.github.com/en/rest/reference/apps#get-a-repository-installation-for-the-authenticated-app>`_
|
||||
:param owner: str
|
||||
:param repo: str
|
||||
:rtype: :class:`github.Installation.Installation`
|
||||
"""
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.create_jwt()}",
|
||||
"Accept": Consts.mediaTypeIntegrationPreview,
|
||||
"User-Agent": "PyGithub/Python",
|
||||
}
|
||||
|
||||
response = requests.get(
|
||||
f"{self.base_url}/repos/{owner}/{repo}/installation",
|
||||
headers=headers,
|
||||
)
|
||||
response_dict = response.json()
|
||||
return Installation.Installation(None, headers, response_dict, True)
|
||||
# Retrocompatibility
|
||||
GithubIntegration = github.GithubIntegration
|
||||
|
||||
+5
-16
@@ -2,6 +2,7 @@ from datetime import datetime
|
||||
from io import BytesIO
|
||||
from typing import Any, Dict, List, Optional, Tuple, Type, TypeVar, Union, overload
|
||||
|
||||
from github.AppAuthentication import AppAuthentication
|
||||
from github.AuthenticatedUser import AuthenticatedUser
|
||||
from github.Commit import Commit
|
||||
from github.ContentFile import ContentFile
|
||||
@@ -10,8 +11,6 @@ from github.Gist import Gist
|
||||
from github.GithubObject import GithubObject, _NotSetType
|
||||
from github.GitignoreTemplate import GitignoreTemplate
|
||||
from github.HookDescription import HookDescription
|
||||
from github.Installation import Installation
|
||||
from github.InstallationAuthorization import InstallationAuthorization
|
||||
from github.Issue import Issue
|
||||
from github.License import License
|
||||
from github.NamedUser import NamedUser
|
||||
@@ -32,10 +31,9 @@ class Github:
|
||||
login_or_token: Optional[str] = ...,
|
||||
password: Optional[str] = ...,
|
||||
jwt: Optional[str] = ...,
|
||||
app_auth: Optional[AppAuthentication] = ...,
|
||||
base_url: str = ...,
|
||||
timeout: int = ...,
|
||||
client_id: Optional[str] = ...,
|
||||
client_secret: Optional[str] = ...,
|
||||
user_agent: str = ...,
|
||||
per_page: int = ...,
|
||||
verify: bool = ...,
|
||||
@@ -67,7 +65,6 @@ class Github:
|
||||
def get_gitignore_templates(self) -> List[str]: ...
|
||||
def get_hook(self, name: str) -> HookDescription: ...
|
||||
def get_hooks(self) -> List[HookDescription]: ...
|
||||
def get_installation(self, id: int) -> Installation: ...
|
||||
def get_license(self, key: Union[str, _NotSetType] = ...) -> License: ...
|
||||
def get_licenses(self) -> PaginatedList[License]: ...
|
||||
def get_organization(self, login: str) -> Organization: ...
|
||||
@@ -87,7 +84,9 @@ class Github:
|
||||
@overload
|
||||
def get_user(self, login: _NotSetType = ...) -> AuthenticatedUser: ...
|
||||
@overload
|
||||
def get_user(self, login: str) -> NamedUser: ...
|
||||
def get_user(
|
||||
self, login: Union[str, _NotSetType] = ...
|
||||
) -> Union[NamedUser, AuthenticatedUser]: ...
|
||||
def get_user_by_id(self, user_id: int) -> NamedUser: ...
|
||||
def get_users(
|
||||
self, since: Union[int, _NotSetType] = ...
|
||||
@@ -139,13 +138,3 @@ class Github:
|
||||
order: Union[str, _NotSetType] = ...,
|
||||
**qualifiers: Any
|
||||
) -> PaginatedList[NamedUser]: ...
|
||||
|
||||
class GithubIntegration:
|
||||
def __init__(
|
||||
self, integration_id: Union[int, str], private_key: str, base_url: str = ...
|
||||
) -> None: ...
|
||||
def create_jwt(self, expiration: int = ...) -> str: ...
|
||||
def get_access_token(
|
||||
self, installation_id: int, user_id: Optional[int] = ...
|
||||
) -> InstallationAuthorization: ...
|
||||
def get_installation(self, owner: str, repo: str) -> Installation: ...
|
||||
|
||||
+49
-1
@@ -51,6 +51,7 @@
|
||||
################################################################################
|
||||
|
||||
import base64
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
@@ -59,10 +60,14 @@ import re
|
||||
import time
|
||||
import urllib
|
||||
from io import IOBase
|
||||
from multiprocessing import RLock
|
||||
|
||||
import requests
|
||||
|
||||
from . import Consts, GithubException
|
||||
from . import Consts, GithubException, GithubIntegration
|
||||
|
||||
# For App authentication, time remaining before token expiration to request a new one
|
||||
ACCESS_TOKEN_REFRESH_THRESHOLD_SECONDS = 20
|
||||
|
||||
|
||||
class RequestsResponse:
|
||||
@@ -294,6 +299,7 @@ class Requester:
|
||||
login_or_token,
|
||||
password,
|
||||
jwt,
|
||||
app_auth,
|
||||
base_url,
|
||||
timeout,
|
||||
user_agent,
|
||||
@@ -304,6 +310,11 @@ class Requester:
|
||||
):
|
||||
self._initializeDebugFeature()
|
||||
|
||||
self.__installation_authorization = None
|
||||
self.__app_auth = app_auth
|
||||
|
||||
self.__auth_lock = RLock()
|
||||
|
||||
if password is not None:
|
||||
login = login_or_token
|
||||
b64 = (
|
||||
@@ -317,6 +328,8 @@ class Requester:
|
||||
self.__authorizationHeader = f"token {token}"
|
||||
elif jwt is not None:
|
||||
self.__authorizationHeader = f"Bearer {jwt}"
|
||||
elif self.__app_auth is not None:
|
||||
self._refresh_token()
|
||||
else:
|
||||
self.__authorizationHeader = None
|
||||
|
||||
@@ -349,6 +362,40 @@ class Requester:
|
||||
self.__userAgent = user_agent
|
||||
self.__verify = verify
|
||||
|
||||
def _must_refresh_token(self) -> bool:
|
||||
"""Check if it is time to refresh the API token gotten from the GitHub app installation"""
|
||||
if not self.__installation_authorization:
|
||||
return False
|
||||
return (
|
||||
self.__installation_authorization.expires_at
|
||||
< datetime.datetime.utcnow()
|
||||
+ datetime.timedelta(seconds=ACCESS_TOKEN_REFRESH_THRESHOLD_SECONDS)
|
||||
)
|
||||
|
||||
def _get_installation_authorization(self):
|
||||
assert self.__app_auth is not None
|
||||
integration = GithubIntegration.GithubIntegration(
|
||||
self.__app_auth.app_id, self.__app_auth.private_key
|
||||
)
|
||||
return integration.get_access_token(
|
||||
self.__app_auth.installation_id,
|
||||
permissions=self.__app_auth.token_permissions,
|
||||
)
|
||||
|
||||
def _refresh_token_if_needed(self) -> None:
|
||||
"""Get a new access token from the GitHub app installation if the one we have is about to expire"""
|
||||
if not self.__installation_authorization:
|
||||
return
|
||||
with self.__auth_lock:
|
||||
if self._must_refresh_token():
|
||||
logging.debug("Refreshing access token")
|
||||
self._refresh_token()
|
||||
|
||||
def _refresh_token(self) -> None:
|
||||
"""In the context of a GitHub app, refresh the access token"""
|
||||
self.__installation_authorization = self._get_installation_authorization()
|
||||
self.__authorizationHeader = f"token {self.__installation_authorization.token}"
|
||||
|
||||
def requestJsonAndCheck(self, verb, url, parameters=None, headers=None, input=None):
|
||||
return self.__check(
|
||||
*self.requestJson(
|
||||
@@ -578,6 +625,7 @@ class Requester:
|
||||
return status, responseHeaders, output
|
||||
|
||||
def __authenticate(self, url, requestHeaders, parameters):
|
||||
self._refresh_token_if_needed()
|
||||
if self.__authorizationHeader is not None:
|
||||
requestHeaders["Authorization"] = self.__authorizationHeader
|
||||
|
||||
|
||||
+20
-12
@@ -4,7 +4,9 @@ from typing import Any, Callable, Dict, Iterator, Optional, Tuple, Union
|
||||
|
||||
from requests.models import Response
|
||||
|
||||
from github.AppAuthentication import AppAuthentication
|
||||
from github.GithubObject import GithubObject
|
||||
from github.InstallationAuthorization import InstallationAuthorization
|
||||
|
||||
from urllib3.util import Retry
|
||||
|
||||
@@ -47,6 +49,8 @@ class HTTPSRequestsConnectionClass:
|
||||
) -> None: ...
|
||||
|
||||
class Requester:
|
||||
__installation_authorization: Optional[InstallationAuthorization] = ...
|
||||
__app_auth: Optional[AppAuthentication] = ...
|
||||
def DEBUG_ON_RESPONSE(
|
||||
self, statusCode: int, responseHeader: Dict[str, str], data: str
|
||||
) -> None: ...
|
||||
@@ -54,7 +58,7 @@ class Requester:
|
||||
def __check(
|
||||
self,
|
||||
status: int,
|
||||
responseHeader: Dict[str, Any],
|
||||
responseHeaders: Dict[str, Any],
|
||||
output: str,
|
||||
) -> Tuple[Dict[str, Any], Dict[str, Any]]: ...
|
||||
def __addParametersToUrl(
|
||||
@@ -65,7 +69,7 @@ class Requester:
|
||||
def __authenticate(
|
||||
self,
|
||||
url: str,
|
||||
responseHeader: Dict[str, Any],
|
||||
requestHeaders: Dict[str, Any],
|
||||
parameters: Dict[str, Any],
|
||||
) -> None: ...
|
||||
def __customConnection(
|
||||
@@ -88,37 +92,37 @@ class Requester:
|
||||
requestHeaders: Dict[str, str],
|
||||
input: Optional[str],
|
||||
status: Optional[int],
|
||||
responseHeader: Dict[str, Any],
|
||||
responseHeaders: Dict[str, Any],
|
||||
output: Optional[str],
|
||||
) -> None: ...
|
||||
def __makeAbsoluteUrl(self, url: str) -> str: ...
|
||||
def __structuredFromJson(self, data: str) -> Optional[Dict[str, Any]]: ...
|
||||
def __requestEncode(
|
||||
self,
|
||||
cnx: Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass],
|
||||
verb: str,
|
||||
url: str,
|
||||
parameters: Dict[str, str] = ...,
|
||||
headers: Dict[str, str] = ...,
|
||||
input: Optional[str] = ...,
|
||||
encode: Callable[[str], str] = ...,
|
||||
parameters: Dict[str, str],
|
||||
requestHeaders: Dict[str, str],
|
||||
input: Optional[str],
|
||||
encode: Callable[[str], str],
|
||||
) -> Tuple[int, Dict[str, Any], str]: ...
|
||||
def __requestRaw(
|
||||
self,
|
||||
cnx: Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass],
|
||||
verb: str,
|
||||
url: str,
|
||||
parameters: Dict[str, str] = ...,
|
||||
headers: Dict[str, str] = ...,
|
||||
input: Optional[str] = ...,
|
||||
requestHeaders: Dict[str, str],
|
||||
input: Optional[str],
|
||||
) -> Tuple[int, Dict[str, Any], str]: ...
|
||||
def __init__(
|
||||
self,
|
||||
login_or_token: Optional[str],
|
||||
password: Optional[str],
|
||||
jwt: Optional[str],
|
||||
app_auth: Optional[AppAuthentication],
|
||||
base_url: str,
|
||||
timeout: int,
|
||||
client_id: Optional[str],
|
||||
client_secret: Optional[str],
|
||||
user_agent: str,
|
||||
per_page: int,
|
||||
verify: bool,
|
||||
@@ -127,6 +131,10 @@ class Requester:
|
||||
) -> None: ...
|
||||
def _initializeDebugFeature(self) -> None: ...
|
||||
def check_me(self, obj: GithubObject) -> None: ...
|
||||
def _must_refresh_token(self) -> bool: ...
|
||||
def _get_installation_authorization(self) -> InstallationAuthorization: ...
|
||||
def _refresh_token_if_needed(self) -> None: ...
|
||||
def _refresh_token(self) -> None: ...
|
||||
@classmethod
|
||||
def injectConnectionClasses(
|
||||
cls, httpConnectionClass: Callable, httpsConnectionClass: Callable
|
||||
|
||||
@@ -1,19 +1,19 @@
|
||||
from typing import Any, Dict, List, Union
|
||||
|
||||
from github.GithubObject import NonCompletableGithubObject
|
||||
|
||||
class SelfHostedActionsRunner(NonCompletableGithubObject):
|
||||
def __repr__(self) -> str: ...
|
||||
def _initAttributes(self) -> None: ...
|
||||
def _useAttributes(self, attributes: Dict[str, Any]) -> None: ...
|
||||
@property
|
||||
def id(self) -> int: ...
|
||||
@property
|
||||
def name(self) -> str: ...
|
||||
@property
|
||||
def os(self) -> str: ...
|
||||
@property
|
||||
def status(self) -> str: ...
|
||||
@property
|
||||
def busy(self) -> bool: ...
|
||||
def labels(self) -> List[Dict[str, Union[str, int]]]: ...
|
||||
from typing import Any, Dict, List, Union
|
||||
|
||||
from github.GithubObject import NonCompletableGithubObject
|
||||
|
||||
class SelfHostedActionsRunner(NonCompletableGithubObject):
|
||||
def __repr__(self) -> str: ...
|
||||
def _initAttributes(self) -> None: ...
|
||||
def _useAttributes(self, attributes: Dict[str, Any]) -> None: ...
|
||||
@property
|
||||
def id(self) -> int: ...
|
||||
@property
|
||||
def name(self) -> str: ...
|
||||
@property
|
||||
def os(self) -> str: ...
|
||||
@property
|
||||
def status(self) -> str: ...
|
||||
@property
|
||||
def busy(self) -> bool: ...
|
||||
def labels(self) -> List[Dict[str, Union[str, int]]]: ...
|
||||
|
||||
+4
-1
@@ -35,6 +35,7 @@ like :class:`github.NamedUser.NamedUser` or :class:`github.Repository.Repository
|
||||
All classes inherit from :class:`github.GithubObject.GithubObject`.
|
||||
"""
|
||||
__all__ = [
|
||||
"AppAuthentication",
|
||||
"BadAttributeException",
|
||||
"BadCredentialsException",
|
||||
"BadUserAgentException",
|
||||
@@ -53,7 +54,9 @@ __all__ = [
|
||||
|
||||
import logging
|
||||
|
||||
from github.MainClass import Github, GithubIntegration
|
||||
from github.AppAuthentication import AppAuthentication
|
||||
from github.GithubIntegration import GithubIntegration
|
||||
from github.MainClass import Github
|
||||
|
||||
from .GithubException import (
|
||||
BadAttributeException,
|
||||
|
||||
+3
-2
@@ -1,11 +1,12 @@
|
||||
from github.AppAuthentication import AppAuthentication as AppAuthentication
|
||||
from github.GithubIntegration import GithubIntegration as GithubIntegration
|
||||
from github.MainClass import Github as Github
|
||||
from github.MainClass import GithubIntegration as GithubIntegration
|
||||
|
||||
from .GithubException import BadAttributeException as BadAttributeException
|
||||
from .GithubException import BadCredentialsException as BadCredentialsException
|
||||
from .GithubException import BadUserAgentException as BadUserAgentException
|
||||
from .GithubException import GithubException as GithubException
|
||||
from .GithubException import IncompletableObject as IncompleteableObject
|
||||
from .GithubException import IncompletableObject as IncompletableObject
|
||||
from .GithubException import RateLimitExceededException as RateLimitExceededException
|
||||
from .GithubException import TwoFactorException as TwoFactorException
|
||||
from .GithubException import UnknownObjectException as UnknownObjectException
|
||||
|
||||
Reference in New Issue
Block a user