GravityAgent（六）：多用户 Web 平台的设计与隔离

多用户 Web 平台的设计与隔离

前面五篇讲的都是”单机能力”：算法能跑、扫描能扫、OS 能识别、Agent 能编排。但要把这些能力开放给一个团队使用，还需要一层多用户 Web 平台。这一篇讲 agent/web_app.py。

为什么需要 Web 平台

命令行 Agent 适合研究者调试，但团队协作会遇到几个问题：

1. 多人共用：不同用户需要各自的配置、聊天历史、文件；
2. 权限控制：谁能提交扫描、谁能看哪些批次；
3. 长任务体验：6Sense 训练几十分钟，命令行可以等，Web 页面不能”转圈”；
4. 文件管理：生成的结果需要能下载、能归档、能隔离。

web_app.py 用 Flask 实现了这些能力。

用户体系与审批流

users.db 的表结构

Web 平台用 agent/users.db 保存所有平台数据：

注册审批

注册不是即时的，而是进入审批队列：

用户提交注册
    │
    ▼
pending_registrations (status='pending')
    │
    ▼
管理员审核
    ├── approve → 写入 users 表，status='active'
    └── reject  → 标记拒绝

内置账号

系统内置两个账号：

• admin/admin123：管理员，每次启动强制对齐；
• realtime/realtime：专供 Agent 自动提交扫描使用。

_ensure_admin_accounts 在每次启动时强制对齐这两个账号，避免因为数据库被改坏导致系统无法登录。

鉴权装饰器

@login_required   # 要求已登录且 status='active'
@admin_required   # 要求 role='admin'

_current_user() 从 session 查 users 表，非 active 状态一律视为未登录。

每用户 Agent 实例

这是多用户平台最巧妙的设计。每个用户可以有独立的 LLM 配置（API key、base_url、model、代理），系统需要为每个用户构建独立的 Agent 实例。

配置合并

def _get_cfg(uid: str) -> dict:
    # user_configs 中的配置合并 DEFAULT_CONFIG
    return {**DEFAULT_CONFIG, **json.loads(row["config_json"])}

DEFAULT_CONFIG 包含 LLM API key、base_url、model、scanner API、scanner DB URL、代理等默认值。

构建 Agent

def _build_agent(uid: str):
    am = _configure_agent_runtime(uid)
    cfg = _get_cfg(uid)
    proxy = cfg["proxy_url"] if cfg.get("enable_proxy") else None
    llm_timeout = httpx.Timeout(connect=30.0, read=LLM_HTTP_TIMEOUT_SEC, ...)
    hc = httpx.Client(proxy=proxy, timeout=llm_timeout) if proxy else httpx.Client(trust_env=False, timeout=llm_timeout)
    llm = ChatOpenAI(model=cfg["llm_model"], temperature=0, api_key=cfg["llm_api_key"],
                     base_url=cfg["llm_base_url"], http_client=hc, http_async_client=ahc)
    tools = [...25 个工具...]
    return create_react_agent(model=llm, tools=tools, prompt=am.SYSTEM_PROMPT)

注意 Web 版 Agent 绑定的是 25 个工具，比 CLI 版少一个 generate_os_confidence_report（该工具通过 Web 的 OS API 单独暴露）。

全局变量切换的取舍

_configure_agent_runtime 会覆写 agent 模块的全局变量：

am.SCANNER_API = cfg["scanner_api"]
am.SCANNER_USER = cfg.get("scanner_user", ...)
am.LLM_API_KEY = cfg["llm_api_key"]
...
set_current_user(uid)  # 设置用户上下文

这是一个有意的工程取舍：agent.py 原本是为单用户 CLI 写的，用了大量模块级全局变量。Web 版通过”请求前切换全局变量”来复用这些工具函数，而不是重写成面向对象的服务。

代价是：同一进程内多个用户不能真正并发执行 Agent，必须依赖锁和队列串行化。对于这个规模的团队使用场景，这个取舍是合理的。

