91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
import base64
|
|
import uuid
|
|
import os
|
|
from typing import Optional, Dict, Any
|
|
from app.models.user import get_db_connection
|
|
from app.config import settings
|
|
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "sonicforge_secret_key_super_secure_2026")
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""
|
|
Hash password using PBKDF2 HMAC SHA-256 with salt.
|
|
Guarantees raw passwords are NEVER stored or exposed in plaintext.
|
|
"""
|
|
salt = b"sonicforge_crypto_salt_2026_secure_"
|
|
key = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, 100000)
|
|
return key.hex()
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""Verify plain password against PBKDF2 hashed password using constant-time comparison."""
|
|
computed_hash = hash_password(plain_password)
|
|
return hmac.compare_digest(computed_hash, hashed_password)
|
|
|
|
def create_token(user_id: str, username: str, role: str, must_change_password: bool) -> str:
|
|
payload = {
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"role": role,
|
|
"must_change_password": bool(must_change_password),
|
|
"exp": time.time() + (3600 * 24 * 7) # 7 days
|
|
}
|
|
payload_str = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8")
|
|
sig = hmac.new(SECRET_KEY.encode("utf-8"), payload_str.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
return f"{payload_str}.{sig}"
|
|
|
|
def decode_token(token: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
parts = token.split(".")
|
|
if len(parts) != 2:
|
|
return None
|
|
payload_str, sig = parts[0], parts[1]
|
|
expected_sig = hmac.new(SECRET_KEY.encode("utf-8"), payload_str.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(sig, expected_sig):
|
|
return None
|
|
|
|
payload_bytes = base64.b64decode(payload_str.encode("utf-8"))
|
|
payload = json.loads(payload_bytes.decode("utf-8"))
|
|
if time.time() > payload.get("exp", 0):
|
|
return None
|
|
return payload
|
|
except Exception:
|
|
return None
|
|
|
|
def seed_admin():
|
|
"""Seed default admin account on initial launch if not exists or update password hash if outdated."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
default_pwd = os.getenv("DEFAULT_ADMIN_PASSWORD", "admin123").strip()
|
|
hashed_pwd = hash_password(default_pwd)
|
|
now = time.time()
|
|
|
|
cursor.execute("SELECT id, hashed_password, must_change_password FROM users WHERE username = ?", ("admin",))
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
admin_id = str(uuid.uuid4())
|
|
cursor.execute("""
|
|
INSERT INTO users (id, username, email, hashed_password, role, must_change_password, created_at, is_active)
|
|
VALUES (?, ?, ?, ?, ?, 1, ?, 1)
|
|
""", (admin_id, "admin", "admin@sonicforge.studio", hashed_pwd, "admin", now))
|
|
|
|
cursor.execute("""
|
|
INSERT INTO user_quotas (user_id, storage_limit_mb, max_tracks)
|
|
VALUES (?, 10240, 64)
|
|
""", (admin_id,))
|
|
conn.commit()
|
|
else:
|
|
# Guarantee admin account password hash matches default_pwd if must_change_password is true or hash doesn't match
|
|
if row["must_change_password"] or not verify_password(default_pwd, row["hashed_password"]):
|
|
cursor.execute("UPDATE users SET hashed_password = ? WHERE id = ?", (hashed_pwd, row["id"]))
|
|
conn.commit()
|
|
|
|
conn.close()
|
|
|
|
# Auto seed on module load
|
|
seed_admin()
|