"""登录认证 + 权限系统。
v2 版本：后台管理员账户存在数据库（AdminUser 表），支持多账户、角色权限。
首次启动时，如果 AdminUser 表为空，会从 SM_ADMIN_USER / SM_ADMIN_PASSWORD
环境变量（或 .env）创建初始超级管理员。
"""
import os
import logging
import secrets
from datetime import datetime, timezone
from functools import wraps
from flask import Blueprint, request, session, redirect, url_for, current_app, abort
log = logging.getLogger("socks.auth")
bp = Blueprint("auth", __name__)
# ── 工具函数 ----------------------------------------------------------------
def _read_env_creds():
"""从环境变量或 .env 文件读取初始管理员凭据。"""
_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 seed_initial_admin():
"""确保至少存在一个超级管理员。
- 如果 admin_users 表为空：从环境变量/.env 创建初始 superadmin
- 如果表非空：什么都不做
返回创建的 admin 对象或 None。
"""
from models import AdminUser
from database import db
if AdminUser.query.count() > 0:
return None
user, pwd = _read_env_creds()
if not pwd or not pwd.strip():
pwd = "admin123"
log.warning("SM_ADMIN_PASSWORD 未设置，使用默认密码 admin123")
admin = AdminUser()
admin.username = user or "admin"
admin.role = "superadmin"
admin.enabled = True
admin.set_password(pwd)
db.session.add(admin)
db.session.commit()
log.info("已创建初始管理员: %s (role=superadmin)", admin.username)
return admin
def get_or_set_password():
"""兼容旧版启动流程（run.py 调用）。
现在做的是：确保有初始管理员存在。
"""
admin = seed_initial_admin()
if admin:
return (admin.username, "(见数据库)")
# 至少返回一个用户名用于显示
u, p = _read_env_creds()
return (u, p or "")
# ── 当前登录用户 ------------------------------------------------------------
def current_admin():
"""返回当前登录的 AdminUser 对象；未登录返回 None。"""
from models import AdminUser
uid = session.get("admin_id")
if not uid:
return None
return AdminUser.query.get(uid)
def login_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not session.get("logged_in") or not session.get("admin_id"):
if request.path.startswith("/api/"):
return {"error": "unauthorized"}, 401
return redirect(url_for("auth.login", next=request.url))
return f(*args, **kwargs)
return decorated
def permission_required(perm):
"""权限装饰器：需要指定权限位才能访问。"""
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
admin = current_admin()
if not admin:
if request.path.startswith("/api/"):
return {"error": "unauthorized"}, 401
return redirect(url_for("auth.login", next=request.url))
if not admin.enabled:
session.clear()
return {"error": "账户已被禁用"}, 403
if not admin.has_permission(perm):
if request.path.startswith("/api/"):
return {"error": "forbidden"}, 403
abort(403)
return f(*args, **kwargs)
return decorated
return decorator
# ── 登录 / 登出 -------------------------------------------------------------
@bp.route("/login", methods=["GET", "POST"])
def login():
if session.get("logged_in") and session.get("admin_id"):
return redirect(url_for("web.index"))
if request.method == "POST":
from models import AdminUser
from database import db
form_user = request.form.get("username", "").strip()
form_pwd = request.form.get("password", "")
# 防爆破
client_ip = request.remote_addr or "unknown"
_fail2ban_record(client_ip, check_only=True)
if _is_banned(client_ip):
log.warning("防爆破已阻断来自 %s 的登录尝试", client_ip)
return {"error": "登录过于频繁，请稍后再试"}, 429
admin = AdminUser.query.filter_by(username=form_user).first()
if admin and admin.enabled and admin.check_password(form_pwd):
# 登录成功
admin.last_login_at = datetime.now(timezone.utc)
admin.last_login_ip = client_ip
db.session.commit()
_reset_fail2ban(client_ip)
session.permanent = True
session["logged_in"] = True
session["admin_id"] = admin.id
session["user"] = admin.username
session["role"] = admin.role
return redirect(request.args.get("next") or url_for("web.index"))
# 登录失败
_record_fail2ban(client_ip)
return {"error": "用户名或密码错误"}, 401
return _render_login()
@bp.route("/logout")
def logout():
session.clear()
return redirect(url_for("auth.login"))
# ── 防爆破（进程内简易实现） ------------------------------------------------
# 注意：多 worker 下每个进程独立，生产环境建议用 Redis 替代。
_fail_attempts = {} # {ip: [timestamps...]}
_ban_until = {} # {ip: timestamp}
_BAN_WINDOW = 600 # 10 分钟窗口
_BAN_MAX = 10 # 最多 10 次失败
_BAN_DURATION = 900 # 封禁 15 分钟
def _is_banned(ip):
now = _now()
if ip in _ban_until and _ban_until[ip] > now:
return True
if ip in _ban_until:
del _ban_until[ip]
return False
def _fail2ban_record(ip, check_only=False):
"""检查 + 清理过期记录（兼容旧版 UserService.check_fail2ban 调用方式）。"""
now = _now()
if ip in _fail_attempts:
_fail_attempts[ip] = [t for t in _fail_attempts[ip] if now - t < _BAN_WINDOW]
return _is_banned(ip), _BAN_MAX - len(_fail_attempts.get(ip, []))
def _record_fail2ban(ip):
now = _now()
if ip not in _fail_attempts:
_fail_attempts[ip] = []
_fail_attempts[ip].append(now)
_fail_attempts[ip] = [t for t in _fail_attempts[ip] if now - t < _BAN_WINDOW]
if len(_fail_attempts[ip]) >= _BAN_MAX:
_ban_until[ip] = now + _BAN_DURATION
log.warning("IP %s 登录失败 %d 次，已封禁 %d 秒",
ip, _BAN_MAX, _BAN_DURATION)
def _reset_fail2ban(ip):
_fail_attempts.pop(ip, None)
_ban_until.pop(ip, None)
def _now():
import time
return time.time()
# ── 登录页 -----------------------------------------------------------------
def _render_login():
return """
登录 · SOCKS Manager
SOCKS Manager
SOCKS5 代理管理平台
"""