host
25d9b3d938
- github_app：纯标准库 RS256 签 App JWT（PKCS#1/PKCS#8）， 换 installation token（现换现用、不缓存），按仓库名查安装 记录，任一步失败回落 None - pr：app_token 路径 push 走 credential helper（token 只进 子进程环境），建 PR 走 REST API（读 default_branch 当 base）； 不传 token 时 gh 路径不变 - server：/pr 端点按项目 repo_url 解析仓库名、现换 token， 响应加 auth_via=app|pat，换失败回落 PAT 不报错 - 测试：JWT/换 token/解析 9 个、pr App 路径 6 个、端点级 App 与 PAT 回落 2 个
245 lines
10 KiB
Python
245 lines
10 KiB
Python
"""App-token 单测：纯标准库 RS256 签名、JWT、installation token
|
||
换取、仓库安装查找、resolve 回落。
|
||
|
||
fixture 是测试专用 RSA key（openssl 现生成、仅供单测，从不用
|
||
于生产；即使公开也无害——它不对应任何真实 GitHub App）。
|
||
不联网：换 token 挂 fake http 客户端。
|
||
"""
|
||
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import sys
|
||
import tempfile
|
||
import os
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.github_app import (
|
||
AppError,
|
||
_parse_rsa_private_key,
|
||
exchange_installation_token,
|
||
find_installation_for_repo,
|
||
mint_app_jwt,
|
||
record_installation_repos,
|
||
resolve_app_token,
|
||
rsa_sign_sha256,
|
||
save_app_config,
|
||
)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.github_app import (
|
||
AppError,
|
||
_parse_rsa_private_key,
|
||
exchange_installation_token,
|
||
find_installation_for_repo,
|
||
mint_app_jwt,
|
||
record_installation_repos,
|
||
resolve_app_token,
|
||
rsa_sign_sha256,
|
||
save_app_config,
|
||
)
|
||
|
||
# 测试专用 RSA 私钥（PKCS#1 与 PKCS#8，同一密钥对的两版编码
|
||
# 是两把不同 key——各测各的格式解析，不要求同模）。
|
||
TEST_KEY_PKCS1 = """
|
||
-----BEGIN RSA PRIVATE KEY-----
|
||
MIIEogIBAAKCAQEA2M/hYSQQAJxQfcLxcRob7cq4miMLbyWv+Diq4qppENU8XvNS
|
||
7p/JSc8XqWRcGX6FusObKjpskakEL1+tf2wBYGW5jbtgiHzg4J46p+iK/OFi9t+8
|
||
O6wjva3+5kWrjvkQIKL/pbGfdvVQjq1j5bg2fbXKbQqb7CbbhxDhRrW3xVt5A+PF
|
||
G3+iP0qHcBvWuN1CIUHVvUORSdkY44+8xCs+RjalMqVsOyPzTuMu+ypfPtaHfcGM
|
||
DWwkvpXUwq9VrszfrGqi3qWJ6nBQy20eXRUeEnorDS7Rolj+/bILhtbUUT/wFzX7
|
||
F/er1zquAQSntgiMUutB50zpPefNe5Jf/uknZwIDAQABAoIBAEK9izVp8nsdozFz
|
||
ZvtX/Czw8G6SRxzJzydr2j64EmuaGGFqKadHDKFp3HDKcgdvCR96wzL2lSEplB45
|
||
J6FH0Ayqg7n7ajXZqye5zXiOfuplPx+bs8fyr3XeL88Wtz7lsG87fdK1oM7vqQKi
|
||
oVO/pxZ/dXang2Poej9TlOU9XvqjIoTo4HNckRhEF7MJV+2nQB/d+edlSt9oLrN4
|
||
UxOEBeTEomj+ujJBWFOasOcztRXMl/cQp3PdSZ5Ap4aXElblK5/MocM+tyh6RU++
|
||
hLVBpxxjH2CQbG+HMn0rxn7NbtpBdYU3GTjBdKJUv0MQ9m8s39AFfX4h1YW7PCte
|
||
uO0tyfECgYEA+pwlgTWnQL5dlwkw+qsg4mhiG0XcDFsfrh6W/TcsUJnfQIGZCJnC
|
||
9NxCJEmfQTxgJWEmTvPj+Mi4iFmoMYlfTo1OqI36pXNnH0fmh4l0l9COAUsarQVJ
|
||
DzDh5LaCpOn8uSfgJSzdoBY4aYkxJfl8AVsApCmHBojp9Xjw1gis1csCgYEA3Xmk
|
||
qm1BeGkUU/nGDCjz6AuIYEbWuj7OwnV/EsvsGsMTDryoWsk0ZTI0KJdyI/RhjMdM
|
||
jnv37EOy8YL1NjD01NinQ7Ac1wY1DSUrA+jJ6ctSkjkPFHKVBmiC2c5TJiTXWxd6
|
||
Ms7fGQesIrdSdbr7+GrkogKCJyKEwskcRNsFIVUCgYAVqKkVIa3yyoEMLJ8vCtss
|
||
thjEgqUh7D4wrjQebkZGMQY6bPjs5V0UDHUq7bH10WH92QeWiz2IotBP+G+w2E6E
|
||
W3IQ3zRqWC7xZSVnUVniRQQA2JzVlz9ZMMBVI/Dmj9JnriqRC++bhlAZzuGQhp9Q
|
||
/qxvg6dHyU1vCaTG2rkVQQKBgGrkviwfhVznYo9RW1wdG/OgNVc6eYL0s3g5m3Zo
|
||
A/G6u2T+e0cuG2LIQ1JgHg/qcHaw7mtKvBNFbYWRTma0ZEaitsA9rP2/GjXY/ZTq
|
||
cyGrJ5Q5g+S7lN+964HSSmF7oJFvmGQTYqxC/rrWvi7ia6CvfuAO0+CtkWQuCr3U
|
||
8/ntAoGAawnmD9G1JVUSo1ONbV2gsg/Lv6SSq0dbWja4B0fhAR4BWtSFQV8ePTwk
|
||
Q5bSZc/zdNHcKPIMr29Aa3TTy7jXHkAs2frAwKGRyErt/cVAbah+BLOWxniTNfgD
|
||
6CvUI8gTLJ+Dklmer2o6b0JZEeSxz93xx7sy9E5SqHGStOZRutA=
|
||
-----END RSA PRIVATE KEY-----
|
||
"""
|
||
|
||
TEST_KEY_PKCS8 = """
|
||
-----BEGIN PRIVATE KEY-----
|
||
MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCgshr1g7WzZ/iF
|
||
tXbL+R2tTkzOsKVWeLli7Xe1+5W9jz+mEd/zWl/AD/QuM0ncNNkOQfyCVKgtzX4p
|
||
k2yerf4QtdbwvVP36aqs6D0wH1AqkNfC338bbN1eNPNtZ2sG8muAppv4hjUBT7ck
|
||
p3hrQ9VU596PFloYlM6M91mkIWGX76oig+pPSL62P82OwYbM3dBKIJsQaAEWbqf6
|
||
V08D4MOUubt8zcS3FIpztLKQ+NPrzWGoPQoaHx0cruDZLWu2RdUTav4S3G0BLJFh
|
||
TzekF6KGIUC/CdB6uOKFGobJxr9Pz/ZhiM/3PE3KBZXWeNS/+lerV2JB3MAAL2zc
|
||
oPTHbUyXAgMBAAECggEADSAETVmeItKoUlIhgK8kCjkzKHigxdYjcjM/dq40q0sY
|
||
IZoj0WO9dARO5zfKMxcKmXxbQaaPe/o6RpljeExxdsTwvYYBOzurHzydbsudR8oV
|
||
IrYQKC8f9Odcm0Akq5MWTmeNUA200PAatH7h5NpMAjIpaUTOmWO3a9JyJb086Y9e
|
||
Tv6Em0iNjZiyfuTz8q3BRcuZvwdXN3ZJVgTqJTN4iUfKmSbhDieIKHw1NieYbimD
|
||
/LNyt6OMcZiU5RKvujY+kGLJBhJ5DO3/yl7+DfHeQjHrnyrddB2O8IwqBZp6wcmS
|
||
5zTGX3vqclMsZCv2mKoVdWA3QlbDc36N7wumr9nYPQKBgQDbNsFGAaQ26bEsp37H
|
||
ixegd9vhqeO8qFuTDMqNiKVs6t0ZV8PUmI0pB8ZzXxtWRiolUXqxZiqvilzhRz4m
|
||
B/1lYKmljpSJdL4fX+t65MNW4pY7S/bqd/5Rt4j+DA77Ez9lcbVH9LwpWrhMXFvU
|
||
+qyFVglNq7vrDN82FHkrDVf3dQKBgQC7qXWE2hDMZ7CerNQu06GSLehQ65YBtRg/
|
||
oQ62dcpEv7vxVn24fzwtMBFlmt3WY+N9Vr8O9dxuiX7PmBOIf1P3BW+qLU+bVExc
|
||
AQychjKoHRsimkqDaX8XqjJkA7fD56zahLKLShKwmUdWoNvM9Vf8jokO7Ky6/Ckk
|
||
8DR1YRI+WwKBgCJ3OWwIVglMNM8Qhp5ZAB36uleAno3K6pr0J+SY12mIg7q4UdHb
|
||
KlaJImqBSYsG/Rlg9iluJsfyuSeyS9Mu7mhCCmo9QJ1b/YhxTFTGIvBGL9ii2O+E
|
||
bjfq418OBkzzecQ5XIczs2HNIazf2JBj2TwditjrZ2BNZcra7TIXpv0BAoGAFhyP
|
||
dknjdNI641dl4OaawQibuP0pPaZC5yxo+WLDpHobAQK56YVEZ+akBh7LSLmYUmpo
|
||
NLEPeg7FXFyE4K5wDEkdsSWJnbm/DZAqFj9RtLFJQzh+0yEZEobBTTuK6ZBgs5az
|
||
GuptbVWCLFjjzTCTX3TD/Xgxmg6gInQTGkmHxsUCgYB0p7w195Q+/fUhdCSnvwNQ
|
||
VlrX5EBjvjkPZFQkXNCFoeIhQuADyQQJ7P9HHlrxSjHsbZOxTTcVQatm0kk8s6Et
|
||
LAxPD9wtwVxzZHHVXCK+l8JNfBmiW14n6BQvkTlGgt1HFojrx4ubz86VwgP8FMuW
|
||
Qq9ELI2S82qfw0VlsoZWxQ==
|
||
-----END PRIVATE KEY-----
|
||
"""
|
||
|
||
|
||
|
||
def _b64url_decode(s):
|
||
return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
|
||
|
||
|
||
def _emsa_sha256(message, em_len):
|
||
prefix = bytes.fromhex("3031300d060960864801650304020105000420")
|
||
t = prefix + hashlib.sha256(message).digest()
|
||
return b"\x00\x01" + b"\xff" * (em_len - len(t) - 3) + b"\x00" + t
|
||
|
||
|
||
def _rsa_verify(pem, message, signature):
|
||
"""纯 Python RSA 公钥验签（与签名实现独立的验算）。"""
|
||
n, e, _d = _parse_rsa_private_key(pem)
|
||
em_len = (n.bit_length() + 7) // 8
|
||
m = pow(int.from_bytes(signature, "big"), e, n)
|
||
return m.to_bytes(em_len, "big") == _emsa_sha256(message, em_len)
|
||
|
||
|
||
class Rs256TestCase(unittest.TestCase):
|
||
def test_sign_and_verify_pkcs1(self):
|
||
message = b"github app jwt signing input"
|
||
sig = rsa_sign_sha256(TEST_KEY_PKCS1, message)
|
||
n, _e, _d = _parse_rsa_private_key(TEST_KEY_PKCS1)
|
||
self.assertEqual(len(sig), (n.bit_length() + 7) // 8)
|
||
self.assertTrue(_rsa_verify(TEST_KEY_PKCS1, message, sig))
|
||
|
||
def test_sign_and_verify_pkcs8(self):
|
||
message = b"another signing input"
|
||
sig = rsa_sign_sha256(TEST_KEY_PKCS8, message)
|
||
self.assertTrue(_rsa_verify(TEST_KEY_PKCS8, message, sig))
|
||
self.assertFalse(_rsa_verify(TEST_KEY_PKCS8, b"tampered", sig))
|
||
|
||
def test_bad_key(self):
|
||
with self.assertRaises(AppError):
|
||
rsa_sign_sha256("not a key", b"x")
|
||
with self.assertRaises(AppError):
|
||
rsa_sign_sha256("-----BEGIN FOO-----\nAAAA\n-----END FOO-----",
|
||
b"x")
|
||
|
||
|
||
class JwtTestCase(unittest.TestCase):
|
||
def test_mint_structure_and_signature(self):
|
||
jwt = mint_app_jwt("5476837", TEST_KEY_PKCS1, now=1000000)
|
||
parts = jwt.split(".")
|
||
self.assertEqual(len(parts), 3)
|
||
header = json.loads(_b64url_decode(parts[0]))
|
||
self.assertEqual(header["alg"], "RS256")
|
||
payload = json.loads(_b64url_decode(parts[1]))
|
||
self.assertEqual(payload["iss"], "5476837")
|
||
self.assertEqual(payload["iat"], 1000000 - 60)
|
||
self.assertEqual(payload["exp"], 1000000 + 540)
|
||
signing_input = (parts[0] + "." + parts[1]).encode("ascii")
|
||
self.assertTrue(_rsa_verify(
|
||
TEST_KEY_PKCS1, signing_input, _b64url_decode(parts[2])))
|
||
|
||
|
||
def _fake_exchange(expected_installation="175380082"):
|
||
def fake(method, url, headers, body):
|
||
assert method == "POST", method
|
||
assert "/app/installations/%s/access_tokens" % expected_installation in url, url
|
||
assert headers["Authorization"].startswith("Bearer "), headers
|
||
assert headers["Authorization"].count(".") == 2
|
||
return 201, {"token": "fake-installation-token",
|
||
"expires_at": "2030-01-01T00:00:00Z",
|
||
"permissions": {"contents": "write"},
|
||
"repository_selection": "selected"}
|
||
return fake
|
||
|
||
|
||
class TokenExchangeTestCase(unittest.TestCase):
|
||
def test_exchange_ok(self):
|
||
result = exchange_installation_token(
|
||
"5476837", TEST_KEY_PKCS1, "175380082",
|
||
http_client=_fake_exchange())
|
||
self.assertEqual(result["token"], "fake-installation-token")
|
||
self.assertEqual(result["repository_selection"], "selected")
|
||
|
||
def test_exchange_failure_raises(self):
|
||
def fake(method, url, headers, body):
|
||
return 401, {"message": "Bad credentials"}
|
||
with self.assertRaises(AppError) as ctx:
|
||
exchange_installation_token(
|
||
"5476837", TEST_KEY_PKCS1, "175380082", http_client=fake)
|
||
self.assertEqual(ctx.exception.reason, "token_exchange_failed")
|
||
|
||
|
||
class ResolveTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.conn = _db_connect(self.db_path)
|
||
|
||
def tearDown(self):
|
||
self.conn.close()
|
||
self.tmp.cleanup()
|
||
|
||
def _save_config(self):
|
||
save_app_config(self.conn, {
|
||
"app_id": 5476837, "slug": "fake-app", "client_id": "Iv1.x",
|
||
"webhook_secret": "fake-secret", "pem": TEST_KEY_PKCS1,
|
||
"html_url": "https://github.com/apps/fake-app"})
|
||
|
||
def test_resolve_full_chain(self):
|
||
self._save_config()
|
||
record_installation_repos(self.conn, 175380082, "octo",
|
||
["octo/a"])
|
||
token = resolve_app_token(self.conn, "octo/a",
|
||
http_client=_fake_exchange())
|
||
self.assertEqual(token, "fake-installation-token")
|
||
|
||
def test_resolve_falls_back_to_none(self):
|
||
# 无配置、无安装、换 token 失败：一律 None（调用方走 PAT）。
|
||
self.assertIsNone(resolve_app_token(
|
||
self.conn, "octo/a", http_client=_fake_exchange()))
|
||
self._save_config()
|
||
self.assertIsNone(resolve_app_token(
|
||
self.conn, "octo/a", http_client=_fake_exchange()))
|
||
record_installation_repos(self.conn, 175380082, "octo",
|
||
["octo/a"])
|
||
|
||
def failing(method, url, headers, body):
|
||
return 401, {"message": "Bad credentials"}
|
||
self.assertIsNone(resolve_app_token(
|
||
self.conn, "octo/a", http_client=failing))
|
||
|
||
def test_find_installation(self):
|
||
record_installation_repos(self.conn, 11, "octo", ["octo/a"])
|
||
self.assertEqual(find_installation_for_repo(self.conn, "octo/a"),
|
||
"11")
|
||
self.assertIsNone(find_installation_for_repo(self.conn, "octo/b"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|