vlad
86cc3fa541
Align the project baseline with the latest admin interface styling and layout structure while documenting setup and usage updates in README.
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from urllib.parse import urlencode
|
|
|
|
import bcrypt
|
|
from jose import jwt
|
|
|
|
def generate_secret_token_urlsafe(length: int = 32) -> str:
|
|
return secrets.token_urlsafe(length)
|
|
|
|
|
|
def generate_secret_token_hex(length: int = 32) -> str:
|
|
return secrets.token_hex(length)
|
|
|
|
|
|
def generate_install_bundle() -> dict[str, str]:
|
|
postgres_user = "compton_app"
|
|
postgres_password = generate_secret_token_urlsafe(32)
|
|
postgres_db = "compton"
|
|
minio_root_user = "minio"
|
|
minio_root_password = generate_secret_token_urlsafe(32)
|
|
return {
|
|
"POSTGRES_USER": postgres_user,
|
|
"POSTGRES_PASSWORD": postgres_password,
|
|
"POSTGRES_DB": postgres_db,
|
|
"DATABASE_URL": f"postgresql+psycopg://{postgres_user}:{postgres_password}@postgres:5432/{postgres_db}",
|
|
"JWT_ACCESS_SECRET": generate_secret_token_hex(32),
|
|
"JWT_REFRESH_PEPPER": generate_secret_token_hex(32),
|
|
"S3_ACCESS_KEY": minio_root_user,
|
|
"S3_SECRET_KEY": minio_root_password,
|
|
"MINIO_ROOT_USER": minio_root_user,
|
|
"MINIO_ROOT_PASSWORD": minio_root_password,
|
|
}
|
|
|
|
|
|
def hash_password(raw_password: str) -> str:
|
|
return bcrypt.hashpw(raw_password.encode("utf-8"), bcrypt.gensalt(rounds=12)).decode("utf-8")
|
|
|
|
|
|
def verify_password(raw_password: str, password_hash: str) -> bool:
|
|
return bcrypt.checkpw(raw_password.encode("utf-8"), password_hash.encode("utf-8"))
|
|
|
|
|
|
def create_access_token(user_id: str, role: str, is_superuser: bool = False) -> str:
|
|
from app.core.config import settings
|
|
|
|
now = datetime.now(UTC)
|
|
payload = {
|
|
"sub": user_id,
|
|
"role": role,
|
|
"is_superuser": is_superuser,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int((now + timedelta(minutes=settings.jwt_access_ttl_min)).timestamp()),
|
|
"jti": generate_secret_token_hex(16),
|
|
}
|
|
return jwt.encode(payload, settings.jwt_access_secret, algorithm="HS256")
|
|
|
|
|
|
def decode_access_token(token: str) -> dict:
|
|
from app.core.config import settings
|
|
|
|
return jwt.decode(token, settings.jwt_access_secret, algorithms=["HS256"])
|
|
|
|
|
|
def generate_refresh_token() -> str:
|
|
return generate_secret_token_urlsafe(48)
|
|
|
|
|
|
def generate_opaque_token() -> str:
|
|
return generate_secret_token_urlsafe(32)
|
|
|
|
|
|
def hash_opaque_token(token: str) -> str:
|
|
return hash_refresh_token(token)
|
|
|
|
|
|
def hash_refresh_token(token: str) -> str:
|
|
from app.core.config import settings
|
|
|
|
return hashlib.sha256(f"{token}:{settings.jwt_refresh_pepper}".encode("utf-8")).hexdigest()
|
|
|
|
|
|
def build_signed_media_url(stored_url: str | None) -> str | None:
|
|
from app.core.config import settings
|
|
|
|
if not stored_url:
|
|
return None
|
|
if not stored_url.startswith("/api/v1/media/files/"):
|
|
return stored_url
|
|
path = stored_url.removeprefix("/api/v1/media/files/")
|
|
expires = int(time.time()) + settings.media_url_ttl_seconds
|
|
signature = _sign_media_path(path, expires)
|
|
query = urlencode({"expires": expires, "sig": signature})
|
|
return f"/api/v1/media/files/{path}?{query}"
|
|
|
|
|
|
def verify_signed_media(path: str, expires: int, signature: str) -> bool:
|
|
if expires < int(time.time()):
|
|
return False
|
|
expected = _sign_media_path(path, expires)
|
|
return hmac.compare_digest(expected, signature)
|
|
|
|
|
|
def _sign_media_path(path: str, expires: int) -> str:
|
|
from app.core.config import settings
|
|
|
|
payload = f"{path}:{expires}"
|
|
return hmac.new(
|
|
settings.jwt_access_secret.encode("utf-8"),
|
|
payload.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|