优先级扫描队列

扫描提交可能很慢（上传大文件到 scanner），不能阻塞 Web 请求。系统实现了一个小顶堆优先级队列：

def _enqueue(uid, role, fpath, count, ...):
    token = secrets.token_hex(8)
    pri = 100 if role == "admin" else 1
    with _pq_lock:
        heapq.heappush(_pq, (-pri, time.time(), token, uid, fpath, ...))
        _pq_meta[token] = {"status": "queued", ...}
    _pq_event.set()
    return token

堆的排序键是 (-priority, time, token, ...)，所以：

• 优先级高的（admin=100）先出队；
• 同优先级按提交时间（FIFO）；
• token 保证时间相同时顺序稳定。

后台 worker 线程消费队列：

def _scan_worker():
    while True:
        _pq_event.wait(timeout=5)
        _pq_event.clear()
        while True:
            item = heapq.heappop(_pq)
            # 状态：queued → submitting → submitted / failed
            batch_id, submit_error = _do_submit(uid, fpath, ...)
            if batch_id:
                # 写入 scan_ownership，实现批次归属隔离
                conn.execute("INSERT OR REPLACE INTO scan_ownership(batch_id,user_id,priority) VALUES(?,?,?)",
                             (batch_id, uid, pri))

前端通过 api_scan_queue(token) 轮询提交进度，通过 api_scan_batches 列出本用户可访问的批次。

批次归属隔离

scan_ownership 表把 batch_id 和 user_id 绑定。普通用户查询批次列表时，只能看到自己提交的批次；管理员可以看到全部。这解决了”多人共用一个 scanner”时的数据隔离问题。

后台聊天任务

Web 不直接同步执行长耗时对话，而是通过后台 job 方式运行：

def _register_chat_job(uid, message, attached_file_path) -> tuple[str, threading.Event]:
    # 生成 token，注册 job，返回 (token, event)

def _run_chat_job(uid, chat_msg, raw_msg, attached_file_path) -> dict:
    # 后台线程执行 agent.invoke

def _start_chat_job(token, uid, chat_msg, raw_msg, attached_file_path):
    # 启动后台线程

前端拿到 token 后轮询 api_chat_job，这样 6Sense 训练、候选生成、在线扫描、OS 识别等长任务不会阻塞 HTTP 请求线程。

细分等待文案

后端根据用户意图推断 pending_label：

def _infer_chat_pending_label(raw_msg, attached_file_path) -> str:
    if _is_file_prediction_request(raw_msg, attached_file_path):
        return "地址生成预测任务执行中"
    if any(k in text for k in ("预测", "候选", "生成地址", ...)):
        return "地址生成预测任务执行中"
    if any(k in text for k in ("模式挖掘", "挖掘模式", "规律", ...)):
        return "模式挖掘任务执行中"
    if any(k in text for k in ("提交扫描", "开始扫描", ...)):
        return "批次扫描提交任务执行中"
    if any(k in text for k in ("扫描进度", "批次状态", ...)):
        return "扫描结果查询任务执行中"
    return "后台任务执行中"

前端轮询时同时展示”已等待 xx 秒”。这项改造直接解决了长任务对话看起来”像卡住了”的问题。

防幻觉的后端兜底

这是 Web 层最有价值的一段代码。它针对一个具体风险：模型没调工具，却声称成功。

识别附件预测请求

def _is_file_prediction_request(raw_msg, attached_file_path) -> bool:
    if not attached_file_path:
        return False
    text = (raw_msg or "").strip().lower()
    if not text:
        return True  # 只传文件不写文字，默认视为预测请求
    keywords = ("预测", "生成", "候选", "predict", "candidate",
                "6sense", "6graph", "6forest", "entropy")
    return any(k in text for k in keywords)

推断算法

def _infer_file_prediction_algorithm(raw_msg) -> str:
    if "6sense" in text: return "6sense"
    if "6forest" in text: return "6forest"
    if "entropy" in text: return "entropy_ip"
    if "6graph" in text: return "6graph"
    return "6graph"  # 默认 6Graph

