feat: add support for GitHub JWT auth

.gitignore

@@ -106,3 +106,6 @@ venv.bak/
 # mypy
 .mypy_cache/
 
+app
+installation
+pem

releasetool/__main__.py

@@ -156,7 +156,20 @@ def reset_config():
 @main.command(name="publish-reporter-start")
 @click.option("--github_token", envvar="GITHUB_TOKEN", default=None)
 @click.option("--pr", envvar="AUTORELEASE_PR", default=None)
-def publish_reporter_start(github_token: str, pr: str):
+@click.option("--app_id_path", envvar="APP_ID_PATH", default=None)
+@click.option("--installation_id_path", envvar="INSTALLATION_ID_PATH", default=None)
+@click.option("--private_key_path", envvar="GITHUB_PRIVATE_KEY_PATH", default=None)
+def publish_reporter_start(
+    github_token: str,
+    pr: str,
+    app_id_path: str,
+    installation_id_path: str,
+    private_key_path: str,
+):
+    if app_id_path:
+        github_token = github_jwt_dict(
+            app_id_path, installation_id_path, private_key_path
+        )
     releasetool.commands.publish_reporter.start(github_token, pr)
 
 
@@ -165,10 +178,36 @@ def publish_reporter_start(github_token: str, pr: str):
 @click.option("--pr", envvar="AUTORELEASE_PR", default=None)
 @click.option("--status", type=bool, default=True)
 @click.option("--details", envvar="PUBLISH_DETAILS", default=None)
-def publish_reporter_finish(github_token: str, pr: str, status: bool, details: str):
+@click.option("--app_id_path", envvar="APP_ID_PATH", default=None)
+@click.option("--installation_id_path", envvar="INSTALLATION_ID_PATH", default=None)
+@click.option("--private_key_path", envvar="GITHUB_PRIVATE_KEY_PATH", default=None)
+def publish_reporter_finish(
+    github_token: str,
+    pr: str,
+    status: bool,
+    details: str,
+    app_id_path: str,
+    installation_id_path: str,
+    private_key_path: str,
+):
+    if app_id_path:
+        github_token = github_jwt_dict(
+            app_id_path, installation_id_path, private_key_path
+        )
     releasetool.commands.publish_reporter.finish(github_token, pr, status, details)
 
 
+def github_jwt_dict(app_id_path: str, installation_id_path: str, private_key_path: str):
+    """An app_id, installation_id, and private_key may be provided, rather
+    than a github_token. This dictionary of values is passed to publish_reporter
+    which exchanges them for a JWT."""
+    return {
+        "app_id": open(app_id_path, "r").read(),
+        "installation_id": open(installation_id_path, "r").read(),
+        "private_key": open(private_key_path, "r").read(),
+    }
+
+
 @main.command(name="publish-reporter-script")
 def publish_reporter_script():
     releasetool.commands.publish_reporter.script()

releasetool/commands/common.py

@@ -103,7 +103,7 @@ def setup_github_context(
         "Please provide your GitHub API token with write:repo_hook and "
         "public_repo (https://github.com/settings/tokens)",
     )
-    ctx.github = releasetool.github.GitHub(github_token)
+    ctx.github = releasetool.github.GitHub(releasetool.github.GitHubToken(github_token))
 
     _determine_origin(ctx)
     _determine_upstream(ctx, owners)

releasetool/commands/publish_reporter.py

@@ -17,7 +17,7 @@
 import os
 import pkgutil
 import re
-from typing import Tuple
+from typing import cast, Tuple, Union
 from requests import HTTPError
 
 import releasetool.github
@@ -73,15 +73,23 @@ def extract_pr_details(pr) -> Tuple[str, str, str]:
     return match.group("owner"), match.group("repo"), match.group("number")
 
 
-def start(github_token: str, pr: str) -> None:
+def start(github_token: Union[str, dict], pr: str) -> None:
     """Reports the start of a publication job to GitHub."""
-    github_token = figure_out_github_token(github_token)
+    # If we are passed a dictionary for github_token, assume we are
+    # retrieveing a JWT, and do not use magic proxy:
+    use_proxy = True
+    if type(github_token) is dict:
+        use_proxy = False
+    else:
+        github_token = figure_out_github_token(cast(str, github_token))
 
     if not github_token or not pr:
         print("No github token or PR specified to report status to, returning.")
         return
 
-    gh = releasetool.github.GitHub(github_token, use_proxy=True)
+    gh = releasetool.github.GitHub(
+        releasetool.github.GitHubToken(github_token), use_proxy=use_proxy
+    )
 
     try:
         owner, repo, number = extract_pr_details(pr)
@@ -109,15 +117,23 @@ def start(github_token: str, pr: str) -> None:
         raise Exception(f"Error commenting on PR: {e.response.status_code}")
 
 
-def finish(github_token: str, pr: str, status: bool, details: str) -> None:
+def finish(github_token: Union[str, dict], pr: str, status: bool, details: str) -> None:
     """Reports the completion of a publication job to GitHub."""
