zhaolei
16f134e328
核心功能: - 6 层检测管线: L0 结构分析 → L1 ClamAV → L2 文档深度检测 → L3 关键词匹配 → L4 YARA → L5 PE/二进制 - FastAPI + PostgreSQL + Redis + Docker Compose 架构 - 客户端 HMAC-SHA256 签名鉴权 + 限流 + 幂等性 - 控制台 UI (登录/Dashboard/诊断中心/策略管理/任务列表/测试中心/产品介绍页) - 三层诊断定时器 (error_scan 30s / db_aggregate 30s / container_check 5min) 并发与稳定性修复: - 诊断定时器 Redis 时间槽位锁 (多 worker 同一槽位只有一个执行) - JSONL 存储 fcntl 跨进程排他锁 + 容错解码 + 原子重写 - 移除 SQLite fallback, 统一 PostgreSQL - API_CLIENTS 脱敏为环境变量, .env 已 .gitignore
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""密码学基础工具：密钥派生、口令哈希、常数时间比较。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import os
|
||
from typing import Tuple
|
||
|
||
# 口令哈希迭代次数。等保三级要求存储口令不可逆，PBKDF2-HMAC-SHA256
|
||
# 迭代 60 万次在现代 CPU 上约 0.3~0.6 秒，兼顾抗暴力破解与可用性。
|
||
PBKDF2_ITERATIONS = 600_000
|
||
PBKDF2_ALGORITHM = "sha256"
|
||
SALT_BYTES = 16
|
||
|
||
|
||
def derive_key(secret: str, salt: str, info: str, length: int = 32) -> bytes:
|
||
"""使用 HKDF-SHA256 从调用方密钥派生用途隔离的子密钥。
|
||
|
||
不同用途使用不同 ``info`` 标签，使得请求签名密钥与响应签名密钥相互独立。
|
||
"""
|
||
prk = hmac.new(
|
||
salt.encode("utf-8"), secret.encode("utf-8"), hashlib.sha256
|
||
).digest()
|
||
okm = b""
|
||
block = b""
|
||
counter = 1
|
||
while len(okm) < length:
|
||
block = hmac.new(prk, block + info.encode("utf-8") + bytes([counter]), hashlib.sha256).digest()
|
||
okm += block
|
||
counter += 1
|
||
return okm[:length]
|
||
|
||
|
||
def derive_client_signing_key(secret: str, client_id: str) -> bytes:
|
||
return derive_key(secret, client_id, "meshsafe/api/v1/request-signing")
|
||
|
||
|
||
def derive_client_response_key(secret: str, client_id: str) -> bytes:
|
||
return derive_key(secret, client_id, "meshsafe/api/v1/response-signing")
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
"""生成 ``pbkdf2_sha256$迭代次数$盐$哈希`` 格式的口令哈希。"""
|
||
salt = os.urandom(SALT_BYTES)
|
||
dk = hashlib.pbkdf2_hmac(
|
||
PBKDF2_ALGORITHM, password.encode("utf-8"), salt, PBKDF2_ITERATIONS
|
||
)
|
||
return "$".join(
|
||
[
|
||
"pbkdf2_sha256",
|
||
str(PBKDF2_ITERATIONS),
|
||
base64.b64encode(salt).decode("ascii"),
|
||
base64.b64encode(dk).decode("ascii"),
|
||
]
|
||
)
|
||
|
||
|
||
def verify_password(password: str, stored: str) -> bool:
|
||
"""校验口令。格式非法时返回 False，不抛出异常。"""
|
||
try:
|
||
algo, iterations, salt_b64, hash_b64 = stored.split("$")
|
||
if algo != "pbkdf2_sha256":
|
||
return False
|
||
salt = base64.b64decode(salt_b64)
|
||
expected = base64.b64decode(hash_b64)
|
||
actual = hashlib.pbkdf2_hmac(
|
||
PBKDF2_ALGORITHM, password.encode("utf-8"), salt, int(iterations)
|
||
)
|
||
return hmac.compare_digest(actual, expected)
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
|
||
def needs_rehash(stored: str) -> bool:
|
||
"""判断存量口令哈希是否需要按当前参数重新计算。"""
|
||
try:
|
||
algo, iterations, _, _ = stored.split("$")
|
||
return algo != "pbkdf2_sha256" or int(iterations) < PBKDF2_ITERATIONS
|
||
except Exception: # noqa: BLE001
|
||
return True
|
||
|
||
|
||
def sha256_bytes(data: bytes) -> str:
|
||
return hashlib.sha256(data).hexdigest()
|
||
|
||
|
||
def sha256_file(path: str, chunk_size: int = 1024 * 1024) -> Tuple[str, int]:
|
||
"""流式计算文件 SHA256 与 MD5，返回 (sha256, md5, size)。"""
|
||
sha = hashlib.sha256()
|
||
md5 = hashlib.md5()
|
||
size = 0
|
||
with open(path, "rb") as fh:
|
||
while True:
|
||
chunk = fh.read(chunk_size)
|
||
if not chunk:
|
||
break
|
||
sha.update(chunk)
|
||
md5.update(chunk)
|
||
size += len(chunk)
|
||
return sha.hexdigest(), md5.hexdigest(), size
|
||
|
||
|
||
def constant_time_equals(a: str, b: str) -> bool:
|
||
return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8"))
|