Skip to content

Instantly share code, notes, and snippets.

@mikegrima
Last active February 14, 2023 15:12
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save mikegrima/604e631e6c970dc2d69dd9bfe2979124 to your computer and use it in GitHub Desktop.
GitHub App Token Code

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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment