74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
from cryptography.fernet import Fernet
|
|
|
|
|
|
def _fernet_key(secret: str) -> bytes:
|
|
digest = hashlib.sha256(secret.encode("utf-8")).digest()
|
|
return base64.urlsafe_b64encode(digest)
|
|
|
|
|
|
class Crypto:
|
|
def __init__(self, secret: str) -> None:
|
|
self._secret = secret.encode("utf-8")
|
|
self._fernet = Fernet(_fernet_key(secret))
|
|
|
|
def encrypt(self, value: str) -> str:
|
|
return self._fernet.encrypt(value.encode("utf-8")).decode("ascii")
|
|
|
|
def decrypt(self, value: str) -> str:
|
|
return self._fernet.decrypt(value.encode("ascii")).decode("utf-8")
|
|
|
|
def hash_password(self, password: str) -> str:
|
|
salt = os.urandom(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 210_000)
|
|
return "pbkdf2_sha256${}${}".format(
|
|
base64.urlsafe_b64encode(salt).decode("ascii"),
|
|
base64.urlsafe_b64encode(digest).decode("ascii"),
|
|
)
|
|
|
|
def verify_password(self, password: str, encoded: str) -> bool:
|
|
try:
|
|
scheme, salt_s, digest_s = encoded.split("$", 2)
|
|
except ValueError:
|
|
return False
|
|
if scheme != "pbkdf2_sha256":
|
|
return False
|
|
salt = base64.urlsafe_b64decode(salt_s.encode("ascii"))
|
|
expected = base64.urlsafe_b64decode(digest_s.encode("ascii"))
|
|
actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 210_000)
|
|
return hmac.compare_digest(expected, actual)
|
|
|
|
def sign_session(self, payload: dict[str, Any], ttl_seconds: int) -> str:
|
|
body = dict(payload)
|
|
body["exp"] = int(time.time()) + ttl_seconds
|
|
encoded = base64.urlsafe_b64encode(
|
|
json.dumps(body, separators=(",", ":"), sort_keys=True).encode("utf-8")
|
|
).decode("ascii")
|
|
sig = hmac.new(self._secret, encoded.encode("ascii"), hashlib.sha256).hexdigest()
|
|
return f"{encoded}.{sig}"
|
|
|
|
def verify_session(self, token: str) -> dict[str, Any] | None:
|
|
try:
|
|
encoded, sig = token.split(".", 1)
|
|
except ValueError:
|
|
return None
|
|
expected = hmac.new(self._secret, encoded.encode("ascii"), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(expected, sig):
|
|
return None
|
|
try:
|
|
payload = json.loads(base64.urlsafe_b64decode(encoded.encode("ascii")))
|
|
except (ValueError, json.JSONDecodeError):
|
|
return None
|
|
if int(payload.get("exp", 0)) < int(time.time()):
|
|
return None
|
|
return payload if isinstance(payload, dict) else None
|