mirror of
https://github.com/halfwaystudent/douyin-sparkflow.git
synced 2026-08-29 03:57:07 +08:00
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
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
|
|
try:
|
|
algorithm, salt, digest = stored_hash.split("$", 2)
|
|
except ValueError:
|
|
return False
|
|
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, *, role="admin", account_refs=None):
|
|
request.session.clear()
|
|
request.session["user"] = username
|
|
request.session["role"] = role
|
|
request.session["account_refs"] = list(account_refs or [])
|
|
request.session["session_id"] = secrets.token_urlsafe(24)
|
|
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 current_principal(request):
|
|
"""Resolve the current session to live role and account assignments."""
|
|
username = current_user(request)
|
|
if not username:
|
|
return None
|
|
session_id = request.session.get("session_id", "")
|
|
role = request.session.get("role")
|
|
admin_username = str(get_app_settings().get("admin_username", "admin")).strip() or "admin"
|
|
if role == "admin" or (role is None and username.casefold() == admin_username.casefold()):
|
|
return {
|
|
"username": admin_username,
|
|
"role": "admin",
|
|
"account_refs": [],
|
|
"session_id": session_id,
|
|
"enabled": True,
|
|
}
|
|
|
|
try:
|
|
from webui.users import find_web_user
|
|
user = find_web_user(username)
|
|
except Exception:
|
|
user = None
|
|
if user and user.get("enabled", True):
|
|
return {
|
|
"username": user["username"],
|
|
"role": "user",
|
|
"account_refs": list(user.get("account_refs", [])),
|
|
"session_id": session_id,
|
|
"enabled": True,
|
|
}
|
|
return None
|
|
|
|
|
|
def is_admin(request):
|
|
principal = current_principal(request)
|
|
return bool(principal and principal.get("role") == "admin")
|
|
|
|
|
|
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"
|