New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Support full GitHub app authentication #1986
Changes from 22 commits
885a522
a719746
c39a183
d8b20e1
30a11a0
c15de72
f8a071b
e957467
add6683
3ba6e76
8df28e9
a461e0e
94ca7be
8e47431
a9c2f81
b88f36f
8534fe9
d502f27
350911e
15f1c0c
99e6b3a
ceb8d7d
df38766
8899a9e
38fb3cd
2c106c6
ce199bf
a50aecd
0e245d0
7d61945
a7ca0cb
7dd9fa9
7e6d5b7
148c14f
3e16166
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,198 @@ | ||
import time | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Moved from |
||
|
||
import deprecated | ||
import jwt | ||
|
||
from github import Consts | ||
from github.GithubException import GithubException, UnknownObjectException | ||
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.JWT_EXPIRY, | ||
): | ||
""" | ||
:param integration_id: int | ||
:param private_key: string | ||
:param base_url: string | ||
:param jwt_expiry: int | ||
""" | ||
assert isinstance(integration_id, (int, str)), integration_id | ||
assert isinstance(private_key, str), private_key | ||
dblanchette marked this conversation as resolved.
Show resolved
Hide resolved
|
||
assert isinstance(base_url, str), base_url | ||
assert isinstance(jwt_expiry, int), jwt_expiry | ||
assert 15 <= jwt_expiry <= 600, jwt_expiry | ||
dblanchette marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
self.base_url = base_url | ||
self.integration_id = integration_id | ||
self.private_key = private_key | ||
self.jwt_expiry = jwt_expiry | ||
self.__requester = Requester( | ||
login_or_token=None, | ||
password=None, | ||
jwt=self.create_jwt(), | ||
app_id=None, | ||
app_private_key=None, | ||
app_installation_id=None, | ||
app_token_permissions=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` | ||
""" | ||
try: | ||
headers, response = self.__requester.requestJsonAndCheck( | ||
"GET", url, headers=self._get_headers() | ||
) | ||
except UnknownObjectException: | ||
raise | ||
except GithubException: | ||
raise | ||
return Installation( | ||
requester=self.__requester, | ||
headers=headers, | ||
attributes=response, | ||
completed=True, | ||
) | ||
|
||
def create_jwt(self): | ||
""" | ||
Creates a signed JWT, valid for 60 seconds by default. Could be set to a maximum of 600 seconds. | ||
https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-a-github-app | ||
|
||
dblanchette marked this conversation as resolved.
Show resolved
Hide resolved
|
||
:return string: | ||
""" | ||
now = int(time.time()) | ||
dblanchette marked this conversation as resolved.
Show resolved
Hide resolved
|
||
payload = {"iat": now, "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} | ||
try: | ||
headers, response = self.__requester.requestJsonAndCheck( | ||
"POST", | ||
f"/app/installations/{installation_id}/access_tokens", | ||
input=body, | ||
) | ||
except UnknownObjectException: | ||
s-t-e-v-e-n-k marked this conversation as resolved.
Show resolved
Hide resolved
|
||
raise | ||
except GithubException: | ||
raise | ||
|
||
return InstallationAuthorization( | ||
requester=self.__requester, | ||
headers=headers, | ||
attributes=response, | ||
completed=True, | ||
) | ||
|
||
@deprecated.deprecated("Use get_repo_installation") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let me know if that does not suit you. I feel that this call is badly named There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is the existing call? It should probably also state it's deprecated in the docstring. Also, it should mention the class, not just the bare function. |
||
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` | ||
""" | ||
return self._get_installed_app(url=f"/repos/{owner}/{repo}/installation") | ||
|
||
def get_installations(self): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is the part I added in that file. Users having a private app with more than one installations can use this call as the main class always uses the first found installation. |
||
""" | ||
: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}") |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
from typing import Union, Optional | ||
|
||
from github.Installation import Installation | ||
from github.InstallationAuthorization import InstallationAuthorization | ||
from github.PaginatedList import PaginatedList | ||
|
||
class GithubIntegration: | ||
def __init__( | ||
self, | ||
integration_id: Union[int, str], | ||
private_key: str, | ||
base_url: str = ..., | ||
jwt_expiry: int = ..., | ||
) -> None: ... | ||
def _get_installed_app(self, url: str) -> Installation: ... | ||
def _get_headers(self) -> dict: ... | ||
dblanchette marked this conversation as resolved.
Show resolved
Hide resolved
|
||
def create_jwt(self, expiration: int = ...) -> str: ... | ||
def get_access_token( | ||
self, installation_id: int, permissions: Optional[dict] = ... | ||
) -> 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: ... |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moved those here since they are used in 2 files now (
MainClass
andGithubIntegration
)There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Comparing this is really difficult. Would have been better to do the refactoring (without changes) in a separate PR and base this PR on that. Then we would see only changes, not moving unchanged code around.