"""登录认证。"""
import os
import hashlib
import logging
import secrets
from functools import wraps
from flask import Blueprint, request, session, redirect, url_for, current_app
log = logging.getLogger("socks.auth")
bp = Blueprint("auth", __name__)
def _get_admin_creds():
_ENV = os.environ.get("SM_ENV_PATH", os.path.join(os.path.dirname(__file__), ".env"))
pwd = os.environ.get("SM_ADMIN_PASSWORD")
user = os.environ.get("SM_ADMIN_USER", "admin")
if not pwd and os.path.exists(_ENV):
try:
for line in open(_ENV):
line = line.strip()
if line and not line.startswith("#"):
k, _, v = line.partition("=")
if k.strip() == "SM_ADMIN_PASSWORD":
pwd = v.strip()
elif k.strip() == "SM_ADMIN_USER":
user = v.strip()
except Exception:
pass
return (user, pwd or "")
def get_or_set_password():
"""启动时检查密码，如果为空则设置默认值并打印提示。"""
import os
u, p = _get_admin_creds()
if not p or not p.strip():
default_pwd = "admin123"
os.environ["SM_ADMIN_PASSWORD"] = default_pwd
os.environ["SM_ADMIN_USER"] = u
_ENV = os.environ.get("SM_ENV_PATH", os.path.join(os.path.dirname(__file__), ".env"))
try:
with open(_ENV, "w") as f:
f.write(f"SM_ADMIN_USER={u}\nSM_ADMIN_PASSWORD={default_pwd}\n")
print(f"⚠️ 密码未设置，已使用默认密码: {default_pwd}")
print(f" 配置文件: {_ENV}")
print(f" 请立即修改密码！")
except Exception as e:
print(f"⚠️ 密码未设置，默认: {default_pwd} (无法写入 .env: {e})")
return (u, default_pwd)
return (u, p)
def get_admin_password_hash():
"""获取管理员密码哈希（用于安全比较）。"""
_, p = _get_admin_creds()
return hashlib.sha256(p.encode("utf-8")).hexdigest()
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not session.get("logged_in"):
if request.path.startswith("/api/"):
return {"error": "unauthorized"}, 401
return redirect(url_for("auth.login", next=request.url))
return f(*args, **kwargs)
return decorated
@bp.route("/login", methods=["GET", "POST"])
def login():
if session.get("logged_in"):
return redirect(url_for("web.index"))
if request.method == "POST":
u, p = _get_admin_creds()
form_user = request.form.get("username", "").strip()
form_pwd = request.form.get("password", "")
# 防爆破检查
client_ip = request.remote_addr or "unknown"
from services.user_service import UserService
us = UserService()
blocked, remaining = us.check_fail2ban(client_ip)
if blocked:
log.warning("防爆破已阻断来自 %s 的登录尝试", client_ip)
return {"error": "登录过于频繁，请稍后再试"}, 429
# 使用常量时间比较防止时序攻击
if form_user == u and secrets.compare_digest(form_pwd, p):
us.record_auth_attempt(client_ip, success=True)
session.permanent = True
session["logged_in"] = True
session["user"] = u
return redirect(request.args.get("next") or url_for("web.index"))
us.record_auth_attempt(client_ip, success=False)
return {"error": "用户名或密码错误"}, 401
return _render_login()
@bp.route("/logout")
def logout():
session.clear()
return redirect(url_for("auth.login"))
def _render_login():
return """
登录 · SOCKS Manager
SOCKS Manager
SOCKS5 代理管理平台
"""