Eet Cloud Kms Extend Design

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

┌─────────────────────────────────────────────────────────────────────────┐
│                         CLI 命令层 (aes, envelope, rsa, hmac, sm4...)   │
└─────────────────────────────────────────────────────────────────────────┘
                                      │
                    ┌─────────────────┴─────────────────┐
                    │  密钥来源判断：-f/--key  vs  --kms-key-id  │
                    └─────────────────┬─────────────────┘
                                      │
         ┌────────────────────────────┼────────────────────────────┐
         │ 本地密钥路径 (-f/-k)         │         KMS 密钥 ID (--kms-key-id)  │
         ▼                            │                            ▼
┌─────────────────────┐               │               ┌─────────────────────┐
│ common.load_*_key()  │               │               │  kms_backend.       │
│ random_str.*         │               │               │  get_key / decrypt   │
└─────────────────────┘               │               └─────────────────────┘
                                       │
                                       ▼
                         ┌─────────────────────────────┐
                         │  密钥获取层 (KeyResolver)    │
                         │  resolve_symmetric_key()   │
                         │  resolve_public_key()       │
                         │  resolve_dek_for_envelope() │
                         └─────────────────────────────┘
                                       │
                                       ▼
                         ┌─────────────────────────────┐
                         │  密码学计算层 (现有不变)     │
                         │  _aes_gcm_encrypt, etc.     │
                         └─────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

# easy_encryption_tool/kms/protocol.py

from typing import Protocol, runtime_checkable

@runtime_checkable
class KMSBackend(Protocol):
    """云 KMS 后端协议。各云厂商实现此接口。"""

    def get_public_key(self, key_id: str) -> bytes:
        """
        获取非对称密钥公钥（PEM 或 DER）。
        :param key_id: 密钥 ID（如 ARN、资源 ID）
        :return: 公钥字节（PEM 或 SubjectPublicKeyInfo）
        :raises KMSError: 密钥不存在、权限不足等
        """
        ...

    def generate_data_key(
        self,
        key_id: str,
        key_spec: str = "AES_256",
        number_of_bytes: int | None = None,
    ) -> tuple[bytes, bytes]:
        """
        生成数据加密密钥（DEK）。
        :param key_id: 主密钥 ID（KEK）
        :param key_spec: 密钥规格，如 "AES_256"
        :param number_of_bytes: 可选，指定字节数（部分 KMS 用 key_spec）
        :return: (plaintext_dek, encrypted_dek)
        :raises KMSError: 密钥不存在、权限不足等
        """
        ...

    def decrypt(self, key_id: str, ciphertext_blob: bytes) -> bytes:
        """
        解密。用于：1) DEK 解密；2) 非对称 RSA/SM2 解密。
        :param key_id: 密钥 ID
        :param ciphertext_blob: 密文（KMS 返回的格式，含元数据）
        :return: 明文
        :raises KMSError: 解密失败
        """
        ...

    def sign(
        self,
        key_id: str,
        message: bytes,
        message_type: str = "DIGEST",
        algorithm: str | None = None,
    ) -> bytes:
        """
        数字签名。
        :param key_id: 签名密钥 ID
        :param message: 消息或摘要（message_type 区分）
        :param message_type: "DIGEST" | "RAW"
        :param algorithm: 如 "RSASSA_PKCS1_V1_5_SHA_256"
        :return: 签名字节
        """
        ...

    def generate_mac(
        self,
        key_id: str,
        message: bytes,
        algorithm: str = "HMAC_SHA_256",
    ) -> bytes:
        """
        生成 HMAC（若 KMS 支持）。
        :param key_id: HMAC 密钥 ID
        :param message: 消息
        :param algorithm: HMAC 算法
        :return: MAC 字节
        """
        ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

# easy_encryption_tool/kms/resolver.py

from typing import Optional
from dataclasses import dataclass

@dataclass
class SymmetricKeyResult:
    """对称密钥解析结果"""
    key_bytes: bytes
    encrypted_key: Optional[bytes] = None  # 解密时从 KMS 获取的密文 DEK
    kms_key_id: Optional[str] = None      # 解密时需回传 KMS

@dataclass
class PublicKeyResult:
    """公钥解析结果"""
    key_bytes: bytes  # PEM 或 DER
    key_type: str     # "rsa" | "sm2" | "ec"

