"""用户服务：认证、流量统计、限速、生命周期。""" 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), }