72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
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)
|