Feature GitHub auth

.gitignore

@@ -160,3 +160,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+*.pem

deploy/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8-bullseye
+FROM python:3.10-bookworm
 
 RUN sed -i "s@http://deb.debian.org@http://mirrors.aliyun.com@g" /etc/apt/sources.list
 RUN sed -i "s@http://security.debian.org@http://mirrors.aliyun.com@g" /etc/apt/sources.list

pyproject.toml

@@ -14,6 +14,8 @@ dependencies = [
     "pymysql>=1.1.0",
     "click>=8.1.7",
     "bson>=0.5.10",
+    "jwt>=1.3.1",
+    "urllib3>=2.1.0",
 ]
 requires-python = ">=3.10"
 readme = "README.md"

requirements.txt

@@ -8,8 +8,10 @@ ca-lark-oauth==0.0.5
 ca-lark-sdk==0.0.7
 ca-lark-webhook==0.0.3
 certifi==2023.11.17
+cffi==1.16.0
 click==8.1.7
 colorama==0.4.6; platform_system == "Windows"
+cryptography==41.0.7
 exceptiongroup==1.2.0; python_version < "3.11"
 Flask==3.0.0
 flask-cors==4.0.0
@@ -21,7 +23,9 @@ httpx==0.26.0
 idna==3.6
 itsdangerous==2.1.2
 Jinja2==3.1.2
+jwt==1.3.1
 MarkupSafe==2.1.3
+pycparser==2.21
 pycryptodome==3.19.1
 pymysql==1.1.0
 python-dateutil==2.8.2
@@ -30,4 +34,5 @@ six==1.16.0
 sniffio==1.3.0
 sqlalchemy==2.0.23
 typing-extensions==4.9.0
+urllib3==2.1.0
 Werkzeug==3.0.1

server/routes/__init__.py

@@ -1 +1,2 @@
+from .github import *
 from .lark import *

server/routes/github.py

@@ -0,0 +1,79 @@
+import logging
+import os
+
+from app import app
+from flask import Blueprint, abort, redirect, request
+from utils.github import get_installation_token, get_jwt, register_by_code
+
+bp = Blueprint("github", __name__, url_prefix="/api/github")
+
+
+@bp.route("/install", methods=["GET"])
+def github_install():
+    """Install GitHub App.
+
+    If not `installation_id`, redirect to install page.
+    If `installation_id`, get installation token.
+
+    If `code`, register by code.
+    """
+    installation_id = request.args.get("installation_id", None)
+
+    if installation_id is None:
+        return redirect(
+            f"https://github.com/apps/{os.environ.get('GITHUB_APP_NAME')}/installations/new"
+        )
+
+    logging.debug(f"installation_id: {installation_id}")
+
+    jwt = get_jwt(
+        os.environ.get("GITHUB_APP_PRIVATE_KEY_PATH"),
+        os.environ.get("GITHUB_APP_ID"),
+    )
+
+    installation_token = get_installation_token(jwt, installation_id)
+    if installation_token is None:
+        logging.debug("Failed to get installation token.")
+
+        # TODO: 统一解决各类 http 请求失败的情况
+        abort(500)
+    logging.debug(f"installation_token: {installation_token}")
+
+    # 如果有 code 参数，则为该用户注册
+    code = request.args.get("code", None)
+    if code is not None:
+        logging.debug(f"code: {code}")
+
+        user_token = register_by_code(code)
+        if user_token is None:
+            logging.debug("Failed to register by code.")
+            abort(500)
+
+        logging.debug(f"user_token: {user_token}")
+
+    return "Success!"
+
+
+@bp.route("/register", methods=["GET"])
+def github_register():
+    """GitHub OAuth register.
+
+    If not `code`, redirect to GitHub OAuth page.
+    If `code`, register by code.
+    """
+    code = request.args.get("code", None)
+    if code is None:
+        return redirect(
+            f"https://github.com/login/oauth/authorize?client_id={os.environ.get('GITHUB_CLIENT_ID')}"
+        )
+
+    logging.debug(f"code: {code}")
+    user_token = register_by_code(code)
+    if user_token is None:
+        return "Failed to register by code."
+
+    logging.debug(f"user_token: {user_token}")
+    return user_token
+
+
+app.register_blueprint(bp)

server/utils/github.py

@@ -0,0 +1,95 @@
+import os
+import time
+from urllib.parse import parse_qs
+
+import httpx
+from jwt import JWT, jwk_from_pem
+
+
+def get_jwt(pem_path: str, app_id: str) -> str:
+    """Generate a JSON Web Token (JWT) for authentication.
+
+    Args:
+        pem_path (str): path to the private key file.
+        app_id (str): GitHub App's identifier.
+
+    Returns:
+        str: JWT.
+    """
+
+    # Open PEM
+    with open(pem_path, "rb") as pem_file:
+        signing_key = jwk_from_pem(pem_file.read())
+
+    payload = {
+        # Issued at time
+        "iat": int(time.time()),
+        # JWT expiration time (10 minutes maximum)
+        "exp": int(time.time()) + 600,
+        # GitHub App's identifier
+        "iss": app_id,
+    }
+
+    # Create JWT
+    jwt_instance = JWT()
+    encoded_jwt = jwt_instance.encode(payload, signing_key, alg="RS256")
+
+    return encoded_jwt
+
+
+def get_installation_token(jwt: str, installation_id: str) -> str | None:
+    """Get installation access token
+
+    Args:
+        jwt (str): The JSON Web Token used for authentication.
+        installation_id (str): The ID of the installation.
+
+    Returns:
+        str: The installation access token.
+    """
+
+    with httpx.Client() as client:
+        response = client.post(
+            f"https://api.github.com/app/installations/{installation_id}/access_tokens",
+            headers={
+                "Accept": "application/vnd.github+json",
+                "Authorization": f"Bearer {jwt}",
+                "X-GitHub-Api-Version": "2022-11-28",
+            },
+        )
+        if response.status_code == 200:
+            return None
+
+        installation_token = response.json().get("token", None)
+        return installation_token
+
+    return None
+
+
+def register_by_code(code: str) -> str | None:
+    """Register by code
+
+    Args:
+        code (str): The code returned by GitHub OAuth.
+
+    Returns:
+        str: The user access token.
+    """
+
+    with httpx.Client() as client:
+        response = client.post(
+            "https://github.com/login/oauth/access_token",
+            params={
+                "client_id": os.environ.get("GITHUB_CLIENT_ID"),
+                "client_secret": os.environ.get("GITHUB_CLIENT_SECRET"),
+                "code": code,
+            },
+        )
+        if response.status_code != 200:
+            return None
+
+        access_token = parse_qs(response.text).get("access_token", None)
+        if access_token is not None:
+            return access_token[0]
+
+    return None