diff --git a/.env.example b/.env.example index 690a87c..7f70f7d 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,8 @@ LAN_CIDRS=192.168.88.0/24 # ROS_USER=rosdpi # ROS_PASSWORD= # ROS_VERIFY_TLS=false +# 采集端口只收这些地址发来的包（一般填路由器 IP，逗号分隔，可写网段）；留空不限制 +# COLLECTOR_ALLOWED=192.168.88.1 # WAN_INTERFACE=pppoe-out1 # RouterOS 完整轮询间隔 / WAN 口速率采样间隔（实时流量图） @@ -81,3 +83,4 @@ LAN_CIDRS=192.168.88.0/24 # SHARE_HUB_URL=https://rosdpi.suyijun.top:8443 # 社区主站（内置，一般不用改）；自建私有社区时改成自己的主站 # SHARE_HUB=false # 只有主站自己设为 true # SHARE_VERIFY_TLS=true # 私有主站用自签证书时才设 false +# SHARE_TRUSTED_PROXIES= # 主站：可信反向代理（留空 = 内网和回环地址） diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5e43f83..d87584f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -13,6 +13,11 @@ permissions: jobs: release: runs-on: ubuntu-latest + # 同时推 main 和 v* tag 时两次运行都会走到这里：串行执行，后到的一次会看到“已发布”直接跳过， + # 不会两边同时 gh release create 撞车 + concurrency: + group: github-release + cancel-in-progress: false env: GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} steps: @@ -58,9 +63,13 @@ jobs: { echo '### 升级' echo - echo '```bash' - echo 'docker compose pull && docker compose up -d' - echo '```' + echo '- Docker：' + echo + echo ' ```bash' + echo ' docker compose pull && docker compose up -d' + echo ' ```' + echo + echo '- Linux 二进制：网页 系统设置 → 版本更新 → 一键更新；或重新运行一键安装命令（配置在 /etc/ros-dpi.env，不会被覆盖）。' echo echo "$NOTES" } > notes.md diff --git a/CHANGELOG.md b/CHANGELOG.md index e9d27ba..db93f6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,56 @@ - 来源地址只认反向代理追加在 X-Forwarded-For 最右边的一跳，可用 `SHARE_TRUSTED_PROXIES` 指定可信代理。 - 同一 IPv6 /48 每天最多注册 20 个实例；同一来源网络最多 2 个实例的投票被收下；投票总量上限 500 万行；实例注册满 1 天，它的投票才参与计票。 - **修复 IPFIX 模板字段长度全为 0 时死循环**（全站审计 P0）：内网任何一台机器往 2055 端口发一个构造好的包，就能让整个进程卡死，网页一起打不开。 +- **全站审计修复**（详见仓库根目录的审计报告）： + - 统计数据： + - 某台设备的 DHCP 主机名超过 128 字符时，MySQL 严格模式会拒绝整批落库，统计全部停写——现在写库前一律按列长截断； + - 数据本身有问题时改为逐段写入、只丢坏的那段（以前整批重试 2 小时后连同所有终端的数据一起丢弃）； + - 落库失败后的补写按各自的时间桶写入，跨天不再记错日期； + - 0 点执行的落库不再把 23:59 那一分钟记进第二天； + - commit 成功但关闭连接出错时不再重复累加；落库之间加锁串行。 + - 性能： + - 本地自学习库改为在数据库里聚合； + - 过了零点的自动重算只补算新的一天（以前每天把 31 天全部重扫，抢住 GIL 导致 UDP 收包丢失）； + - 访问日志、时间线、洞察的查询加上时间下界，走索引； + - 补建 DnsCache.expires、DomainGeo.updated 两个索引（已有的库启动时自动补上）； + - SQLite 锁等待放宽到 30 秒。 + - 识别： + - 本地自学习库不再学入 HTTPS / 国内网络这类笼统结论（它们会挡住后面更准的识别、还会自我强化），社区共享同样不上报； + - CNAME 链取第一个认得出应用的名字； + - ECH 外层名（cloudflare-ech.com）不再覆盖终端的 DNS 记录； + - DNAT 下行用 NAT 后的端口匹配连接； + - 支持带 IPv6 扩展头的包； + - ASN 库加载后分类缓存失效； + - 自定义规则改为在新识别器上建好再整体替换； + - v2fly 类别猜测里 `FM\b` 写成了字面退格符，从未生效，已修正。 + - 采集安全： + - 新增 `COLLECTOR_ALLOWED`：采集端口只收路由器发来的包； + - IPFIX 模板表、TLS / QUIC 重组、超长域名都加了上限（以前内网任何一台机器都能用构造的包撑爆内存或拖慢 CPU）； + - DnsMap 加锁，修复访问目标接口间歇性 500。 + - Web 安全： + - 管控脚本里的设备名转义 `$`、去掉换行，域名只收规范主机名（以前恶意设备名能在管理员粘贴脚本时注入 RouterOS 命令）； + - 登录锁定只认反向代理追加的最右一跳，并增加按用户名的总失败数锁定； + - 从公网首次设置需要启动日志里的一次性安装码，内网不受影响； + - 通知渠道的请求头、代理地址也做遮挡，报错信息里的令牌脱敏，改了发送地址必须重新填写密钥； + - CSV 导出防公式注入； + - 删除账号后同名重建，旧 token 不再有效； + - 只读用户看到的外部数据源地址去掉账号密码。 + - 一键更新与发版： + - 一键更新校验 SHA256SUMS、加锁防重复点击、留 `.bak` 便于回滚、请求超时放宽到 10 分钟； + - 新增 arm64 二进制； + - Release 等测试和两个架构的二进制都就绪后才发布，已发布的版本不再被覆盖； + - Docker 浮动标签 `:0.6` 随主线自动更新。 + - install.sh： + - 校验 SHA256； + - 数据目录改为 700，服务加 UMask 等加固； + - 自定义配置改放 `/etc/ros-dpi.env`，重装不覆盖，旧服务文件里的配置自动搬入； + - 重装后会重启服务。 + - 前端： + - 修改密码后当前会话不掉线； + - 几个慢接口不再误报超时； + - 加载失败有提示和重试； + - 下拉框保存失败会退回原值。 +- **审计规则**：跨午夜的生效时段（如 22:00-06:00）按时段累计，0 点不再清零重报。 - **接入向导**（系统设置 → 接入与自检）新增 3b、3c 两步：路由器开了 IPv6 时在 `/ipv6 firewall mangle` 再加一份 DNS / SNI 嗅探（4 条）；开了 FastTrack 时让 IPv4 / IPv6 每条连接的前 10KB 不走 FastTrack。此前向导只有 IPv4 命令，IPv6 流量的 DNS 应答和握手全都镜像不到。 - **接入自检**读取 `/ipv6/firewall/filter`：IPv6 的 FastTrack 规则（ROS 7.18 起支持）没有限定 connection-bytes 时提示，并给出修改命令。 - README 与 `docs/RouterOS配置.md` 补充 IPv6 一节：为什么必须做、4 条命令、镜像目标可以用 ROS-DPI 的 IPv4 地址、流量采集不用另配。 diff --git a/DOCKERHUB.md b/DOCKERHUB.md index 942c04a..2f7051b 100644 --- a/DOCKERHUB.md +++ b/DOCKERHUB.md @@ -36,7 +36,7 @@ services: ## What's new -- **0.6.22**: 社区主站地址可在网页填写，新增 VPS 主站部署模板（`deploy/hub/`，Docker + Caddy 自动 HTTPS）；接入向导与文档补上 IPv6 嗅探（`/ipv6 firewall mangle` 4 条）和 FastTrack（IPv4 / IPv6）命令；自检检查 IPv6 的 FastTrack。Setup guide and docs now include IPv6 sniff rules and FastTrack commands for both families; the self-check also flags IPv6 FastTrack without a connection-bytes limit. +- **0.6.22**: 全站审计修复（落库容错、采集端口防护 `COLLECTOR_ALLOWED`、管控脚本防注入、一键更新校验 SHA256、arm64 二进制等）；社区主站地址可在网页填写，新增 VPS 主站部署模板（`deploy/hub/`，Docker + Caddy 自动 HTTPS）；接入向导与文档补上 IPv6 嗅探（`/ipv6 firewall mangle` 4 条）和 FastTrack（IPv4 / IPv6）命令；自检检查 IPv6 的 FastTrack。Setup guide and docs now include IPv6 sniff rules and FastTrack commands for both families; the self-check also flags IPv6 FastTrack without a connection-bytes limit. - **0.6.21**: 社区共享库重做——项目自带社区主站，不再需要 WebDAV / 第三方存储：令牌注册、按来源网络计票（≥2 个不同网络一致才发布，防伪造实例刷票）、ETag 增量拉取、本地缓存；修复 0.6.20 部署量接口 500、共享任务从未运行、开关保存不上。接入自检新增 IPv6 嗅探规则检查，FastTrack 已限定 connection-bytes 时不再误报，并直接给出修改命令。Community library rebuilt around a built-in hub (no WebDAV/third-party storage): per-instance tokens, votes counted per source network, ETag pulls, local cache; fixes the 0.6.20 deployments 500 and the never-running share job. Self-check now flags missing IPv6 sniff rules. - **0.6.12 / 0.6.11 / 0.6.10 / 0.6.9 / 0.6.8**: 自学习与共享系列。**本地自学习库**：本网络实测的「IP → 应用」（IPv6 按 /64 前缀聚合）与「域名 → 应用」长期记忆，每小时重建，DNS 缓存过期、DoH、裸 IP 流量都能按记忆归类；**社区共享库**：各部署实例把实测结论匿名共享（中转聚合、至少 2 个实例见过才发布），本地实测优先、社区只补空白；**全球部署量**：实例心跳上报，中转公开已部署 / 在线 / 离线数量；共享总开关默认打开；**一键自更新**：Linux 二进制部署在网页上检测并一键升级（自动下载、替换、重启服务）；**Linux 单文件二进制部署**：`install.sh` 一键安装 systemd 服务（SQLite 零配置起步）；修复落库卡死（代理长域名撑爆 access_log.remote 列导致每分钟落库失败、在线终端归零）；组播 / 广播流量不再计入统计；17 个 DPI 大类配色互异。 - **0.6.7**: fixes for the 0.6.5 / 0.6.6 history reclassification and rule auto-registration. Reclassification now only touches days before today and only upgrades vague rows (HTTPS / Chinese websites / unknown / CDN) to a specific app — it no longer downgrades apps identified live by IP range, port or ASN. New upstream blackmatrix7 directories are downloaded first and then assigned by a vote of the existing libraries. The manual dark theme now uses the 17 distinct group colors, and the AdGuard Home password is masked in the operation log. @@ -124,6 +124,8 @@ Open `http://:8080`. The setup wizard has two steps. 打开网页后按两 | `SHARE_HUB_URL` | `https://rosdpi.suyijun.top:8443` | community hub (built in); point it at your own hub for a private community | | `SHARE_HUB` | `false` | run this instance as the community hub (only the hub sets `true`) | | `SHARE_VERIFY_TLS` | `true` | verify the hub certificate; set `false` only for a private hub with a self-signed cert | +| `COLLECTOR_ALLOWED` | empty | accept IPFIX / TZSP only from these IPs / CIDRs (usually the router); empty = any | +| `SHARE_TRUSTED_PROXIES` | empty | hub only: trusted reverse proxies whose `X-Forwarded-For` is honoured; empty = private / loopback | | `TZ` | `Asia/Shanghai` | statistics time zone (weeks start Monday) | Full list: https://github.com/suyijun8182/ros-dpi#环境变量 diff --git a/README.md b/README.md index 1c01096..afdd326 100644 --- a/README.md +++ b/README.md @@ -217,6 +217,8 @@ EOF | `SHARE_HUB_URL` | `https://rosdpi.suyijun.top:8443` | 社区主站地址（内置，一般不用改）；自建私有社区时改成自己的主站 | | `SHARE_HUB` | `false` | 本实例作为社区主站（只有主站部署设为 `true`） | | `SHARE_VERIFY_TLS` | `true` | 校验主站证书；私有社区用自签证书时才设 `false` | +| `SHARE_TRUSTED_PROXIES` | 空 | 主站：可信反向代理地址 / 网段（逗号分隔），只信它们传来的 `X-Forwarded-For`；留空 = 内网和回环地址 | +| `COLLECTOR_ALLOWED` | 空 | 采集端口（IPFIX / TZSP）只收这些地址的包，一般填路由器 IP；留空不限制。内网有不可信设备时建议设置 | | `FAKEIP_CIDRS` | `198.18.0.0/15,fc00::/18` | FakeIP 地址段 | | `ASN_DB` / `CHINA_DB` / `IPLIST_DB` / `BM7_DB` / `HAGEZI_DB` / `UT1_DB` / `OUI_DB` | `true` | 各识别库自动下载开关；没有外网时设 `false` | | `WAN_INTERFACE` | 自动 | WAN 接口名，默认按默认路由识别 | diff --git a/backend/app/audit.py b/backend/app/audit.py index 7a3c694..8f896e4 100644 --- a/backend/app/audit.py +++ b/backend/app/audit.py @@ -15,7 +15,7 @@ from datetime import datetime from zoneinfo import ZoneInfo import httpx -from sqlalchemy import select +from sqlalchemy import select, update from app import database, notify, people, sharing from app.config import get_settings @@ -41,6 +41,18 @@ def in_window(rule, ts: float) -> bool: return ok_time and ok_day +def window_start(rule, ts: float) -> float: + """这一段生效时段从哪天 0 点算起：跨午夜的时段（22:00-06:00）凌晨那一段属于前一天， + 累计流量和“一天一次”都按它算，不能到 0 点就清零重报""" + d = datetime.fromtimestamp(ts, _tz()) + a, b = rule.time_from or "00:00", rule.time_to or "24:00" + start = d.replace(hour=0, minute=0, second=0, microsecond=0) + if a > b and d.strftime("%H:%M") < b: + from datetime import timedelta + start = (start - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + return start.timestamp() + + def domain_match(domain: str, suffix: str) -> bool: suffix = suffix.lower().strip(".") return domain == suffix or domain.endswith("." + suffix) @@ -76,12 +88,16 @@ class Auditor: if not rules: return cl = self.engine.classifier - day = datetime.fromtimestamp(now, _tz()).strftime("%Y%m%d") events = [] + # 只保留近两天的累计（以前按天累积的键永远不删） + if len(self._acc) > 1000: + keep = _day_start(now) - 86400 + self._acc = defaultdict(int, {k: v for k, v in self._acc.items() if k[2] >= keep}) sharing_hits = None for rule in rules: if not in_window(rule, now): continue + day = window_start(rule, now) only = people.expand_scope([x for x in (rule.hosts or "").split(",") if x]) hits: dict[str, dict] = defaultdict(lambda: {"bytes": 0, "items": defaultdict(int)}) if rule.kind == "new_host": @@ -136,7 +152,7 @@ class Auditor: self._acc[(rule.id, host, day)] += h["bytes"] if self._acc[(rule.id, host, day)] < rule.min_bytes: continue - if self._fired.get(key, 0) >= _day_start(now): # 一天一次 + if self._fired.get(key, 0) >= day: # 每个生效时段一次 continue elif now - self._fired.get(key, 0) < 3600: # 一小时一次 continue @@ -206,17 +222,25 @@ class Auditor: return from app import stats with database.session() as db: + if not getattr(self, "_sending_reset", False): + # 上次进程在发送途中退出：“发送中”的事件放回待发 + db.execute(update(AuditEvent).where(AuditEvent.notified == 3).values(notified=2)) + self._sending_reset = True evs = db.scalars(select(AuditEvent).where(AuditEvent.notified == 2).order_by(AuditEvent.ts).limit(500)).all() if not evs: + db.commit() return rules = {r.id: r for r in db.scalars(select(AuditRule))} meta = stats.host_meta(db, {e.host for e in evs}) - items = [] - for e in evs: - r = rules.get(e.rule_id) - if r is None: - continue - items.append((e.id, r, e.host, json.loads(e.detail or "{}"), meta[e.host])) + items = [] + for e in evs: + r = rules.get(e.rule_id) + # 先标成“发送中”（3）再推送：推送比一分钟长时，下一轮不会把同一批再查出来重复发送； + # 规则已删除的直接作废，不能一直占着前 500 条让新积压的永远排不上 + e.notified = 3 if r is not None else 0 + if r is not None: + items.append((e.id, r, e.host, json.loads(e.detail or "{}"), meta[e.host])) + db.commit() self._dispatch(items, time.time(), title=f"ROS-DPI 免打扰期间的审计提醒（{len(items)} 条）") def _label(self, x: str) -> str: diff --git a/backend/app/classify/v2fly.py b/backend/app/classify/v2fly.py index d7c89a8..bcc730d 100644 --- a/backend/app/classify/v2fly.py +++ b/backend/app/classify/v2fly.py @@ -73,7 +73,7 @@ NAME_HINTS = [ ] # 分类文件里按注释分组的应用：注释（多为中文名）里含这些词时，改用更准确的小类 LABEL_HINTS = [ - (r"小说|阅读|书城|漫画|文学", "reading"), (r"音乐|电台|播客|FM", "music"), (r"直播", "live"), + (r"小说|阅读|书城|漫画|文学", "reading"), (r"音乐|电台|播客|FM\b", "music"), (r"直播", "live"), (r"游戏|手游|电竞", "game"), (r"银行|Bank", "bank"), (r"证券|基金|期货|股票", "stock"), (r"医院|医疗|医学|健康", "life"), (r"大学|学院|中学|教育|学习|题库|考试|University", "education"), (r"视频|影视|电影|动漫|番剧", "video"), diff --git a/backend/app/database.py b/backend/app/database.py index 4665e4f..3777fc7 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -37,7 +37,8 @@ def _make_engine(url: str): path = url.split("///", 1)[-1] if path and path != ":memory:": os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) - eng = create_engine(url, connect_args={"check_same_thread": False}) + # 锁等待 30 秒（默认 5 秒）：历史重算持有写锁时，每分钟落库和 API 不至于直接报 database is locked + eng = create_engine(url, connect_args={"check_same_thread": False, "timeout": 30}) @event.listens_for(eng, "connect") def _pragma(conn, _): @@ -183,6 +184,12 @@ def add_missing_columns(): ctype = col.type.compile(dialect=engine.dialect) conn.exec_driver_sql(f"ALTER TABLE `{table.name}` MODIFY `{col.name}` {ctype}") print(f"[db] 已把 {table.name}.{col.name} 从 {db_len} 加宽到 {length}", flush=True) + # 新版本加的索引：create_all 不会给已有的表补，这里补上 + have_idx = {i["name"] for i in insp.get_indexes(table.name)} + for idx in table.indexes: + if idx.name and idx.name not in have_idx: + idx.create(conn) + print(f"[db] 已为 {table.name} 添加索引 {idx.name}", flush=True) def add_ready_hook(fn): diff --git a/backend/app/engine.py b/backend/app/engine.py index 8f73a6e..63ad13d 100644 --- a/backend/app/engine.py +++ b/backend/app/engine.py @@ -59,6 +59,7 @@ class Engine: self.sni = SniExtractor() self.classifier = Classifier() self.lock = threading.Lock() + self._flush_lock = threading.Lock() # CNAME 链里挑能认出应用的那一跳（按当时的识别器） dnsmap.cname_matcher = lambda name: self.classifier.match_domain(name) self.allowed_sources = parse_cidrs(s.collector_allowed) if s.collector_allowed else [] @@ -510,6 +511,16 @@ class Engine: def flush(self, now: float | None = None, final: bool = False): if not database.is_configured(): return # 安装向导完成前先留在内存里，配置好后的第一次落库一起写入 + # 落库串行：关闭时的最后一次落库可能和定时落库撞上，_pending / last_flush 会互相覆盖 + if not self._flush_lock.acquire(timeout=30): + print("[engine] 上一次落库 30 秒还没结束，这次跳过", flush=True) + return + try: + self._flush(now, final) + finally: + self._flush_lock.release() + + def _flush(self, now: float | None, final: bool): now = now or time.time() with self.lock: visits, visit_upd = self.take_visits(now, final) @@ -743,13 +754,24 @@ class Engine: print(f"[db] 已回填 {len(doms)} 个域名的主域名列", flush=True) def cleanup(self): + # 内存表的清理不依赖数据库：安装向导完成前也要做，否则长时间不配置会一直涨 + self.dns.purge() + with self.lock: + self.names.purge() + if not database.is_configured(): + # 还没有数据库：访问记录、域名 / IP 统计攒着也写不出去，超量就丢，别把内存撑爆 + if len(self._closed_visits) > 100_000: + self._closed_visits = self._closed_visits[-100_000:] + for name in ("_dom", "_ipdest", "_seen"): + if len(getattr(self, name)) > 200_000: + getattr(self, name).clear() if not database.is_configured(): return from app import compliance r = compliance.retention() # 网页里设置的留存天数优先于环境变量 now = int(time.time()) with database.session() as db: - # 分批删除（每批单独提交）：存量大的表一次删会造成长事务、大锁 + # 分批删除（每批单独提交）：存量大的表一次删会造成长事务、大锁；一张表出错不影响其它表 for model, col, days in ((TrafficMinute, TrafficMinute.ts, r["minute"]), (TrafficHour, TrafficHour.ts, r["hour"]), (TrafficDay, TrafficDay.ts, r["day"]), @@ -758,12 +780,13 @@ class Engine: (AccessLog, AccessLog.ts_start, r["log"]), (AuditEvent, AuditEvent.ts, r["event"]), (OpLog, OpLog.ts, r["event"])): - _purge(db, model, col, days) + try: + _purge(db, model, col, days) + except Exception as e: # noqa: BLE001 + db.rollback() + print(f"[cleanup] {model.__tablename__} 清理失败：{type(e).__name__}: {str(e)[:120]}", flush=True) db.execute(delete(DnsCache).where(DnsCache.expires < now)) db.commit() - self.dns.purge() - with self.lock: - self.names.purge() def is_generic(self, app: str) -> bool: return app in GENERIC_APPS diff --git a/backend/app/integrations.py b/backend/app/integrations.py index 9fbe823..8ff3d94 100644 --- a/backend/app/integrations.py +++ b/backend/app/integrations.py @@ -22,6 +22,12 @@ KEYS = ("singbox_url", "singbox_secret", "proxy_ips", "mosdns_url", "adguard_url "adguard_password") + +def _no_userinfo(url: str) -> str: + """状态接口里的地址去掉 user:pass@（/api/system 只读用户也能看）""" + import re + return re.sub(r"://[^/@]*@", "://", url or "") + def effective(db_values: dict) -> dict: """网页填写的值优先，其次环境变量""" s = get_settings() @@ -131,7 +137,7 @@ class SingboxPoller: self.ok, self.error, self.last = True, "", int(now) def status(self) -> dict: - return {"enabled": self.enabled, "url": self.url, "has_secret": bool(self.secret), "ok": self.ok, + return {"enabled": self.enabled, "url": _no_userinfo(self.url), "has_secret": bool(self.secret), "ok": self.ok, "error": self.error, "last": self.last, "conns": self.conns, "bytes": self.bytes, "version": self.version, "proxy_ip": self.host_ip()} @@ -195,7 +201,7 @@ class MosdnsPoller: self.ok, self.error, self.last = True, "", int(time.time()) def status(self) -> dict: - return {"enabled": self.enabled, "url": self.url, "ok": self.ok, "error": self.error, + return {"enabled": self.enabled, "url": _no_userinfo(self.url), "ok": self.ok, "error": self.error, "last": self.last, "answers": self.answers} @@ -344,7 +350,7 @@ class AdguardPoller: ips.append((val, ttl)) def status(self) -> dict: - return {"enabled": self.enabled, "url": self.url, "ok": self.ok, "error": self.error, + return {"enabled": self.enabled, "url": _no_userinfo(self.url), "ok": self.ok, "error": self.error, "last": self.last, "answers": self.answers} diff --git a/backend/app/main.py b/backend/app/main.py index e6864db..5da2af2 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -193,9 +193,11 @@ async def lifespan(app: FastAPI): if not os.environ.get("ROSDPI_NO_JOBS"): # 测试里不跑后台任务，避免整点落库打乱测试数据 sched.start() print(f"[startup] ROS-DPI {__version__} 已启动，Web 端口 {s.http_port}", flush=True) + from app.security import setup_code + print(f"[startup] 首次设置的安装码：{setup_code()}（只有从公网地址打开安装向导时才需要；内网访问不用）", flush=True) yield if sched.running: - sched.shutdown(wait=False) + sched.shutdown(wait=False) # 不等长任务（历史重算可能跑几分钟）；落库之间由 engine 的落库锁串行 for t in transports: t.close() _safe(lambda: engine.flush(final=True), "flush")() # 关闭前把还没结束的访问也写进访问日志 diff --git a/backend/app/models.py b/backend/app/models.py index 8036919..62ee736 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -1,4 +1,6 @@ """数据表。时间一律存 epoch 秒（BigInteger），分桶起点：分钟/小时按 UTC 对齐，天按配置时区的 0 点。""" +import secrets + from sqlalchemy import BigInteger, Column, Index, Integer, String, Text from app.database import Base @@ -80,7 +82,7 @@ class DnsCache(Base): ip = Column(String(45), primary_key=True) name = Column(String(255), nullable=False) cname = Column(String(255)) - expires = Column(BigInteger, nullable=False) + expires = Column(BigInteger, nullable=False, index=True) # 清理过期条目按它删 class DomainGeo(Base): @@ -89,7 +91,7 @@ class DomainGeo(Base): domain = Column(String(255), primary_key=True) ip = Column(String(45)) cc = Column(String(2)) - updated = Column(BigInteger) + updated = Column(BigInteger, index=True) # 按保留期清理时按它删 class CustomRule(Base): @@ -115,7 +117,9 @@ class User(Base): username = Column(String(64), unique=True, nullable=False) password_hash = Column(String(128), nullable=False) role = Column(String(16), nullable=True) # admin | user；空 = admin（升级前建的账号都是管理员） - token_version = Column(Integer, nullable=True) # 改密码 / 改权限时 +1，旧 token 立即失效 + # 改密码 / 改权限时 +1，旧 token 立即失效。新建账号从随机值开始：删掉 alice 再建一个同名账号时， + # 被删账号手里 ver=0 的旧 token 不能对上新账号 + token_version = Column(Integer, nullable=True, default=lambda: secrets.randbelow(1 << 30) + 1) created = Column(Integer, nullable=True) last_login = Column(Integer, nullable=True) diff --git a/backend/app/notify.py b/backend/app/notify.py index 3b0d82b..b6a4837 100644 --- a/backend/app/notify.py +++ b/backend/app/notify.py @@ -45,7 +45,9 @@ DEFAULTS: dict[str, dict] = { "webhook": {"urls": [], "secret": "", "headers": [], "template": "", "timeout_ms": 5000, "max_retries": 3}, } CHANNEL_TYPES = tuple(DEFAULTS) -SECRET = {"url", "urls", "secret", "bot_token", "password", "token", "sendkey", "app_secret"} +SECRET = {"url", "urls", "secret", "bot_token", "password", "token", "sendkey", "app_secret", "proxy"} +# 改了这些“发往哪里”的字段时，被遮住的密钥不能沿用旧值——否则把地址改成自己的服务器就能套出已保存的密钥 +ADDRESS = {"api_base", "server", "host", "proxy_base", "proxy"} REQUIRED = {"telegram": ["bot_token", "chat_id"], "feishu": ["app_id", "app_secret", "chat_ids"], "qq": ["app_id", "app_secret"], "bark": ["urls"], "email": ["host", "to"], "pushplus": ["token"], "serverchan": ["sendkey"], "dingtalk": ["url"], "discord": ["url"], "slack": ["url"], @@ -146,12 +148,27 @@ def mask(cfg: dict[str, dict]) -> dict[str, dict]: if isinstance(v, str) and len(v) > 12: return v[:12] + "…" return v - return {t: {k: (m(v) if k in SECRET else v) for k, v in c.items()} for t, c in cfg.items()} + def mh(v): + # webhook 自定义请求头的值常带 Authorization: Bearer … + return [{**h, "value": m(h.get("value") or "") if len(h.get("value") or "") > 12 else + ("…" if h.get("value") else "")} if isinstance(h, dict) else h for h in (v or [])] + return {t: {k: (m(v) if k in SECRET else mh(v) if k == "headers" else v) for k, v in c.items()} + for t, c in cfg.items()} def unmask(new: dict, old: dict) -> dict: - """以“…”结尾的值（列表按位置）沿用原值""" + """以“…”结尾的值（列表按位置）沿用原值；发送地址改了时不沿用（要求重新填写密钥）""" out = dict(new) + masked = any((isinstance(new.get(k), str) and new[k].endswith("…")) or + (isinstance(new.get(k), list) and any(isinstance(x, str) and x.endswith("…") for x in new[k])) + for k in SECRET) + if masked and any(str(new.get(k) or "").rstrip("…") != str(old.get(k) or "")[:len(str(new.get(k) or "").rstrip("…"))] + for k in ADDRESS if k in new and not str(new.get(k) or "").endswith("…")): + raise ValueError("改了发送地址，请重新填写密钥 / 令牌") + if isinstance(new.get("headers"), list): + oh = old.get("headers") if isinstance(old.get("headers"), list) else [] + out["headers"] = [{**h, "value": oh[i].get("value", "")} if isinstance(h, dict) and str(h.get("value") or "").endswith("…") + and i < len(oh) and isinstance(oh[i], dict) else h for i, h in enumerate(new["headers"])] for k in SECRET & set(new): v, o = new[k], old.get(k) if isinstance(v, list): @@ -570,7 +587,7 @@ def send(title: str, text: str, types: list[str] | None = None, wait: bool = Fal send_one(t, cfg[t], title, text) r = {"channel": t, "ok": True, "error": ""} except Exception as e: # noqa: BLE001 - r = {"channel": t, "ok": False, "error": f"{type(e).__name__}: {str(e)[:200]}"} + r = {"channel": t, "ok": False, "error": scrub_error(f"{type(e).__name__}: {str(e)[:300]}")[:200]} with _lock: _last[t] = {**r, "ts": int(time.time())} results.append(r) @@ -581,6 +598,17 @@ def send(title: str, text: str, types: list[str] | None = None, wait: bool = Fal return results +def scrub_error(s: str) -> str: + """报错里常带完整 URL（Telegram 的 /bot<令牌>/、Server 酱的 /.send、?access_token=…）， + 会随接口返回给前端：把这些密钥换成 ***""" + import re + s = re.sub(r"/bot[^/\s]+", "/bot***", s) + s = re.sub(r"(?i)((?:access_)?token|key|secret|sendkey|sign)=[^&\s'\"]+", r"\1=***", s) + s = re.sub(r"/[A-Za-z0-9_\-]{24,}(?=[./?\s'\"]|$)", "/***", s) + s = re.sub(r"://[^/@\s]+@", "://***@", s) + return s + + def last_results() -> dict: with _lock: return dict(_last) diff --git a/backend/app/reclass.py b/backend/app/reclass.py index 208628a..3edc8e7 100644 --- a/backend/app/reclass.py +++ b/backend/app/reclass.py @@ -54,7 +54,7 @@ def _retryable(e: Exception) -> bool: """MySQL 死锁（1213）/ 锁等待超时（1205）/ 主键冲突（1062，并发重算撞车）：落库任务每分钟在写同一批表， 自动任务和手动按钮也可能同时跑，撞上重试即可""" s = str(e) - return "(1213" in s or "(1205" in s or "(1062" in s + return "(1213" in s or "(1205" in s or "(1062" in s or "database is locked" in s def reclassify(engine, db, before: int | None = None, attempts: int = 3, since: int | None = None) -> dict: diff --git a/backend/app/routers/audit.py b/backend/app/routers/audit.py index c0aecca..85c928e 100644 --- a/backend/app/routers/audit.py +++ b/backend/app/routers/audit.py @@ -88,11 +88,14 @@ def logs_csv(request: Request, range: str = Query("today", pattern=RANGES), star buf.write("") w = csv.writer(buf) w.writerow(["开始", "结束", "终端", "终端 IP", "访问目标", "远端 IP", "端口", "应用", "大类", "连接数", "上行字节", "下行字节", "识别依据"]) + def safe(v): + # 主机名、访问目标来自网络（设备可以随便设）：以 = + - @ 开头的单元格在 Excel 里会被当成公式执行 + return "'" + v if isinstance(v, str) and v[:1] in ("=", "+", "-", "@", "\t", "\r") else v for r in rows: x = _row(r, meta) - w.writerow([datetime.fromtimestamp(r.ts_start, tz).strftime("%Y-%m-%d %H:%M:%S"), + w.writerow([safe(v) for v in [datetime.fromtimestamp(r.ts_start, tz).strftime("%Y-%m-%d %H:%M:%S"), datetime.fromtimestamp(r.ts_end, tz).strftime("%Y-%m-%d %H:%M:%S"), x["host_name"], r.host_ip, - r.dest, r.remote, r.port, x["app"]["name"], x["app"]["group_name"], r.conns, r.up, r.down, r.source]) + r.dest, r.remote, r.port, x["app"]["name"], x["app"]["group_name"], r.conns, r.up, r.down, r.source]]) name = f"rosdpi-access-{datetime.now(tz).strftime('%Y%m%d-%H%M')}.csv" return StreamingResponse(iter([buf.getvalue()]), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{name}"'}) @@ -265,7 +268,10 @@ class NotifyTest(BaseModel): def _clean(t: str, c: dict, old: dict) -> dict: """只保留该渠道认识的字段，类型按默认值转换；以“…”结尾的密钥沿用原值""" base = notify.DEFAULTS[t] - c = notify.unmask(c, old) + try: + c = notify.unmask(c, old) + except ValueError as e: + raise HTTPException(400, f"{notify_name(t)}：{e}") from None out = {"enabled": bool(c.get("enabled"))} for k, dv in base.items(): v = c.get(k, dv) diff --git a/backend/app/routers/auth.py b/backend/app/routers/auth.py index 23c18f4..aa216d6 100644 --- a/backend/app/routers/auth.py +++ b/backend/app/routers/auth.py @@ -8,7 +8,7 @@ from sqlalchemy.orm import Session from app import oplog from app.database import get_db from app.models import Setting, User -from app.security import (check_login_lock, current_user, forget_roles, make_token, note_login_fail, note_login_ok, +from app.security import (check_login_lock, check_setup_code, current_user, forget_roles, make_token, note_login_fail, note_login_ok, pwd, require_admin, role_of) router = APIRouter(prefix="/api/auth", tags=["auth"]) @@ -38,7 +38,9 @@ def status(db: Session = Depends(get_db)): def setup(req: SetupReq, request: Request, db: Session = Depends(get_db)): if db.query(User).count() > 0: raise HTTPException(400, "管理员已存在") - db.add(User(username=req.username, password_hash=pwd.hash(req.password), role="admin", created=int(time.time()))) + check_setup_code(request) + u = User(username=req.username, password_hash=pwd.hash(req.password), role="admin", created=int(time.time())) + db.add(u) # settings 主键唯一：两个请求同时走到这里时，只有一个能提交成功（防安装窗口期被并发接管） db.add(Setting(key="setup_done", value=str(int(time.time())))) try: @@ -47,7 +49,8 @@ def setup(req: SetupReq, request: Request, db: Session = Depends(get_db)): db.rollback() raise HTTPException(400, "管理员已存在") forget_roles() - return {"token": make_token(req.username), "username": req.username, "role": "admin"} + db.flush() + return {"token": make_token(req.username, u.token_version), "username": req.username, "role": "admin"} @router.post("/login") diff --git a/backend/app/routers/setup.py b/backend/app/routers/setup.py index fd5d734..bd83a58 100644 --- a/backend/app/routers/setup.py +++ b/backend/app/routers/setup.py @@ -2,11 +2,12 @@ 这些接口不需要登录，但只在数据库尚未配置时可用。 """ -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, Field from app import bootstrap, database from app.config import get_settings +from app.security import check_setup_code router = APIRouter(prefix="/api/setup", tags=["setup"]) @@ -42,8 +43,9 @@ def _guard(): @router.post("/test") -def test(cfg: DBConfig): +def test(cfg: DBConfig, request: Request): _guard() + check_setup_code(request) ok, msg = bootstrap.test_connection(cfg.model_dump()) if not ok: raise HTTPException(400, msg) @@ -51,8 +53,9 @@ def test(cfg: DBConfig): @router.post("/save") -def save(cfg: DBConfig): +def save(cfg: DBConfig, request: Request): _guard() + check_setup_code(request) data = cfg.model_dump() ok, msg = bootstrap.test_connection(data) if not ok: diff --git a/backend/app/security.py b/backend/app/security.py index 6b81822..de17705 100644 --- a/backend/app/security.py +++ b/backend/app/security.py @@ -169,3 +169,30 @@ def note_login_fail(request: Request, username: str): def note_login_ok(request: Request, username: str): _login_fails.pop(_fail_key(request, username), None) _user_fails.pop(username, None) + + +# ---------------- 首次设置的安装码 ---------------- +# 安装向导和创建管理员的接口在设置完成前不需要登录：谁先打开网页谁就能把实例接管。内网访问照旧不受影响； +# 从公网地址访问（例如已经通过反向代理暴露出去）时，要填启动日志里打印的一次性安装码 +_setup_code = secrets.token_hex(4) + + +def setup_code() -> str: + return _setup_code + + +def check_setup_code(request: Request): + import hmac + import ipaddress + + from app.oplog import client_ip + ip = client_ip(request) + try: + a = ipaddress.ip_address(ip) + if a.is_private or a.is_loopback or a.is_link_local: + return + except ValueError: + return # 测试客户端等没有真实地址的情况 + code = (request.headers.get("x-setup-code") or "").strip().lower() + if not hmac.compare_digest(code, _setup_code): + raise HTTPException(403, "setup_code_required") diff --git a/backend/tests/test_audit_security.py b/backend/tests/test_audit_security.py new file mode 100644 index 0000000..fdbbd91 --- /dev/null +++ b/backend/tests/test_audit_security.py @@ -0,0 +1,56 @@ +"""全站审计：安全类修复（首次设置安装码、通知密钥遮挡与报错脱敏、删号重建后旧 token 失效、跨午夜时段累计）""" +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app import notify, security + + +def _req(peer, xff=None, code=None): + h = {} + if xff: + h["x-forwarded-for"] = xff + if code: + h["x-setup-code"] = code + return SimpleNamespace(client=SimpleNamespace(host=peer), headers=h) + + +def test_setup_code_only_for_public_visitors(): + security.check_setup_code(_req("192.168.88.20")) # 内网：不需要 + with pytest.raises(HTTPException): + security.check_setup_code(_req("192.168.88.1", xff="8.8.8.8")) # 经反代来自公网：要安装码 + security.check_setup_code(_req("192.168.88.1", xff="8.8.8.8", code=security.setup_code())) + with pytest.raises(HTTPException): + security.check_setup_code(_req("8.8.4.4", xff="192.168.1.5")) # 直连公网伪造 XFF 没用 + + +def test_notify_mask_and_scrub(): + cfg = {"telegram": {"bot_token": "123456789:ABCDEFGHIJKLMN", "api_base": "", "proxy": "http://u:pw@p:1080"}, + "webhook": {"headers": [{"key": "Authorization", "value": "Bearer abcdefghijklmnop"}]}} + m = notify.mask(cfg) + assert m["telegram"]["bot_token"].endswith("…") and m["telegram"]["proxy"].endswith("…") + assert m["webhook"]["headers"][0]["value"].endswith("…") + with pytest.raises(ValueError): # 改了发送地址却沿用遮住的令牌：拒绝 + notify.unmask({**m["telegram"], "api_base": "https://evil.example"}, cfg["telegram"]) + assert notify.unmask(m["webhook"], cfg["webhook"])["headers"][0]["value"] == "Bearer abcdefghijklmnop" + s = notify.scrub_error("ConnectError https://api.telegram.org/bot123:ABC/sendMessage?token=xyz") + assert "123:ABC" not in s and "xyz" not in s + + +def test_new_user_token_version_random(): + from app.models import User + col = User.__table__.c.token_version + vals = {col.default.arg(None) for _ in range(5)} + assert all(v and v > 0 for v in vals) and len(vals) > 1 + + +def test_audit_window_across_midnight(): + from datetime import datetime + + from app import audit, timeutil + rule = SimpleNamespace(time_from="22:00", time_to="06:00", days="1234567") + tz = timeutil.tz() + late = datetime(2026, 10, 25, 23, 30, tzinfo=tz).timestamp() + early = datetime(2026, 10, 26, 1, 30, tzinfo=tz).timestamp() + assert audit.window_start(rule, late) == audit.window_start(rule, early) # 同一段生效时段 diff --git a/frontend/src/csv.js b/frontend/src/csv.js index 77b223d..a081ae6 100644 --- a/frontend/src/csv.js +++ b/frontend/src/csv.js @@ -1,7 +1,9 @@ // 导出 CSV：带 UTF-8 BOM，Excel 直接打开不乱码。columns: [{ label, value: row => any }] export function exportCsv(name, columns, rows) { const cell = v => { - const s = v == null ? '' : String(v) + let s = v == null ? '' : String(v) + // 主机名、域名来自网络：以 = + - @ 开头的文本在 Excel 里会被当成公式执行，前面加 ' 当纯文本 + if (typeof v === 'string' && /^[=+\-@\t\r]/.test(s)) s = "'" + s return /[",\n\r]/.test(s) ? `"${s.replace(/"/g, '""')}"` : s } const lines = [columns.map(c => cell(c.label)).join(',')] diff --git a/frontend/src/i18n/en.js b/frontend/src/i18n/en.js index d7bf97e..c629a4c 100644 --- a/frontend/src/i18n/en.js +++ b/frontend/src/i18n/en.js @@ -39,6 +39,7 @@ export default { peak: 'Peak {v}', mode: { groups: 'By app group', pair: 'Up / down' }, }, setup: { + code: 'Setup code', codeNeeded: 'First-time setup from a public address needs the setup code printed in the container / service startup log ("首次设置的安装码: xxxxxxxx"). Not needed on the LAN.', title: 'Connect to MySQL', intro: 'Statistics are stored in your MySQL. Enter the database and account you prepared for ROS-DPI.', step1: 'Connect database', step2: 'Create admin', host: 'MySQL host', port: 'Port', user: 'User', password: 'Password', database: 'Database', hostHint: 'If MySQL runs on this same machine, use its LAN IP, not 127.0.0.1. Tables are created automatically.', diff --git a/frontend/src/i18n/zh.js b/frontend/src/i18n/zh.js index 519c89b..fef4237 100644 --- a/frontend/src/i18n/zh.js +++ b/frontend/src/i18n/zh.js @@ -39,6 +39,7 @@ export default { peak: '峰值 {v}', mode: { groups: '按应用大类', pair: '上行 / 下行' }, }, setup: { + code: '安装码', codeNeeded: '你正在从公网地址进行首次设置：请填写安装码（在容器 / 服务的启动日志里，形如“首次设置的安装码：xxxxxxxx”）。内网访问不需要。', title: '连接 MySQL 数据库', intro: '统计数据保存在你的 MySQL 中。请填写为 ROS-DPI 准备的数据库和账号。', step1: '连接数据库', step2: '设置管理员', host: 'MySQL 地址', port: '端口', user: '账号', password: '密码', database: '数据库名', hostHint: 'MySQL 和本程序在同一台机器时，请填这台机器的局域网 IP，不要填 127.0.0.1。表会自动创建。', diff --git a/frontend/src/views/Login.vue b/frontend/src/views/Login.vue index 29686d3..d7b4743 100644 --- a/frontend/src/views/Login.vue +++ b/frontend/src/views/Login.vue @@ -9,6 +9,7 @@ +
{{ error }}
@@ -31,6 +32,8 @@ const password = ref('') const password2 = ref('') const error = ref('') const busy = ref(false) +const needCode = ref(false) +const code = ref('') onMounted(async () => { try { setup.value = !(await http.get('/auth/status')).initialized } catch (e) { error.value = e.message } @@ -44,11 +47,13 @@ async function submit() { } busy.value = true try { - const r = await http.post(setup.value ? '/auth/setup' : '/auth/login', { username: username.value, password: password.value }) + const r = await http.post(setup.value ? '/auth/setup' : '/auth/login', { username: username.value, password: password.value }, + code.value ? { headers: { 'X-Setup-Code': code.value } } : undefined) setAuth(r.token, r.username, r.role) router.replace(route.query.next || '/') } catch (e) { - error.value = e.message + // 从公网首次设置：要填启动日志里的安装码 + if (e.message === 'setup_code_required') { needCode.value = true; error.value = t('setup.codeNeeded') } else error.value = e.message } finally { busy.value = false } diff --git a/frontend/src/views/Setup.vue b/frontend/src/views/Setup.vue index 839ac3f..af0495d 100644 --- a/frontend/src/views/Setup.vue +++ b/frontend/src/views/Setup.vue @@ -26,6 +26,7 @@

{{ t('setup.hostHint') }}

+
{{ okMsg }}
{{ error }}
@@ -80,21 +81,28 @@ onMounted(async () => { }) onUnmounted(() => clearInterval(timer)) +const needCode = ref(false) +const code = ref('') +const codeOpt = () => ({ timeout: 60000, headers: code.value ? { 'X-Setup-Code': code.value } : {} }) +function showErr(e) { + if (e.message === 'setup_code_required') { needCode.value = true; error.value = t('setup.codeNeeded') } else error.value = e.message +} + async function test() { error.value = okMsg.value = '' busy.value = 'test' - try { okMsg.value = (await http.post('/setup/test', f, { timeout: 60000 })).message } catch (e) { error.value = e.message } finally { busy.value = '' } + try { okMsg.value = (await http.post('/setup/test', f, codeOpt())).message } catch (e) { showErr(e) } finally { busy.value = '' } } async function save() { error.value = okMsg.value = '' busy.value = 'save' try { - await http.post('/setup/save', f, { timeout: 60000 }) // 首次建表可能较慢 + await http.post('/setup/save', f, codeOpt()) // 首次建表可能较慢 Object.assign(setupState, { configured: true, present: true }) router.replace({ name: 'login' }) } catch (e) { - error.value = e.message + showErr(e) } finally { busy.value = '' } diff --git a/install.sh b/install.sh index 1c32297..a5a7390 100644 --- a/install.sh +++ b/install.sh @@ -49,7 +49,9 @@ chmod 700 "$DIR/data" chmod -R go-rwx "$DIR/data" # 旧版本装出来的库文件是 644，一并收紧 # 用户配置单独放在 EnvironmentFile：只在不存在时创建（全是注释示例），重装 / 升级都不覆盖 +NEW_ENV=0 if [ ! -f "$ENV_FILE" ]; then + NEW_ENV=1 cat > "$ENV_FILE" <<'ENVF' # ROS-DPI 自定义配置：取消注释并修改后执行 systemctl restart ros-dpi # 本文件不会被 install.sh 覆盖；端口和数据目录由服务文件给出，这里设置会覆盖它们 @@ -65,6 +67,12 @@ if [ ! -f "$ENV_FILE" ]; then # ROS_USER= # ROS_PASSWORD= ENVF + # 0.6.21 及以前的配置写在服务文件里（Environment=DATABASE_URL=…）：搬进来，否则重装后会退回 SQLite，看起来像数据全丢了 + if [ "$REINSTALL" = 1 ]; then + grep '^Environment=' /etc/systemd/system/ros-dpi.service \ + | grep -v -e '^Environment="\{0,1\}HTTP_PORT=' -e '^Environment="\{0,1\}DATA_DIR=' \ + | sed -e 's/^Environment=//' -e 's/^"\(.*\)"$/\1/' >> "$ENV_FILE" || true + fi fi chmod 600 "$ENV_FILE"