直接执行

def _fallback_predict_from_file(uid, raw_msg, attached_file_path):
    am = _configure_agent_runtime(uid)
    result = am.predict_candidates_from_file.invoke({
        "file_path": attached_file_path,
        "algorithm": _infer_file_prediction_algorithm(raw_msg),
        "budget": 5000,
        "seed_limit": 0,
        "prefix_filter": _infer_prefix_filter(raw_msg),
        "save_file": True,
    })
    return (
        "检测到模型未实际调用预测工具，已由后端直接执行附件预测。\n\n" + result,
        ["predict_candidates_from_file"],
    )

注意返回文案里明确写了”检测到模型未实际调用预测工具”——这是诚实的可观测性，而不是掩盖问题。

文件归档

当 Agent 回复中包含绝对路径时，Web 会尝试归档：

def _organize_reply_file_links(uid, reply) -> str:
    paths = _extract_file_paths(reply)
    ready = _wait_until_files_ready(paths, timeout_sec)
    for p in ready:
        dst = ...  # download/<uid>/<feature>/
        if not _path_under_root(dst, DOWNLOAD_DIR):
            continue
        # 复制并替换回复中的路径

归档逻辑有三个关键点：

1. 等待文件真正落盘（_wait_until_files_ready），避免竞态；
2. 路径必须在 DOWNLOAD_DIR 下（_path_under_root），防止越权写入；
3. 失败时显示”未归档文件(<文件名>)”，而不是假装成功。

用户隔离存储

所有生成结果按用户隔离：

download/<uid>/
├── os_identify/          # OS 识别报告（JSON/Markdown）
├── candidates/           # 候选地址生成结果
├── patterns/             # 模式挖掘结果
├── alive_ips/            # 存活地址导出
├── batch_results/        # 批次扫描结果
└── service_probe/        # 服务探测结果

上传目录同样隔离：

upload/<uid>/

路径由 runtime_paths.py 统一生成，用户上下文通过 set_current_user(uid) 设置。

路径穿越防护

文件下载接口必须防止目录穿越攻击。核心是 _path_under_root：

def _path_under_root(path: Path, root: Path) -> bool:
    try:
        p = path.resolve()
        r = root.resolve()
    except Exception:
        return False
    return p == r or str(p).startswith(str(r) + os.sep)

先 resolve() 展开 .. 和符号链接，再判断是否真的在根目录下。注意结尾的 os.sep——如果没有它，/download/evil 会被误判为在 /download 下。

会话安全

app.config.update(
    SESSION_COOKIE_HTTPONLY=True,      # JS 无法读取 cookie
    SESSION_COOKIE_SAMESITE="Lax",     # 缓解 CSRF
    PERMANENT_SESSION_LIFETIME=timedelta(days=7),
)

secret_key 持久化到 agent/.secret_key 文件，这样服务重启后 session 不会失效：

_SK_FILE = AGENT_DIR / ".secret_key"
if _SK_FILE.exists():
    app.secret_key = _SK_FILE.read_text().strip()
else:
    _sk = secrets.token_hex(32)
    _SK_FILE.write_text(_sk)
    app.secret_key = _sk

路由分组

Web 平台有 40 多个路由，按功能分组：

独立数据可视化 WebUI

除了主平台，还有一个独立的只读数据控制台 dbui/app.py，用于：

1. 自动发现仓库内 SQLite 数据库；
2. 连接 gravity-scanner PostgreSQL 主库；
3. 统一浏览表、字段、样本数据和分布；
4. 执行只读 SQL 查询。

它和主平台完全分离，只读、无鉴权复杂度，适合运维排查数据。

小结

多用户 Web 平台的设计要点：

下一篇是系列的最后一篇，我们来复盘这个项目在工程化过程中踩过的真实坑——模块冲突、除零 bug、WAL 落库、参数错位，以及它们各自是怎么被定位和修复的。