63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""数据加密与口令散列工具。
|
||
|
||
- DataEncryptor：AES-256-GCM（依赖 cryptography），密钥由主密钥经 HKDF 派生
|
||
- hash_password / verify_password：PBKDF2-HMAC-SHA256（stdlib）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import os
|
||
|
||
try:
|
||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
_HAS_CRYPTO = True
|
||
except ImportError: # pragma: no cover
|
||
_HAS_CRYPTO = False
|
||
|
||
|
||
def hash_password(password: str, *, iterations: int = 120_000, salt: bytes | None = None) -> str:
|
||
"""PBKDF2-HMAC-SHA256 口令散列，输出 `pbkdf2$iterations$salt_b64$hash_b64`"""
|
||
salt = salt or os.urandom(16)
|
||
dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations)
|
||
return "pbkdf2${}${}${}".format(
|
||
iterations, base64.b64encode(salt).decode(), base64.b64encode(dk).decode())
|
||
|
||
|
||
def verify_password(password: str, stored: str) -> bool:
|
||
try:
|
||
scheme, iters, salt_b64, hash_b64 = stored.split("$")
|
||
if scheme != "pbkdf2":
|
||
return False
|
||
dk = hashlib.pbkdf2_hmac("sha256", password.encode(),
|
||
base64.b64decode(salt_b64), int(iters))
|
||
return hmac.compare_digest(dk, base64.b64decode(hash_b64))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
class DataEncryptor:
|
||
"""AES-256-GCM 字段级加密器（用于手机号/身份证等敏感字段落库加密）。"""
|
||
|
||
def __init__(self, master_key: bytes, *, info: bytes = b"wms-field-encryption"):
|
||
if not _HAS_CRYPTO:
|
||
raise RuntimeError("需要安装 cryptography 库以使用 AES-256-GCM 加密")
|
||
if len(master_key) < 16:
|
||
raise ValueError("master_key 长度至少 16 字节")
|
||
# HKDF-SHA256 派生 32 字节数据密钥
|
||
prk = hmac.new(b"wms-hkdf-salt", master_key, hashlib.sha256).digest()
|
||
self._key = hmac.new(prk, info + b"\x01", hashlib.sha256).digest()
|
||
self._aes = AESGCM(self._key)
|
||
|
||
def encrypt(self, plaintext: str) -> str:
|
||
nonce = os.urandom(12)
|
||
ct = self._aes.encrypt(nonce, plaintext.encode("utf-8"), None)
|
||
return base64.b64encode(nonce + ct).decode("ascii")
|
||
|
||
def decrypt(self, token: str) -> str:
|
||
raw = base64.b64decode(token)
|
||
nonce, ct = raw[:12], raw[12:]
|
||
return self._aes.decrypt(nonce, ct, None).decode("utf-8")
|