This is code that makes it very easy to obtain a GitHub token for a GitHub application in Python.
This code works by performing the full song and dance required to obtain a token to operate in a GitHub organization. This manages the secret in AWS Secrets manager, caches the tokens and credentails, and refreshes them when neeeded.
The best part is the @github_auth decorator, which injects the credentials into your function when you supply the org name
automagically.
| import os | |
| from typing import Any, Dict | |
| CONFIG_DICT = { | |
| "TEST": { | |
| "GITHUB_ORGS": { | |
| "ORGNAME": {"AppId": "APP-ID-HERE", "InstallationId": "INSTALLATION-ID-HERE"}, | |
| } | |
| }, | |
| "PROD": { | |
| "GITHUB_ORGS": { | |
| "ORGNAME": {"AppId": "APP-ID-HERE", "InstallationId": "INSTALLATION-ID-HERE"}, | |
| }, | |
| }, | |
| } | |
| class ConfigurationManager: | |
| """A configuration management class.""" | |
| def __init__(self, environment: str): | |
| """This is the main constructor that will take in the name of the environment to use.""" | |
| self._configuration = CONFIG_DICT[environment] | |
| def load_config_environment(self, environment: str): | |
| """This will load a different configuration environment than what the environment variable is set as.""" | |
| self._configuration = CONFIG_DICT[environment] | |
| @property | |
| def config(self) -> Dict[str, Any]: | |
| """This returns the configuration dictionary for the configured environment.""" | |
| return self._configuration | |
| CONFIGURATION_MANAGER = ConfigurationManager(os.environ.get("YOUR_ENVIRONMENT", "TEST")) |
| """GitHub authentication requires a few things to be done: | |
| 1. The app needs to be made in the org | |
| 2. The app needs to be installed in the org | |
| 3. You need to make what is referred to as an "app token", which makes use of the app's private key | |
| 4. With the "app token" you then need to obtain the "installation token" to operate in the org. | |
| 5. Once you have the "installation token", then you are able to make API calls to GitHub. | |
| """ | |
| import hmac | |
| from datetime import datetime, timezone | |
| from functools import wraps | |
| import logging | |
| from typing import Dict, Any, Callable | |
| import jwt | |
| import requests | |
| from configuration import CONFIGURATION_MANAGER | |
| from secrets import SECRETS_MANAGER | |
| LOGGER = logging.getLogger(__name__) | |
| class GitHubAuthError(Exception): | |
| """Error raised if we get an invalid response code back from GitHub when trying to get the installation token.""" | |
| def auth_webhook(event_payload: Dict[str, Any]) -> bool: | |
| """This will verify that the webhook is in fact from GitHub and properly signed by GitHub. | |
| This follows GitHub's HMAC documentation: https://docs.github.com/en/developers/webhooks-and-events/webhooks/securing-your-webhooks | |
| :returns bool: True if this is good, False otherwise. If false, you should raise some exceptions and even alarms. 😱 | |
| """ | |
| try: | |
| github_signature = event_payload["headers"]["x-hub-signature-256"] | |
| body = event_payload["body"].encode("utf-8") | |
| except KeyError as kerr: | |
| LOGGER.exception(kerr) | |
| LOGGER.error("[🚩] Missing required details in GitHub auth payload. See stacktrace. Not authenticated.") | |
| return False | |
| hash_obj = hmac.new(SECRETS_MANAGER.secrets["WEBHOOK_SECRET"].encode("utf-8"), body, digestmod="sha256") | |
| digest = f"sha256={hash_obj.hexdigest()}" | |
| return hmac.compare_digest(digest, github_signature) | |
| class GitHubAuthManager: | |
| """This is a class that manages the GitHub authentication as a singleton.""" | |
| def __init__(self): | |
| self._app_tokens: Dict[str, str] = {} | |
| self._installation_tokens: Dict[str, Dict[str, Any]] = {} | |
| def _make_app_token(self, organization: str, app_id: str, secret: str) -> None: | |
| """This will generate the "app" Bearer tokens that are used for obtaining the corresponding GitHub installation tokens. | |
| This follows GitHub's documentation here: | |
| https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-a-github-app | |
| """ | |
| LOGGER.debug(f"[🔑] Generating app token for org: {organization}...") | |
| now = int(datetime.now(tz=timezone.utc).timestamp()) | |
| payload = { | |
| "iat": now - 60, # issued at time, 60 seconds in the past to allow for clock drift | |
| "exp": now + 300, # JWT expiration time (can be a max of 10 minute, but we are setting to 3) | |
| "iss": app_id, # GitHub App's identifier | |
| } | |
| token = jwt.encode(payload, secret, algorithm="RS256") | |
| self._app_tokens[organization] = token | |
| def _make_installation_token(self, organization: str) -> None: | |
| """This will make the installation token for the corresponding app. | |
| Follows the instructions here: https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-an-installation | |
| """ | |
| LOGGER.debug(f"[🔑] Generating the installation token for org: {organization}...") | |
| installation_id = CONFIGURATION_MANAGER.config["GITHUB_ORGS"][organization]["InstallationId"] | |
| auth_header = {"Authorization": f"Bearer {self._app_tokens[organization]}"} | |
| result = requests.post(f"https://api.github.com/app/installations/{installation_id}/access_tokens", headers=auth_header, timeout=20) | |
| if result.status_code != 201: | |
| LOGGER.error(f"[💥] Invalid response back from GitHub while obtaining the installation token: {result.status_code}") | |
| raise GitHubAuthError() | |
| result_json = result.json() | |
| token = result_json["token"] | |
| expiration = int(datetime.strptime(result_json["expires_at"], "%Y-%m-%dT%H:%M:%SZ").timestamp()) | |
| self._installation_tokens[organization] = {"expiration": expiration, "token": token} | |
| def authenticate(self, organization: str) -> Dict[str, str]: | |
| """This will perform all the logic required to authenticate to GitHub. | |
| This will raise a ValueError if the organization name is not in the configuration. | |
| """ | |
| # If we have unexpired cached credentials, then use them: | |
| current_creds = self._installation_tokens.get(organization, {}) | |
| if current_creds.get("expiration", 0) > int(datetime.now(tz=timezone.utc).timestamp()): | |
| LOGGER.debug("[💵] Using cached credentials.") | |
| return {"Authorization": f"Bearer {current_creds['token']}"} | |
| # Make the app token: | |
| try: | |
| app_id = CONFIGURATION_MANAGER.config["GITHUB_ORGS"][organization]["AppId"] # noqa | |
| secret = SECRETS_MANAGER.secrets["GITHUB_ORGS"][organization] | |
| self._make_app_token(organization, app_id, secret) | |
| except KeyError: | |
| LOGGER.error( | |
| f"[💥] GitHub organization: {organization} is not configured to be managed by this stack. Update the `GITHUB_ORGS` and " | |
| "corresponding secret store if this is wrong." | |
| ) | |
| raise | |
| # Make the installation token: | |
| self._make_installation_token(organization) | |
| return {"Authorization": f"Bearer {self._installation_tokens[organization]['token']}"} | |
| GITHUB_AUTH_MANGER = GitHubAuthManager() | |
| def github_auth(func: Callable) -> Callable: | |
| """This is a decorator for injecting GitHub Authorization headers into your function.""" | |
| @wraps(func) | |
| def wrapped_function(organization: str, *args, **kwargs) -> Any: | |
| """This is the wrapped function that will get the GitHub credentials injected into it. | |
| The function needs to take 1 positional arg: the organization name, which is a string. | |
| The auth header is passed in via a keyword arg named `github_auth_header`, which is a Dict[str, str]. | |
| Example usage: | |
| @github_auth | |
| def my_function(organization: str, github_auth_header: Dict[str, str] = None) -> None: | |
| ... | |
| requests.post("https://api.github.com...", headers=github_auth_header) | |
| ... | |
| """ | |
| kwargs["github_auth_header"] = GITHUB_AUTH_MANGER.authenticate(organization) | |
| return func(organization, *args, **kwargs) | |
| return wrapped_function |
| import json | |
| import logging | |
| from typing import Any, Dict | |
| import boto3 | |
| """ | |
| Make your secrets look like a dictionary: | |
| { | |
| "GITHUB_ORGS": { | |
| "ORG_NAME": "The app that is tied to the org's private key here..." | |
| ... | |
| }, | |
| } | |
| """ | |
| LOGGER = logging.getLogger(__name__) | |
| class SecretsManager: | |
| """This an AWS secrets manager class.""" | |
| def __init__(self): | |
| """Default constructor""" | |
| self._secrets = None | |
| def load_secrets(self) -> None: | |
| """This will perform the work to load the secret value from AWS Secrets manager. | |
| Note: If there are exceptions encountered fetching the secret, it will raise up the stack. | |
| """ | |
| LOGGER.debug(f"[🤐] Loading secrets from Secrets Manager ARN: 'YOUR ARN HERE'") | |
| client = boto3.client("secretsmanager", "YOUR REGION HERE") | |
| loaded = client.get_secret_value(SecretId="YOUR SECRET ID HERE") | |
| self._secrets = json.loads(loaded["SecretString"]) | |
| LOGGER.debug("[🔑] Secrets loaded successfully") | |
| @property | |
| def secrets(self) -> Dict[str, Any]: | |
| """Fetch the Secret Dictionary. This will load the secrets if not already loaded.""" | |
| if not self._secrets: | |
| self.load_secrets() | |
| return self._secrets | |
| SECRETS_MANAGER = SecretsManager() |