mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
refactor(types): narrow RSA key verification and key derivation
This commit is contained in:
jxxghp
@@ -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)
|
||||
"""
|
||||
|
||||
@@ -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` 为准，
|
||||
|
||||
@@ -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 未进入包级覆盖率门禁 |
|
||||
|
||||
|
||||
1
mypy.ini
1
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
71
tests/test_crypto.py
Normal file
71
tests/test_crypto.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user