75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def get_aes_key() -> bytes:
|
|
"""Derive a 32-byte AES key from the configured encryption key."""
|
|
return hashlib.sha256(settings.ENCRYPTION_KEY.encode()).digest()
|
|
|
|
|
|
def encrypt_temp_token(record_id: str, expires_in: int = 3600) -> str:
|
|
"""Create an encrypted token containing record_id and expiry timestamp."""
|
|
payload = json.dumps({"rid": record_id, "exp": int(time.time()) + expires_in})
|
|
aesgcm = AESGCM(get_aes_key())
|
|
nonce = AESGCM.generate_key(bit_length=96)
|
|
ciphertext = aesgcm.encrypt(nonce, payload.encode(), None)
|
|
token_bytes = nonce + ciphertext
|
|
return base64.urlsafe_b64encode(token_bytes).decode()
|
|
|
|
|
|
def decrypt_temp_token(token: str) -> str | None:
|
|
"""Decrypt token and return record_id if valid and not expired."""
|
|
try:
|
|
token_bytes = base64.urlsafe_b64decode(token)
|
|
nonce = token_bytes[:12]
|
|
ciphertext = token_bytes[12:]
|
|
aesgcm = AESGCM(get_aes_key())
|
|
payload = aesgcm.decrypt(nonce, ciphertext, None)
|
|
data = json.loads(payload)
|
|
if data["exp"] < time.time():
|
|
return None
|
|
return data["rid"]
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def encrypt_text(plaintext: str) -> str:
|
|
"""使用 AES-256-GCM 加密字符串，返回 base64 编码的密文。"""
|
|
import os
|
|
aesgcm = AESGCM(get_aes_key())
|
|
nonce = os.urandom(12) # 96-bit nonce for GCM
|
|
ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None)
|
|
return base64.urlsafe_b64encode(nonce + ciphertext).decode()
|
|
|
|
|
|
def decrypt_text(token: str) -> str | None:
|
|
"""解密 AES-256-GCM 加密的字符串。失败返回 None。"""
|
|
try:
|
|
token_bytes = base64.urlsafe_b64decode(token)
|
|
nonce = token_bytes[:12]
|
|
ciphertext = token_bytes[12:]
|
|
aesgcm = AESGCM(get_aes_key())
|
|
return aesgcm.decrypt(nonce, ciphertext, None).decode()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def hmac_sign(data: str) -> str:
|
|
"""Create HMAC-SHA256 signature."""
|
|
return hmac.new(
|
|
settings.SECRET_KEY.encode(), data.encode(), hashlib.sha256
|
|
).hexdigest()
|
|
|
|
|
|
def hmac_verify(data: str, signature: str) -> bool:
|
|
"""Verify HMAC-SHA256 signature."""
|
|
expected = hmac_sign(data)
|
|
return hmac.compare_digest(expected, signature)
|