rebuild(stage-3): establish accounts permissions and secure settings

next/backend/security/__init__.py

@@ -0,0 +1,4 @@
+from backend.security.cipher import SecretCipher, load_or_create_cipher
+from backend.security.passwords import PasswordHasher, PasswordPolicyError
+
+__all__ = ["PasswordHasher", "PasswordPolicyError", "SecretCipher", "load_or_create_cipher"]

next/backend/security/cipher.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from cryptography.fernet import Fernet, InvalidToken
+
+from backend.bootstrap.settings import ConfigurationError, Settings
+
+
+@dataclass(frozen=True, slots=True)
+class SecretCipher:
+    _fernet: Fernet
+
+    @classmethod
+    def from_key(cls, key: str) -> SecretCipher:
+        try:
+            return cls(Fernet(key.encode("ascii")))
+        except (ValueError, UnicodeEncodeError) as exc:
+            raise ConfigurationError("APP_ENCRYPTION_KEY is not a valid Fernet key") from exc
+
+    def encrypt(self, value: str) -> str:
+        return self._fernet.encrypt(value.encode("utf-8")).decode("ascii")
+
+    def decrypt(self, value: str) -> str:
+        try:
+            return self._fernet.decrypt(value.encode("ascii")).decode("utf-8")
+        except (InvalidToken, UnicodeDecodeError, UnicodeEncodeError) as exc:
+            raise ValueError("Encrypted value cannot be decrypted") from exc
+
+
+def load_or_create_cipher(settings: Settings) -> SecretCipher:
+    if settings.encryption_key is not None:
+        return SecretCipher.from_key(settings.encryption_key)
+    if settings.environment == "production":
+        raise ConfigurationError("APP_ENCRYPTION_KEY is required in production")
+
+    key_file = settings.data_directory / ".encryption.key"
+    key_file.parent.mkdir(parents=True, exist_ok=True)
+    if key_file.exists():
+        return SecretCipher.from_key(key_file.read_text(encoding="ascii").strip())
+
+    key = Fernet.generate_key().decode("ascii")
+    try:
+        with key_file.open("x", encoding="ascii") as handle:
+            handle.write(key)
+        os.chmod(key_file, 0o600)
+    except FileExistsError:
+        key = key_file.read_text(encoding="ascii").strip()
+    return SecretCipher.from_key(key)

next/backend/security/passwords.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import secrets
+from dataclasses import dataclass
+
+
+class PasswordPolicyError(ValueError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class PasswordHasher:
+    work_factor: int = 2**15
+    block_size: int = 8
+    parallelism: int = 1
+    salt_bytes: int = 16
+    key_bytes: int = 32
+
+    def validate(self, password: str) -> None:
+        if not 8 <= len(password) <= 128:
+            raise PasswordPolicyError("密码长度应为8至128位。")
+        categories = (
+            any(character.isalpha() for character in password),
+            any(character.isdigit() for character in password),
+            any(not character.isalnum() for character in password),
+        )
+        if sum(categories) < 2:
+            raise PasswordPolicyError("密码至少包含字母、数字、符号中的两类。")
+
+    def hash(self, password: str) -> str:
+        self.validate(password)
+        salt = secrets.token_bytes(self.salt_bytes)
+        digest = hashlib.scrypt(
+            password.encode("utf-8"),
+            salt=salt,
+            n=self.work_factor,
+            r=self.block_size,
+            p=self.parallelism,
+            dklen=self.key_bytes,
+            maxmem=64 * 1024 * 1024,
+        )
+        encoded_salt = base64.urlsafe_b64encode(salt).decode("ascii")
+        encoded_digest = base64.urlsafe_b64encode(digest).decode("ascii")
+        return (
+            f"scrypt${self.work_factor}${self.block_size}${self.parallelism}"
+            f"${encoded_salt}${encoded_digest}"
+        )
+
+    def verify(self, password: str, encoded: str) -> bool:
+        try:
+            scheme, work_factor, block_size, parallelism, salt, expected = encoded.split("$")
+            if scheme != "scrypt":
+                return False
+            digest = hashlib.scrypt(
+                password.encode("utf-8"),
+                salt=base64.urlsafe_b64decode(salt.encode("ascii")),
+                n=int(work_factor),
+                r=int(block_size),
+                p=int(parallelism),
+                dklen=self.key_bytes,
+                maxmem=64 * 1024 * 1024,
+            )
+            return hmac.compare_digest(base64.urlsafe_b64encode(digest).decode("ascii"), expected)
+        except (TypeError, ValueError):
+            return False