Import sanitized project structure and GitHub docs

DouYinSparkFlow/webui/auth.py

@@ -0,0 +1,73 @@
+import hashlib
+import hmac
+import secrets
+
+from utils.config import get_app_settings, save_app_settings
+
+
+def hash_password(password, salt=None):
+    salt = salt or secrets.token_hex(16)
+    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 480000)
+    return f"pbkdf2_sha256${salt}${digest.hex()}"
+
+
+def verify_password(password, stored_hash):
+    if not stored_hash or "$" not in stored_hash:
+        return False
+
+    algorithm, salt, digest = stored_hash.split("$", 2)
+    if algorithm != "pbkdf2_sha256":
+        return False
+    candidate = hash_password(password, salt=salt)
+    return hmac.compare_digest(candidate, stored_hash)
+
+
+def is_bootstrapped():
+    return bool(get_app_settings().get("admin_password_hash"))
+
+
+def bootstrap_admin_password(password, username="admin"):
+    settings = get_app_settings(force_reload=True)
+    settings["admin_username"] = username.strip() or "admin"
+    settings["admin_password_hash"] = hash_password(password)
+    return save_app_settings(settings)
+
+
+def update_admin_password(password):
+    settings = get_app_settings(force_reload=True)
+    settings["admin_password_hash"] = hash_password(password)
+    return save_app_settings(settings)
+
+
+def issue_session(request, username):
+    request.session.clear()
+    request.session["user"] = username
+    request.session["csrf_token"] = secrets.token_urlsafe(24)
+
+
+def clear_session(request):
+    request.session.clear()
+
+
+def current_user(request):
+    return request.session.get("user")
+
+
+def csrf_token(request):
+    token = request.session.get("csrf_token")
+    if not token:
+        token = secrets.token_urlsafe(24)
+        request.session["csrf_token"] = token
+    return token
+
+
+def validate_csrf(request, submitted_token):
+    stored_token = request.session.get("csrf_token")
+    return bool(stored_token and submitted_token and hmac.compare_digest(stored_token, submitted_token))
+
+
+def is_https_request(request):
+    forwarded_proto = request.headers.get("x-forwarded-proto", "")
+    if forwarded_proto:
+        return forwarded_proto.lower() == "https"
+    return request.url.scheme == "https"