Your Name
e44802ae40
web form 提交时 expire_at 字段是 HTML datetime-local 字符串 (如 '2027-12-31T23:59'), SQLAlchemy 直接写入 datetime 列会抛 TypeError. 同样, API 路径里如果传 ISO 字符串而不带 tzinfo, 写入后再读出是 naive datetime, is_active() 比较时 会抛 'can't compare offset-naive and offset-aware datetimes'. 修复: - services/user_service.py: 增加 _parse_expire_at() helper, 统一转成 UTC-aware datetime, 支持 datetime-local 'T'/' ' 多种分隔符 + ISO 格式 - services/user_service.py: create_user/update_user 调用 helper - models.py: User.is_active() 兼容 naive datetime (SQLite 不存 tzinfo)
287 lines
11 KiB
Python
287 lines
11 KiB
Python
"""用户服务：认证、流量统计、限速、生命周期。"""
|
|
import logging
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
|
|
from models import User, AuditLog
|
|
from database import db
|
|
|
|
log = logging.getLogger("socks.users")
|
|
|
|
# 防爆破记录: {src_ip: [(timestamp, ...)]}
|
|
_fail2ban = defaultdict(list)
|
|
_FAIL2BAN_MAX = 10
|
|
_FAIL2BAN_WINDOW = 600
|
|
|
|
|
|
def _parse_expire_at(value):
|
|
"""从 HTML form / API JSON 接收 expire_at, 转成 UTC-aware datetime 或 None.
|
|
|
|
支持:
|
|
- None / 空字符串 -> None
|
|
- ISO 字符串 (含 'T' 或 ' ') -> datetime (UTC)
|
|
- 已经是 datetime 的 -> 视为 UTC (如果是 naive 就补 tz)
|
|
- 非法格式 -> None
|
|
"""
|
|
if not value:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
|
s = str(value).strip()
|
|
if not s:
|
|
return None
|
|
# HTML datetime-local 格式 "2026-12-31T23:59" 或 "2026-12-31 23:59"
|
|
for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
|
|
try:
|
|
dt = datetime.strptime(s, fmt)
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
continue
|
|
# ISO 完整格式 (fromisoformat 自动处理 tzinfo)
|
|
try:
|
|
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
|
|
if not dt.tzinfo:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
class UserService:
|
|
"""用户管理服务。"""
|
|
|
|
def __init__(self, db_app=None):
|
|
self.db_app = db_app
|
|
self._active_conn_counts = defaultdict(int)
|
|
|
|
def set_db_app(self, app):
|
|
self.db_app = app
|
|
|
|
# ── 用户 CRUD ──────────────────────────────────────────────
|
|
|
|
def get_user(self, username):
|
|
with db.app.app_context():
|
|
return db.session.query(User).filter_by(username=username).first()
|
|
|
|
def list_users(self, page=1, per_page=20, search=""):
|
|
with db.app.app_context():
|
|
q = db.session.query(User)
|
|
if search:
|
|
q = q.filter(User.username.ilike(f"%{search}%"))
|
|
total = q.count()
|
|
users = q.order_by(User.created_at.desc()).offset(
|
|
(page - 1) * per_page
|
|
).limit(per_page).all()
|
|
return {
|
|
"users": [{
|
|
"id": u.id,
|
|
"username": u.username,
|
|
"enabled": u.enabled,
|
|
"banned": u.banned,
|
|
"bandwidth_down": u.bandwidth_down,
|
|
"bandwidth_up": u.bandwidth_up,
|
|
"max_concurrent": u.max_concurrent,
|
|
"total_traffic_mb": u.total_traffic_mb,
|
|
"monthly_traffic_mb": u.monthly_traffic_mb,
|
|
"bytes_in": u.bytes_in,
|
|
"bytes_out": u.bytes_out,
|
|
"bytes_total_mb": round((u.bytes_in + u.bytes_out) / 1024 / 1024, 1),
|
|
"expire_at": u.expire_at.isoformat() if u.expire_at else None,
|
|
"ip_whitelist": u.ip_whitelist,
|
|
"ip_blacklist": u.ip_blacklist,
|
|
"active": u.is_active(),
|
|
"created_at": u.created_at.isoformat(),
|
|
} for u in users],
|
|
"total": total,
|
|
"page": page,
|
|
"pages": (total + per_page - 1) // per_page,
|
|
}
|
|
|
|
def create_user(self, username, password="", **kwargs):
|
|
# expire_at 在 web form 里是字符串 (datetime-local), 转成 datetime
|
|
if "expire_at" in kwargs:
|
|
kwargs["expire_at"] = _parse_expire_at(kwargs["expire_at"])
|
|
with db.app.app_context():
|
|
if db.session.query(User).filter_by(username=username).first():
|
|
return {"error": "用户名已存在"}
|
|
u = User(username=username, **kwargs)
|
|
if password:
|
|
u.set_password(password)
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
log.info("创建用户: %s", username)
|
|
return {"id": u.id}
|
|
|
|
def update_user(self, uid, **kwargs):
|
|
# expire_at 字符串转 datetime
|
|
if "expire_at" in kwargs:
|
|
kwargs["expire_at"] = _parse_expire_at(kwargs["expire_at"])
|
|
with db.app.app_context():
|
|
u = db.session.query(User).get(uid)
|
|
if not u:
|
|
return {"error": "用户不存在"}
|
|
password = kwargs.pop("password", None)
|
|
for k, v in kwargs.items():
|
|
if hasattr(u, k) and v is not None:
|
|
setattr(u, k, v)
|
|
if password:
|
|
u.set_password(password)
|
|
db.session.commit()
|
|
log.info("更新用户: %s", u.username)
|
|
return {"ok": True}
|
|
|
|
def delete_user(self, uid):
|
|
with db.app.app_context():
|
|
u = db.session.query(User).get(uid)
|
|
if not u:
|
|
return {"error": "用户不存在"}
|
|
db.session.delete(u)
|
|
db.session.commit()
|
|
log.info("删除用户: %s", u.username)
|
|
return {"ok": True}
|
|
|
|
def toggle_user(self, uid, enabled):
|
|
with db.app.app_context():
|
|
u = db.session.query(User).get(uid)
|
|
if u:
|
|
u.enabled = enabled
|
|
db.session.commit()
|
|
return {"enabled": u.enabled if u else None}
|
|
|
|
def ban_user(self, uid, banned):
|
|
with db.app.app_context():
|
|
u = db.session.query(User).get(uid)
|
|
if u:
|
|
u.banned = banned
|
|
db.session.commit()
|
|
return {"banned": u.banned if u else None}
|
|
|
|
# ── 流量统计 ───────────────────────────────────────────────
|
|
|
|
def add_traffic(self, username, bytes_in, bytes_out):
|
|
"""增加用户流量统计。"""
|
|
if not username:
|
|
return
|
|
with db.app.app_context():
|
|
u = db.session.query(User).filter_by(username=username).first()
|
|
if not u:
|
|
return
|
|
u.bytes_in += bytes_in
|
|
u.bytes_out += bytes_out
|
|
# 检查流量上限
|
|
total_mb = (u.bytes_in + u.bytes_out) / 1024 / 1024
|
|
if u.total_traffic_mb and total_mb >= u.total_traffic_mb:
|
|
u.enabled = False
|
|
log.warning("用户 %s 总流量超限: %.1fMB / %.1fMB",
|
|
username, total_mb, u.total_traffic_mb)
|
|
# 月流量检查
|
|
from datetime import date
|
|
today = date.today()
|
|
if u.monthly_traffic_mb:
|
|
# 简单检查：假设 bytes_in/bytes_out 是累计的，需要更精确的实现
|
|
pass
|
|
db.session.commit()
|
|
|
|
# ── 连接追踪 ───────────────────────────────────────────────
|
|
|
|
def get_active_connections(self, username):
|
|
"""获取指定用户的活跃连接数。"""
|
|
if not username:
|
|
return 0
|
|
return self._active_conn_counts.get(username, 0)
|
|
|
|
def register_connection(self, username):
|
|
if username:
|
|
self._active_conn_counts[username] += 1
|
|
|
|
def unregister_connection(self, username):
|
|
if username:
|
|
self._active_conn_counts[username] = max(
|
|
0, self._active_conn_counts.get(username, 0) - 1
|
|
)
|
|
|
|
# ── 审计日志 ───────────────────────────────────────────────
|
|
|
|
async def log_event(self, event, user=None, src_ip=None, detail=""):
|
|
with db.app.app_context():
|
|
try:
|
|
log_entry = AuditLog(
|
|
event=event,
|
|
user=user,
|
|
src_ip=src_ip,
|
|
detail=detail,
|
|
)
|
|
db.session.add(log_entry)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
log.error("写审计日志失败: %s", e)
|
|
|
|
def query_logs(self, page=1, per_page=50, event="", user="",
|
|
src_ip="", start_date=None, end_date=None):
|
|
with db.app.app_context():
|
|
q = db.session.query(AuditLog).order_by(AuditLog.timestamp.desc())
|
|
if event:
|
|
q = q.filter(AuditLog.event == event)
|
|
if user:
|
|
q = q.filter(AuditLog.user.ilike(f"%{user}%"))
|
|
if src_ip:
|
|
q = q.filter(AuditLog.src_ip.ilike(f"%{src_ip}%"))
|
|
if start_date:
|
|
q = q.filter(AuditLog.timestamp >= start_date)
|
|
if end_date:
|
|
q = q.filter(AuditLog.timestamp <= end_date)
|
|
total = q.count()
|
|
items = q.offset((page - 1) * per_page).limit(per_page).all()
|
|
return {
|
|
"logs": [{
|
|
"id": l.id,
|
|
"timestamp": l.timestamp.isoformat(),
|
|
"event": l.event,
|
|
"user": l.user,
|
|
"src_ip": l.src_ip,
|
|
"detail": l.detail,
|
|
} for l in items],
|
|
"total": total,
|
|
"page": page,
|
|
}
|
|
|
|
# ── 防爆破 ─────────────────────────────────────────────────
|
|
|
|
def check_fail2ban(self, src_ip):
|
|
"""检查源IP是否应被防爆破。返回 (blocked, remaining)。"""
|
|
now = time.time()
|
|
records = _fail2ban.get(src_ip, [])
|
|
# 清理过期记录
|
|
records = [t for t in records if now - t < _FAIL2BAN_WINDOW]
|
|
_fail2ban[src_ip] = records
|
|
|
|
if len(records) >= _FAIL2BAN_MAX:
|
|
return True, 0
|
|
return False, max(0, _FAIL2BAN_MAX - len(records))
|
|
|
|
def record_auth_attempt(self, src_ip, success=False):
|
|
if not success:
|
|
_fail2ban[src_ip].append(time.time())
|
|
|
|
# ── 统计信息 ───────────────────────────────────────────────
|
|
|
|
def get_stats(self):
|
|
with db.app.app_context():
|
|
total_users = db.session.query(User).count()
|
|
active_users = db.session.query(User).filter_by(
|
|
enabled=True, banned=False
|
|
).count()
|
|
total_bytes_in = db.session.query(
|
|
db.func.sum(User.bytes_in)
|
|
).scalar() or 0
|
|
total_bytes_out = db.session.query(
|
|
db.func.sum(User.bytes_out)
|
|
).scalar() or 0
|
|
return {
|
|
"total_users": total_users,
|
|
"active_users": active_users,
|
|
"total_bytes_in_mb": round(total_bytes_in / 1024 / 1024, 1),
|
|
"total_bytes_out_mb": round(total_bytes_out / 1024 / 1024, 1),
|
|
}
|