235 lines
8.1 KiB
Python
235 lines
8.1 KiB
Python
import json
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
|
||
from cli.exit_codes import ARGUMENT_ERROR, RUNTIME_ERROR
|
||
|
||
from .gateway import CryptoInfrastructureGateway
|
||
|
||
MIN_RSA_KEY_SIZE = 2048
|
||
RSA_KEY_SIZE_STEP = 256
|
||
|
||
|
||
class CryptoDomainSupport:
|
||
"""
|
||
传输加密领域支持对象。
|
||
|
||
该对象负责 RSA 密钥规则校验、密钥对生成和历史密钥规整，
|
||
避免主运行时服务继续承载局部安全规则和数据拼装细节。
|
||
|
||
:param infrastructure_gateway: 传输加密基础设施网关
|
||
"""
|
||
|
||
def __init__(self, infrastructure_gateway: CryptoInfrastructureGateway) -> None:
|
||
"""
|
||
初始化传输加密领域支持对象。
|
||
|
||
:param infrastructure_gateway: 传输加密基础设施网关
|
||
:return: None
|
||
"""
|
||
self.infrastructure_gateway = infrastructure_gateway
|
||
|
||
@staticmethod
|
||
def validate_rsa_key_size(key_size: int) -> None:
|
||
"""
|
||
校验 RSA 密钥长度是否合法。
|
||
|
||
:param key_size: RSA 密钥长度
|
||
:return: None
|
||
:raises ValueError: 密钥长度不满足约束时抛出异常
|
||
"""
|
||
if key_size < MIN_RSA_KEY_SIZE or key_size % RSA_KEY_SIZE_STEP != 0:
|
||
raise ValueError('RSA 密钥长度必须大于等于 2048，且为 256 的整数倍')
|
||
|
||
def generate_rsa_key_pair(self, key_size: int) -> tuple[str, str]:
|
||
"""
|
||
生成 PEM 格式的 RSA 密钥对。
|
||
|
||
:param key_size: RSA 密钥长度
|
||
:return: 私钥与公钥 PEM 字符串
|
||
"""
|
||
rsa_module = self.infrastructure_gateway.get_rsa_module()
|
||
serialization_module = self.infrastructure_gateway.get_serialization_module()
|
||
private_key = rsa_module.generate_private_key(public_exponent=65537, key_size=key_size)
|
||
private_key_pem = private_key.private_bytes(
|
||
encoding=serialization_module.Encoding.PEM,
|
||
format=serialization_module.PrivateFormat.PKCS8,
|
||
encryption_algorithm=serialization_module.NoEncryption(),
|
||
).decode('utf-8')
|
||
public_key_pem = (
|
||
private_key.public_key()
|
||
.public_bytes(
|
||
encoding=serialization_module.Encoding.PEM,
|
||
format=serialization_module.PublicFormat.SubjectPublicKeyInfo,
|
||
)
|
||
.decode('utf-8')
|
||
)
|
||
return private_key_pem, public_key_pem
|
||
|
||
def load_existing_legacy_key_pairs(self) -> list[dict[str, str]]:
|
||
"""
|
||
解析当前配置中的历史密钥列表。
|
||
|
||
:return: 历史密钥列表
|
||
:raises ValueError: 历史密钥配置非法时抛出异常
|
||
"""
|
||
transport_crypto_config = self.infrastructure_gateway.get_transport_crypto_config()
|
||
raw_legacy_key_pairs = transport_crypto_config.transport_crypto_legacy_key_pairs or '[]'
|
||
parsed_legacy_key_pairs = json.loads(raw_legacy_key_pairs)
|
||
if not isinstance(parsed_legacy_key_pairs, list):
|
||
raise ValueError('TRANSPORT_CRYPTO_LEGACY_KEY_PAIRS 必须为 JSON 数组')
|
||
return [item for item in parsed_legacy_key_pairs if isinstance(item, dict)]
|
||
|
||
@staticmethod
|
||
def merge_rotation_legacy_key_pairs(
|
||
legacy_key_pairs: list[dict[str, str]],
|
||
current_key_pair: Any,
|
||
next_kid: str,
|
||
) -> list[dict[str, str]]:
|
||
"""
|
||
合并轮换后的历史密钥列表。
|
||
|
||
:param legacy_key_pairs: 当前历史密钥列表
|
||
:param current_key_pair: 当前生效密钥对对象
|
||
:param next_kid: 即将切换的新密钥版本
|
||
:return: 合并后的历史密钥列表
|
||
"""
|
||
normalized_legacy_key_pairs: dict[str, dict[str, str]] = {
|
||
str(item.get('kid')): item for item in legacy_key_pairs if item.get('kid')
|
||
}
|
||
normalized_legacy_key_pairs[current_key_pair.kid] = {
|
||
'kid': current_key_pair.kid,
|
||
'privateKey': current_key_pair.private_key_pem,
|
||
'publicKey': current_key_pair.public_key_pem,
|
||
}
|
||
normalized_legacy_key_pairs.pop(next_kid, None)
|
||
return list(normalized_legacy_key_pairs.values())
|
||
|
||
@staticmethod
|
||
def build_key_pair_env_patch(kid: str, private_key_pem: str, public_key_pem: str) -> dict[str, str]:
|
||
"""
|
||
构建密钥对写回环境变量所需的补丁数据。
|
||
|
||
:param kid: 密钥版本标识
|
||
:param private_key_pem: 私钥 PEM 文本
|
||
:param public_key_pem: 公钥 PEM 文本
|
||
:return: 环境变量补丁数据
|
||
"""
|
||
return {
|
||
'TRANSPORT_CRYPTO_KID': kid,
|
||
'TRANSPORT_CRYPTO_PRIVATE_KEY': private_key_pem,
|
||
'TRANSPORT_CRYPTO_PUBLIC_KEY': public_key_pem,
|
||
}
|
||
|
||
def build_rotation_env_patch(
|
||
self,
|
||
next_kid: str,
|
||
private_key_pem: str,
|
||
public_key_pem: str,
|
||
merged_legacy_key_pairs: list[dict[str, str]],
|
||
) -> dict[str, str]:
|
||
"""
|
||
构建密钥轮换写回环境变量所需的补丁数据。
|
||
|
||
:param next_kid: 新密钥版本标识
|
||
:param private_key_pem: 新私钥 PEM 文本
|
||
:param public_key_pem: 新公钥 PEM 文本
|
||
:param merged_legacy_key_pairs: 合并后的历史密钥列表
|
||
:return: 环境变量补丁数据
|
||
"""
|
||
env_patch = self.build_key_pair_env_patch(next_kid, private_key_pem, public_key_pem)
|
||
env_patch['TRANSPORT_CRYPTO_LEGACY_KEY_PAIRS'] = json.dumps(merged_legacy_key_pairs, ensure_ascii=False)
|
||
return env_patch
|
||
|
||
|
||
class CryptoResultSupport:
|
||
"""
|
||
传输加密结果支持对象。
|
||
|
||
该对象负责参数错误/运行时错误结果翻译，以及固定轮换提示文案构建，
|
||
避免主运行时服务继续承载重复异常映射与静态 payload 片段。
|
||
"""
|
||
|
||
@staticmethod
|
||
def build_argument_error_result(message: str, exc: Exception) -> dict[str, Any]:
|
||
"""
|
||
构建参数错误结果。
|
||
|
||
:param message: 失败消息
|
||
:param exc: 原始异常
|
||
:return: 标准参数错误结果
|
||
"""
|
||
return {
|
||
'ok': False,
|
||
'message': message,
|
||
'error': str(exc),
|
||
'exit_code': ARGUMENT_ERROR,
|
||
}
|
||
|
||
@staticmethod
|
||
def build_runtime_error_result(message: str, exc: Exception) -> dict[str, Any]:
|
||
"""
|
||
构建运行时错误结果。
|
||
|
||
:param message: 失败消息
|
||
:param exc: 原始异常
|
||
:return: 标准运行时错误结果
|
||
"""
|
||
return {
|
||
'ok': False,
|
||
'message': message,
|
||
'error': str(exc),
|
||
'exit_code': RUNTIME_ERROR,
|
||
}
|
||
|
||
def run_argument_guarded(
|
||
self,
|
||
operation: Callable[[], dict[str, Any]],
|
||
*,
|
||
failure_message: str,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
以统一参数/运行时错误翻译执行操作。
|
||
|
||
:param operation: 待执行操作
|
||
:param failure_message: 失败消息
|
||
:return: 操作结果
|
||
"""
|
||
try:
|
||
return operation()
|
||
except ValueError as exc:
|
||
return self.build_argument_error_result(failure_message, exc)
|
||
except Exception as exc:
|
||
return self.build_runtime_error_result(failure_message, exc)
|
||
|
||
def run_runtime_guarded(
|
||
self,
|
||
operation: Callable[[], dict[str, Any]],
|
||
*,
|
||
failure_message: str,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
以统一运行时错误翻译执行操作。
|
||
|
||
:param operation: 待执行操作
|
||
:param failure_message: 失败消息
|
||
:return: 操作结果
|
||
"""
|
||
try:
|
||
return operation()
|
||
except Exception as exc:
|
||
return self.build_runtime_error_result(failure_message, exc)
|
||
|
||
@staticmethod
|
||
def build_rotation_apply_steps() -> list[str]:
|
||
"""
|
||
构建密钥轮换后的应用步骤说明。
|
||
|
||
:return: 轮换应用步骤列表
|
||
"""
|
||
return [
|
||
'将 envPatch 中的新密钥写入目标环境配置',
|
||
'发布后保留 legacyKeyPairs，确保旧请求在轮换窗口内可解密',
|
||
'确认前端已刷新到 nextKid 后，再移除不再需要的历史密钥',
|
||
]
|