def resolve_symmetric_key(
    *,
    key: Optional[str] = None,
    key_b64: Optional[str] = None,
    kms_key_id: Optional[str] = None,
    kms_backend: Optional[str] = None,
    kms_region: Optional[str] = None,
    kms_endpoint: Optional[str] = None,
    encrypted_key: Optional[bytes] = None,
    for_encrypt: bool = True,
) -> SymmetricKeyResult:
    """
    解析对称密钥。优先 KMS，否则本地。
    - for_encrypt=True + kms_key_id: 调用 KMS GenerateDataKey
    - for_encrypt=False + kms_key_id + encrypted_key: 调用 KMS Decrypt
    - 否则: 从 key/key_b64 解析（现有逻辑）
    """
    ...

def resolve_public_key(
    *,
    file_path: Optional[str] = None,
    kms_key_id: Optional[str] = None,
    kms_backend: Optional[str] = None,
    kms_region: Optional[str] = None,
    kms_endpoint: Optional[str] = None,
) -> PublicKeyResult:
    """
    解析公钥。kms_key_id 优先，否则从 file_path 加载。
    """
    ...

def resolve_dek_for_envelope_encrypt(
    *,
    public_key_path: Optional[str] = None,
    kms_key_id: Optional[str] = None,
    kms_backend: Optional[str] = None,
    kms_region: Optional[str] = None,
    kms_endpoint: Optional[str] = None,
) -> tuple[bytes, bytes, Optional[str], int]:
    """
    信封加密时获取 DEK。
    - 本地: (dek, enc_dek, None, wrap_key_type)
    - KMS: (dek, enc_dek, kms_key_id, WRAP_KMS)，需 kms_backend、kms_region；kms_endpoint 可选
    返回: (plaintext_dek, encrypted_dek, kms_key_id_or_none, wrap_key_type)
    """
    ...

def resolve_dek_for_envelope_decrypt(
    *,
    private_key_path: Optional[str] = None,
    private_key_password: Optional[bytes] = None,
    kms_key_id: Optional[str] = None,
    kms_backend: Optional[str] = None,
    kms_region: Optional[str] = None,
    kms_endpoint: Optional[str] = None,
    encrypted_dek: bytes,
    wrap_key_type: int,
) -> bytes:
    """
    信封解密时获取 DEK。
    - wrap_key_type=WRAP_RSA: 本地 RSA 解密
    - wrap_key_type=WRAP_KMS: 调用 KMS Decrypt，需 kms_backend、kms_region；kms_endpoint 可选
    """
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# easy_encryption_tool/kms/backends/__init__.py

def get_backend(
    backend_name: str,
    region: str,
    endpoint: str | None = None,
    **kwargs,
) -> KMSBackend:
    """
    根据名称获取 KMS 后端实例。
    :param backend_name: "aws" | "aliyun" | "huawei"
    :param region: 区域（必选）
    :param endpoint: 自定义 endpoint；None 时由各 backend 根据 region 构造默认
    :param kwargs: 各后端其他配置
    """
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# easy_encryption_tool/kms/errors.py

class KMSError(Exception):
    """KMS 相关错误基类"""
    pass

class KMSNotAvailableError(KMSError):
    """KMS 可选依赖未安装"""
    pass

# easy_encryption_tool/kms/availability.py

def is_kms_available() -> bool:
    """检查 KMS 可选依赖是否已安装"""
    try:
        import boto3  # 或 aliyun sdk
        return True
    except ImportError:
        return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 伪代码
parsed = _parse_envelope(raw)  # 与现有一致
if parsed["wrap_key_type"] == WRAP_RSA:
    dek = _rsa_decrypt_dek(pri_key, parsed["enc_dek"])
elif parsed["wrap_key_type"] == WRAP_KMS:
    if not kms_key_id or not kms_backend or not kms_region:
        error("KMS envelope requires --kms-backend, --kms-key-id, --kms-region")
        return
    backend = get_backend(kms_backend, region=kms_region, endpoint=kms_endpoint)
    dek = backend.decrypt(kms_key_id, parsed["enc_dek"])
else:
    error(DECRYPT_FAILED_MSG)
# 后续 AES-GCM 解密、plaintext_binding 校验与现有一致

1
2
3
4
5
6
7
8
9

# 加密：KMS 生成 DEK，密文标准输出，encrypted_dek 单独输出
eet aes encrypt -A encrypt -i "data" \
  --kms-backend aws --kms-key-id arn:aws:kms:cn-north-1:123:key/xxx --kms-region cn-north-1 \
  -o cipher.b64 --encrypted-dek-out dek.b64

