feat(ai): issue verified application preview decisions
This commit is contained in:
56
server/src/app/core/expense_application_fingerprint_keys.py
Normal file
56
server/src/app/core/expense_application_fingerprint_keys.py
Normal file
@@ -0,0 +1,56 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
from pathlib import Path
|
||||
|
||||
from app.core.config import SERVER_DIR
|
||||
|
||||
FINGERPRINT_KEY_DIRECTORY = SERVER_DIR / ".secrets" / "expense-application-fingerprints"
|
||||
ACTIVE_FINGERPRINT_KEY_VERSION = "v1"
|
||||
KEY_BYTES = 32
|
||||
_VERSION_PATTERN = re.compile(r"^[a-zA-Z0-9._-]{1,32}$")
|
||||
|
||||
|
||||
def get_expense_application_fingerprint_key(
|
||||
version: str,
|
||||
*,
|
||||
create: bool,
|
||||
) -> bytes:
|
||||
normalized_version = str(version or "").strip()
|
||||
if not _VERSION_PATTERN.fullmatch(normalized_version):
|
||||
raise ValueError("费用申请指纹密钥版本无效。")
|
||||
key_path = FINGERPRINT_KEY_DIRECTORY / f"{normalized_version}.key"
|
||||
if not key_path.exists():
|
||||
if not create:
|
||||
raise ValueError("费用申请指纹密钥版本不可用,不能核验旧预览。")
|
||||
_create_key_atomically(key_path)
|
||||
if key_path.is_symlink() or not key_path.is_file():
|
||||
raise ValueError("费用申请指纹密钥文件无效。")
|
||||
os.chmod(key_path, 0o600)
|
||||
encoded = key_path.read_text(encoding="utf-8").strip()
|
||||
try:
|
||||
key = base64.urlsafe_b64decode(encoded.encode("ascii"))
|
||||
except (binascii.Error, ValueError, UnicodeError) as error:
|
||||
raise ValueError("费用申请指纹密钥内容无效。") from error
|
||||
if len(key) != KEY_BYTES:
|
||||
raise ValueError("费用申请指纹密钥长度无效。")
|
||||
return key
|
||||
|
||||
|
||||
def _create_key_atomically(key_path: Path) -> None:
|
||||
key_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
os.chmod(key_path.parent, 0o700)
|
||||
encoded = base64.urlsafe_b64encode(secrets.token_bytes(KEY_BYTES))
|
||||
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
||||
try:
|
||||
descriptor = os.open(key_path, flags, 0o600)
|
||||
except FileExistsError:
|
||||
return
|
||||
with os.fdopen(descriptor, "wb") as stream:
|
||||
stream.write(encoded)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
Reference in New Issue
Block a user