| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| import urllib.parse |
| from typing import TYPE_CHECKING, Any |
|
|
| import github.AccessToken |
| import github.Auth |
| from github.Consts import DEFAULT_BASE_URL, DEFAULT_OAUTH_URL |
| from github.GithubException import BadCredentialsException, GithubException |
| from github.GithubObject import Attribute, NonCompletableGithubObject, NotSet |
| from github.Requester import Requester |
|
|
| if TYPE_CHECKING: |
| from github.AccessToken import AccessToken |
| from github.Auth import AppUserAuth |
|
|
|
|
| class ApplicationOAuth(NonCompletableGithubObject): |
| """ |
| This class is used for identifying and authorizing users for Github Apps. |
| |
| The reference can be found at |
| https://docs.github.com/en/developers/apps/building-github-apps/identifying-and-authorizing-users-for-github-apps |
| |
| """ |
|
|
| def _initAttributes(self) -> None: |
| self._client_id: Attribute[str] = NotSet |
| self._client_secret: Attribute[str] = NotSet |
|
|
| def __init__( |
| self, |
| requester: Requester, |
| headers: dict[str, Any], |
| attributes: Any, |
| ) -> None: |
| |
| requester = requester.withAuth(auth=None) |
| super().__init__(requester, headers, attributes) |
|
|
| def __repr__(self) -> str: |
| return self.get__repr__({"client_id": self._client_id.value}) |
|
|
| @property |
| def client_id(self) -> str: |
| return self._client_id.value |
|
|
| @property |
| def client_secret(self) -> str: |
| return self._client_secret.value |
|
|
| def get_oauth_url(self, path: str) -> str: |
| if not path.startswith("/"): |
| path = f"/{path}" |
|
|
| if self._requester.base_url == DEFAULT_BASE_URL: |
| base_url = DEFAULT_OAUTH_URL |
| else: |
| base_url = f"{self._requester.scheme}://{self._requester.hostname_and_port}/login/oauth" |
| return f"{base_url}{path}" |
|
|
| def get_login_url( |
| self, |
| redirect_uri: str | None = None, |
| state: str | None = None, |
| login: str | None = None, |
| ) -> str: |
| """ |
| Return the URL you need to redirect a user to in order to authorize your App. |
| """ |
| parameters = {"client_id": self.client_id} |
| if redirect_uri is not None: |
| assert isinstance(redirect_uri, str), redirect_uri |
| parameters["redirect_uri"] = redirect_uri |
| if state is not None: |
| assert isinstance(state, str), state |
| parameters["state"] = state |
| if login is not None: |
| assert isinstance(login, str), login |
| parameters["login"] = login |
|
|
| query = urllib.parse.urlencode(parameters) |
|
|
| return self.get_oauth_url(f"/authorize?{query}") |
|
|
| def get_access_token(self, code: str, state: str | None = None) -> AccessToken: |
| """ |
| :calls: `POST /login/oauth/access_token <https://docs.github.com/en/developers/apps/identifying-and-authorizing-users-for-github-apps>`_ |
| """ |
| assert isinstance(code, str), code |
| post_parameters = { |
| "code": code, |
| "client_id": self.client_id, |
| "client_secret": self.client_secret, |
| } |
|
|
| if state is not None: |
| post_parameters["state"] = state |
|
|
| headers, data = self._checkError( |
| *self._requester.requestJsonAndCheck( |
| "POST", |
| self.get_oauth_url("/access_token"), |
| headers={"Accept": "application/json"}, |
| input=post_parameters, |
| ) |
| ) |
|
|
| return github.AccessToken.AccessToken( |
| requester=self._requester, |
| headers=headers, |
| attributes=data, |
| ) |
|
|
| def get_app_user_auth(self, token: AccessToken) -> AppUserAuth: |
| return github.Auth.AppUserAuth( |
| client_id=self.client_id, |
| client_secret=self.client_secret, |
| token=token.token, |
| token_type=token.type, |
| expires_at=token.expires_at, |
| refresh_token=token.refresh_token, |
| refresh_expires_at=token.refresh_expires_at, |
| requester=self._requester, |
| ) |
|
|
| def refresh_access_token(self, refresh_token: str) -> AccessToken: |
| """ |
| :calls: `POST /login/oauth/access_token <https://docs.github.com/en/developers/apps/identifying-and-authorizing-users-for-github-apps>`_ |
| :param refresh_token: string |
| """ |
| assert isinstance(refresh_token, str) |
| post_parameters = { |
| "client_id": self.client_id, |
| "client_secret": self.client_secret, |
| "grant_type": "refresh_token", |
| "refresh_token": refresh_token, |
| } |
|
|
| headers, data = self._checkError( |
| *self._requester.requestJsonAndCheck( |
| "POST", |
| self.get_oauth_url("/access_token"), |
| headers={"Accept": "application/json"}, |
| input=post_parameters, |
| ) |
| ) |
|
|
| return github.AccessToken.AccessToken( |
| requester=self._requester, |
| headers=headers, |
| attributes=data, |
| ) |
|
|
| @staticmethod |
| def _checkError(headers: dict[str, Any], data: Any) -> tuple[dict[str, Any], Any]: |
| if isinstance(data, dict) and "error" in data: |
| if data["error"] == "bad_verification_code": |
| raise BadCredentialsException(200, data, headers) |
| raise GithubException(200, data, headers) |
|
|
| return headers, data |
|
|
| def _useAttributes(self, attributes: dict[str, Any]) -> None: |
| if "client_id" in attributes: |
| self._client_id = self._makeStringAttribute(attributes["client_id"]) |
| if "client_secret" in attributes: |
| self._client_secret = self._makeStringAttribute(attributes["client_secret"]) |
|
|