# 解密：密文 -i，encrypted_dek、nonce 通过 --encrypted-dek-file、--nonce-file 传入
eet aes decrypt -A decrypt -i cipher.b64 \
  --kms-backend aws --kms-key-id xxx --kms-region cn-north-1 \
  --encrypted-dek-file dek.b64 --nonce-file nonce.b64

1
2
3
4
5
6
7

# 加密：与 RSA 信封用法一致，输出单一 base64 信封
eet envelope encrypt --kms-backend aws --kms-key-id arn:aws:kms:... --kms-region cn-north-1 \
  -i "data" -o envelope.b64

# 解密：-i 为信封，-f 与 KMS 三参数互斥
eet envelope decrypt --kms-backend aws --kms-key-id arn:aws:kms:... --kms-region cn-north-1 \
  -i envelope.b64 -o plain.txt

1
2
3
4
5

# 加密：密文为标准 RSA/SM2 格式，与本地模式一致
eet rsa encrypt --kms-backend aws --kms-key-id arn:aws:kms:... --kms-region cn-north-1 -i "data"

# 解密（调用 KMS API）
eet rsa decrypt --kms-backend aws --kms-key-id arn:aws:kms:... --kms-region cn-north-1 -i <base64_cipher>

1
2
3
4
5
6
7

# 计算：MAC 标准 hex 输出，encrypted_dek 单独输出
eet hmac -i "data" --kms-backend aws --kms-key-id xxx --kms-region cn-north-1 \
  --encrypted-dek-out dek.b64

# 校验：需传入 encrypted_dek 以还原 key
eet hmac verify -i "data" --mac xxx --kms-backend aws --kms-key-id xxx --kms-region cn-north-1 \
  --encrypted-dek-file dek.b64

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 原逻辑：pub_key = _load_rsa_public_key(public_key)
# 新逻辑（仅此段为新增）：

if kms_key_id:
    if public_key:  # -f 与 KMS 参数互斥
        error("Cannot use both -f and --kms-key-id")
        return
    backend = get_backend(kms_backend, region=kms_region, endpoint=kms_endpoint)
    dek, enc_dek, _, wrap_key_type = resolve_dek_for_envelope_encrypt(
        kms_key_id=kms_key_id, kms_backend=kms_backend, kms_region=kms_region, kms_endpoint=kms_endpoint
    )
else:
    pub_key = _load_rsa_public_key(public_key)
    if pub_key is None:
        error("Public key must be RSA 2048/3072/4096 bits")
        return
    dek = random_str.generate_random_bytes(AES_DEK_LEN)
    enc_dek = _rsa_encrypt_dek(pub_key, dek)
    wrap_key_type = WRAP_RSA

# 以下与现有完全一致
nonce = random_str.generate_random_bytes(GCM_NONCE_LEN)
ciphertext = _aes_gcm_encrypt(dek, nonce, aad_bytes, plaintext)
plaintext_binding = hmac.new(dek, plaintext, hashlib.sha256).digest()
envelope = _build_envelope(..., enc_dek=enc_dek, wrap_key_type=wrap_key_type, ...)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 原逻辑：pub_key = common.load_public_key(encoding=encoding, file_path=public_key)
# 新逻辑：

if kms_key_id:
    if public_key:
        error("Cannot use both -f and --kms-key-id")
        return
    backend = get_backend(kms_backend, region=kms_region, endpoint=kms_endpoint)
    pub_key_bytes = backend.get_public_key(kms_key_id)
    pub_key = serialization.load_der_public_key(pub_key_bytes)  # 或 load_pem
else:
    pub_key = common.load_public_key(encoding=encoding, file_path=public_key)

# 以下不变
if pub_key is None:
    return
cipher = pub_key.encrypt(...)

1
2
3
4
5
6
7

def validate_key_source(ctx, param, value):
    kms = ctx.params.get("kms_key_id")
    if value and kms:
        raise click.BadParameter("-f/--public-key and --kms-key-id are mutually exclusive")
    if not value and not kms:
        raise click.BadParameter("Either -f/--public-key or --kms-key-id is required")
    return value

1
2
3
4
5
6
7
8
9

easy_encryption_tool/kms/
├── __init__.py
├── protocol.py
├── resolver.py
├── availability.py
├── errors.py
└── backends/
    ├── __init__.py
    └── aws.py