Initial certificate system

backend/app/core/security.py

@@ -0,0 +1,71 @@
+import base64
+import hashlib
+import hmac
+import json
+import secrets
+import time
+from typing import Any
+
+from app.core.config import settings
+
+
+def hash_password(password: str, salt: str | None = None) -> str:
+    password_salt = salt or secrets.token_hex(16)
+    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), password_salt.encode("utf-8"), 120_000)
+    return f"pbkdf2_sha256${password_salt}${digest.hex()}"
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    try:
+        scheme, salt, expected = password_hash.split("$", 2)
+    except ValueError:
+        return False
+    if scheme != "pbkdf2_sha256":
+        return False
+    actual = hash_password(password, salt).split("$", 2)[2]
+    return hmac.compare_digest(actual, expected)
+
+
+def _b64encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+def _b64decode(data: str) -> bytes:
+    padding = "=" * (-len(data) % 4)
+    return base64.urlsafe_b64decode((data + padding).encode("ascii"))
+
+
+def create_access_token(subject: str, role_code: str, expires_in: int = 60 * 60 * 8) -> str:
+    header = {"alg": "HS256", "typ": "JWT"}
+    payload = {"sub": subject, "role": role_code, "exp": int(time.time()) + expires_in}
+    signing_input = ".".join(
+        [
+            _b64encode(json.dumps(header, separators=(",", ":")).encode("utf-8")),
+            _b64encode(json.dumps(payload, separators=(",", ":")).encode("utf-8")),
+        ]
+    )
+    signature = hmac.new(settings.secret_key.encode("utf-8"), signing_input.encode("ascii"), hashlib.sha256).digest()
+    return f"{signing_input}.{_b64encode(signature)}"
+
+
+def decode_access_token(token: str) -> dict[str, Any] | None:
+    try:
+        header_b64, payload_b64, signature_b64 = token.split(".", 2)
+        signing_input = f"{header_b64}.{payload_b64}"
+        expected = hmac.new(settings.secret_key.encode("utf-8"), signing_input.encode("ascii"), hashlib.sha256).digest()
+        if not hmac.compare_digest(_b64encode(expected), signature_b64):
+            return None
+        payload = json.loads(_b64decode(payload_b64))
+        if int(payload.get("exp", 0)) < int(time.time()):
+            return None
+        return payload
+    except Exception:
+        return None
+
+
+def hash_token(token: str) -> str:
+    return hashlib.sha256(f"{settings.secret_key}:{token}".encode("utf-8")).hexdigest()
+
+
+def generate_public_token() -> str:
+    return secrets.token_urlsafe(32)