From 8a39ece6789fe9888ad0e6bfdececba666056fb2 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 16 Oct 2026 12:37:19 +0800 Subject: [PATCH] refactor(types): narrow RSA key verification and key derivation --- app/foundation/crypto.py | 20 ++++-- docs/code-quality.md | 2 +- docs/refactor/optimization-checklist.md | 4 +- mypy.ini | 1 + .../fixtures/architecture/mypy-baseline.json | 5 -- tests/test_crypto.py | 71 +++++++++++++++++++ 6 files changed, 89 insertions(+), 14 deletions(-) create mode 100644 tests/test_crypto.py diff --git a/app/foundation/crypto.py b/app/foundation/crypto.py index f4e3581d7..ef18ea408 100644 --- a/app/foundation/crypto.py +++ b/app/foundation/crypto.py @@ -65,15 +65,23 @@ class RSAUtils: private_key_bytes = base64.b64decode(private_key) # 加载公钥 - public_key = serialization.load_der_public_key(public_key_bytes, backend=default_backend()) + loaded_public_key = serialization.load_der_public_key( + public_key_bytes, backend=default_backend(), + ) # 加载私钥 - private_key = serialization.load_der_private_key(private_key_bytes, password=None, - backend=default_backend()) + loaded_private_key = serialization.load_der_private_key( + private_key_bytes, password=None, backend=default_backend(), + ) + + if not isinstance(loaded_public_key, rsa.RSAPublicKey): + return False + if not isinstance(loaded_private_key, rsa.RSAPrivateKey): + return False # 测试加解密 message = b'test' - encrypted_message = public_key.encrypt( + encrypted_message = loaded_public_key.encrypt( message, asym_padding.OAEP( mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), @@ -82,7 +90,7 @@ class RSAUtils: ) ) - decrypted_message = private_key.decrypt( + decrypted_message: bytes = loaded_private_key.decrypt( encrypted_message, asym_padding.OAEP( mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), @@ -143,7 +151,7 @@ class CryptoJsUtils: """兼容 CryptoJS OpenSSL 格式的 AES 加解密工具。""" @staticmethod - def bytes_to_key(data: bytes, salt: bytes, output=48) -> bytes: + def bytes_to_key(data: bytes, salt: bytes, output: int = 48) -> bytes: """ 生成加密/解密所需的密钥和初始化向量 (IV) """ diff --git a/docs/code-quality.md b/docs/code-quality.md index 70e4d6e23..9ca679484 100644 --- a/docs/code-quality.md +++ b/docs/code-quality.md @@ -75,7 +75,7 @@ uv run --locked --no-sync python scripts/architecture/complexity.py 回调和返回值类型；全仓错误计数为零并不保证单独纳入 strict frontier 后也为零。 2026-10-16 开始逐文件清理类型债务，`app/domain/title.py` 通过局部可空类型声明保留 -既有搜索解析行为，并纳入严格清单；当前清单共 65 个文件（64 个生产文件和 1 个脚本）。 +既有搜索解析行为，并纳入严格清单；当前清单共 66 个文件（65 个生产文件和 1 个脚本）。 每个治理提交聚焦一个生产文件及其必要测试、门禁配置、真实下降基线和指标文档。 验收同时检查完整严格清单与全宿主 Linux 分析，避免 `follow_imports = skip` 隐藏依赖 签名冲突；不通过忽略诊断或放宽类型合同接受回退。最新文件清单以 `mypy.ini` 为准， diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 9ccc5b109..ad5c8821e 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -106,9 +106,9 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | | Module Contract | 215 specs / 211 methods / 263 calls | 动态方法名为 0；分类事实富化通过显式合同进入插件调度，旧 YAML 分类方法已移除；download 新增 custom_words 可选参数，旧 provider 签名通过显式合同兼容 | | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | -| Python 源码量 | 372,035 行 | 本次重新扫描物理行数，排除 `app/plugins/**`；74 个文件超过 1,000 行，12 个超过 2,000 行；源码尺寸仅作观察 | +| Python 源码量 | 372,043 行 | 本次重新扫描物理行数，排除 `app/plugins/**`；74 个文件超过 1,000 行，12 个超过 2,000 行；源码尺寸仅作观察 | | 长方法 | 309 个超过 80 行 | 本次 AST 扫描排除 `app/plugins/**`；61 个超过 150 行，17 个超过 250 行；新增函数仍受 C901/嵌套深度门禁约束 | -| 全量 mypy 历史债务 | 9,034 / 487 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项；Transmission 进度回退值归一化再减少 2 项，逐文件逐错误码无增长 | +| 全量 mypy 历史债务 | 9,030 / 486 文件 | 通知排版补齐 Slack 消息块类型，历史债务减少 2 项；工作流资源过滤改用领域对象调用过滤链，恢复的节点输出在消费前归一化，并收紧可空资源列表的类型低水位；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降；微信 ClawBot 入站 item_list 媒体解析改为按子项键名分派后，类型错误减少 4 项；站点解析器分享率改用浮点初值，类型错误减少 20 项；本次下载与文件安全修复再减少 3 项；Transmission 进度回退值归一化再减少 2 项，逐文件逐错误码无增长 | | Ruff 历史诊断 | 100 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | diff --git a/mypy.ini b/mypy.ini index 15015d74e..95e0e747b 100644 --- a/mypy.ini +++ b/mypy.ini @@ -17,6 +17,7 @@ files = app/foundation/url.py, app/foundation/reflection.py, app/foundation/environment.py, + app/foundation/crypto.py, app/foundation/singleton.py, app/adapters/network/cloudflare.py, app/adapters/network/ip.py, diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index c2ed43c37..7fb88d1e2 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -1265,11 +1265,6 @@ "app/factory.py": { "misc": 1 }, - "app/foundation/crypto.py": { - "no-any-return": 1, - "no-untyped-def": 1, - "union-attr": 2 - }, "app/foundation/dom.py": { "assignment": 2, "no-untyped-def": 5 diff --git a/tests/test_crypto.py b/tests/test_crypto.py new file mode 100644 index 000000000..eac3b9951 --- /dev/null +++ b/tests/test_crypto.py @@ -0,0 +1,71 @@ +"""RSA 密钥校验的算法边界及失败返回合同。""" + +import base64 +from itertools import product +from typing import Optional + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa + +from app.foundation.crypto import RSAUtils + + +@pytest.fixture(scope="module") +def key_pairs() -> dict[str, tuple[str, str]]: + """创建两对 RSA 和两种非 RSA 密钥，覆盖算法与身份不匹配的情况。""" + private_keys = { + "rsa_first": rsa.generate_private_key(public_exponent=65537, key_size=1024), + "rsa_second": rsa.generate_private_key(public_exponent=65537, key_size=1024), + "ec": ec.generate_private_key(ec.SECP256R1()), + "ed25519": ed25519.Ed25519PrivateKey.generate(), + } + result = {} + for name, private_key in private_keys.items(): + private_der = private_key.private_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + public_der = private_key.public_key().public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + result[name] = ( + base64.b64encode(private_der).decode("ascii"), + base64.b64encode(public_der).decode("ascii"), + ) + return result + + +@pytest.mark.parametrize( + ("private_name", "public_name"), + list(product(("rsa_first", "rsa_second", "ec", "ed25519"), repeat=2)), +) +def test_verify_rsa_keys_requires_matching_rsa_pair(key_pairs, private_name, public_name) -> None: + """只有对应的 RSA 公私钥通过；错配或非 RSA 算法均返回 False。""" + private_key = key_pairs[private_name][0] + public_key = key_pairs[public_name][1] + expected = private_name == public_name and private_name.startswith("rsa_") + + assert RSAUtils.verify_rsa_keys(private_key, public_key) is expected + + +@pytest.mark.parametrize( + ("private_key", "public_key"), + [(None, None), ("", ""), ("invalid", None), (None, "invalid"), ("invalid", "invalid")], +) +def test_verify_rsa_keys_rejects_empty_or_invalid_data( + private_key: Optional[str], public_key: Optional[str], +) -> None: + """缺少密钥或解码失败时保留 False 返回，不向调用方泄漏解析异常。""" + assert RSAUtils.verify_rsa_keys(private_key, public_key) is False + + +@pytest.mark.parametrize("invalid", ["invalid", ""]) +def test_verify_rsa_keys_rejects_one_damaged_key(key_pairs, invalid: str) -> None: + """任一侧单独损坏时都应失败，即使另一侧仍是可解析的 RSA 密钥。""" + private_key, public_key = key_pairs["rsa_first"] + + assert RSAUtils.verify_rsa_keys(invalid, public_key) is False + assert RSAUtils.verify_rsa_keys(private_key, invalid) is False