-    github_token = figure_out_github_token(github_token)
+    # If we are passed a dictionary for github_token, assume we are
+    # retrieveing a JWT, and do not use magic proxy:
+    use_proxy = True
+    if type(github_token) is dict:
+        use_proxy = False
+    else:
+        github_token = figure_out_github_token(cast(str, github_token))
 
     if not github_token or not pr:
         print("No github token or PR specified to report status to, returning.")
         return
 
-    gh = releasetool.github.GitHub(github_token, use_proxy=True)
+    gh = releasetool.github.GitHub(
+        releasetool.github.GitHubToken(github_token), use_proxy=use_proxy
+    )
 
     try:
         owner, repo, number = extract_pr_details(pr)

releasetool/commands/publish_reporter.sh

@@ -14,6 +14,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+export APP_ID_PATH="$KOKORO_GFILE_DIR/secret_manager/releasetool-publish-reporter-app"
+export INSTALLATION_ID_PATH="$KOKORO_GFILE_DIR/secret_manager/releasetool-publish-reporter-googleapis-installation"
+export GITHUB_PRIVATE_KEY_PATH="$KOKORO_GFILE_DIR/secret_manager/releasetool-publish-reporter-pem"
+
 # Install an exit hook to report status.
 releasetool_finish_report() {
     rv=$?

releasetool/github.py

@@ -13,16 +13,29 @@
 # limitations under the License.
 
 import base64
+import json
 import os
 import re
-from typing import List, Sequence, Union
+import time
 
+from typing import cast, List, Sequence, Union
+
+import jwt
 import requests
+from cryptography.hazmat.backends import default_backend
 
 
 _GITHUB_ROOT: str = "https://api.github.com"
 _GITHUB_UI_ROOT: str = "https://github.com"
-_MAGIC_GITHUB_PROXY_ROOT: str = "https://magic-github-proxy.endpoints.devrel-prod.cloud.goog"
+# TODO: remove references to magic proxy, once we have confirmed the JWT-based
+# approach is working well.
+#
+# magic-proxy provided similar functionality to installation-scoped access
+# tokens, by allowing a token to be generated scoped to a single repository,
+# with limited credentials.
+_MAGIC_GITHUB_PROXY_ROOT: str = (
+    "https://magic-github-proxy.endpoints.devrel-prod.cloud.goog"
+)
 
 
 def _find_devrel_api_key() -> str:
@@ -54,14 +67,71 @@ def _find_devrel_api_key() -> str:
     return magic_github_proxy_key
 
 
+class GitHubToken:
+    def __init__(self, token: Union[str, dict]):
+        self.auth_type = "Bearer"
+        # If a dictionary is provided for token, assume it
+        # contains app_id, installation, private_key, such that we
+        # can fetch a JWT:
+        if type(token) is dict:
+            self.auth_type = "token"
+            token_dict = cast(dict, token)
+            self.token = get_installation_access_token(
+                token_dict["app_id"],
+                token_dict["installation_id"],
+                token_dict["private_key"],
+            )
+
+    def get_auth_type(self) -> str:
+        return self.auth_type
+
+    def get_token(self) -> str:
+        return self.token
+
+
+def get_installation_access_token(
+    app_id: str, installation_id: str, private_key_str: str
+) -> str:
+    """Use GitHub API to exchange app_id, installation_id, and private_key
+    for an installation-specific access_token, see:
+    https://developer.github.com/apps/building-github-apps/authenticating-with-github-apps/#authenticating-as-a-github-app
+    """
+    time_since_epoch_in_seconds = int(time.time())
+    payload = {
+        "iat": time_since_epoch_in_seconds,
+        "exp": time_since_epoch_in_seconds + (10 * 60),
+        "iss": app_id,
+    }
+
+    private_key_bytes = private_key_str.encode()
+    private_key = default_backend().load_pem_private_key(private_key_bytes, None)
+    app_jwt = jwt.encode(payload, private_key, algorithm="RS256")
+
+    headers = {
+        "Authorization": "Bearer {}".format(app_jwt.decode()),
+        "Accept": "application/vnd.github.machine-man-preview+json",
+    }
+
+    resp = requests.post(
+        "https://api.github.com/app/installations/{}/access_tokens".format(
+            installation_id
+        ),
+        headers=headers,
+    )
+
+    if resp.status_code != 201:
+        raise Exception("Could exchange certificate for JWT.")
+    return json.loads(resp.content.decode())["token"]
+
+
 class GitHub:
-    def __init__(self, token: str, use_proxy: bool = False) -> None:
+    def __init__(self, token: GitHubToken, use_proxy: bool = False) -> None:
         self.session: requests.Session = requests.Session()
         self.GITHUB_ROOT = _GITHUB_ROOT
         self.session.headers.update(
             {
                 "Accept": "application/vnd.github.v3+json",
-                "Authorization": f"Bearer {token}",
+                "Authorization": f"{token.get_auth_type()} {token.get_token()}",
             }
         )
 

setup.py

@@ -22,8 +22,10 @@
     "requests",
     "attrs",
     "click",
+    "cryptography",
     "keyring",
     "packaging",
+    "pyjwt",
     "pyperclip",
     "python-dateutil",
 ]