129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
import base64
|
|
import uuid
|
|
import os
|
|
import secrets
|
|
from typing import Optional, Dict, Any
|
|
from app.models.user import get_db_connection
|
|
from app.config import settings
|
|
|
|
COOKIE_NAME = "sf_token"
|
|
X_AUTH_HEADER = "X-Auth-Token"
|
|
|
|
def _load_or_create_secret_key() -> str:
|
|
"""Persistent random SECRET_KEY.
|
|
|
|
Priority: env SECRET_KEY > {STORAGE_DIR}/.secret_key (auto-generated on
|
|
first run). Never falls back to a hardcoded value: a known secret lets
|
|
anyone forge admin tokens.
|
|
"""
|
|
env_key = os.getenv("SECRET_KEY", "").strip()
|
|
if env_key:
|
|
return env_key
|
|
key_file = os.path.join(settings.STORAGE_DIR, ".secret_key")
|
|
try:
|
|
os.makedirs(settings.STORAGE_DIR, exist_ok=True)
|
|
if os.path.exists(key_file):
|
|
with open(key_file, "r") as f:
|
|
key = f.read().strip()
|
|
if len(key) >= 32:
|
|
return key
|
|
key = secrets.token_hex(32)
|
|
with open(key_file, "w") as f:
|
|
f.write(key)
|
|
return key
|
|
except Exception:
|
|
# Last resort: ephemeral random key (all tokens invalid on restart).
|
|
return secrets.token_hex(32)
|
|
|
|
SECRET_KEY = _load_or_create_secret_key()
|
|
|
|
def hash_password(password: str, salt: Optional[str] = None) -> str:
|
|
"""
|
|
Hash password using PBKDF2 HMAC SHA-256 with a per-user random salt.
|
|
Guarantees raw passwords are NEVER stored or exposed in plaintext.
|
|
Returns format: 'salt_hex:hashed_key_hex' (colon-delimited, stores both values).
|
|
If salt is provided, uses that salt (for verification).
|
|
If salt is None, generates a new 32-byte random salt.
|
|
"""
|
|
if salt is None:
|
|
salt = secrets.token_hex(32)
|
|
key = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt.encode('utf-8'), 600000)
|
|
return f"{salt}:{key.hex()}"
|
|
|
|
def verify_password(plain_password: str, stored_value: str) -> bool:
|
|
"""Verify plain password against stored 'salt:hash' using constant-time comparison.
|
|
Also supports legacy-format hash (hex only, without salt) for backward compatibility."""
|
|
if ':' in stored_value:
|
|
parts = stored_value.split(':', 1)
|
|
salt = parts[0]
|
|
expected_hash = parts[1]
|
|
computed = hash_password(plain_password, salt)
|
|
computed_hash = computed.split(':', 1)[1]
|
|
return hmac.compare_digest(computed_hash, expected_hash)
|
|
else:
|
|
expected_hash = stored_value
|
|
computed_hash = hashlib.pbkdf2_hmac('sha256', plain_password.encode('utf-8'), b"sonicforge_crypto_salt_2026_secure_", 600000).hex()
|
|
return hmac.compare_digest(computed_hash, expected_hash)
|
|
|
|
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 only — never resets existing password."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
default_pwd = (os.getenv("DEFAULT_ADMIN_PASSWORD") or "admin123").strip()
|
|
hashed_pwd = hash_password(default_pwd)
|
|
now = time.time()
|
|
|
|
cursor.execute("SELECT id FROM users WHERE username = ?", ("admin",))
|
|
if not cursor.fetchone():
|
|
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()
|
|
|
|
conn.close()
|
|
|
|
# Auto seed on module load
|
|
seed_admin()
|