diff --git a/CHANGELOG.md b/CHANGELOG.md index 9517495..ebce77b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,33 @@ # Changelog +## [0.6.2] - 2026-10-24 + +通知渠道照搬 EasySub、使用场景重做、识别顺序调整（debug12） + +通知渠道（与 EasySub 完全一致） + +- 14 种渠道，磁贴顺序、字段名、发送实现都和 EasySub 相同：Telegram Bot、飞书 Bot、QQ Bot、Bark、Email、Pushplus、Server酱、企业微信、钉钉、Discord、Slack、ntfy、Gotify、Webhook。 +- **Telegram**：新增 ADMIN ID、HTTP 代理；API 反代可以不带 https://；新增「验证机器人」（getMe）和「获取 Chat ID」（getUpdates，自动填入最近一条消息的会话）。 +- **飞书**改为应用机器人（App ID + App Secret + 多个 Chat ID），和 EasySub 一样。**旧的飞书群机器人 Webhook 配置需要重新填写。** +- **新增 QQ 机器人**：App ID / Secret，可同时推送多个群聊和私聊 OpenID。 +- **企业微信**：可选「群机器人」或「自建应用」。自建应用填企业ID、AgentId、Secret，可设 API 代理（应对企业可信 IP 限制），按成员 / 部门 / 标签推送，消息类型支持文本、Markdown、文本卡片（可点击跳转）；新增「检查应用」（返回应用名和可见范围）、access_token 缓存和失效自动重取、接收人无效时报错，以及可展开的配置说明（含 Nginx / socat 反代示例）。 +- Bark 新增图标；Email 改为 SSL/TLS 开关（关闭时自动尝试 STARTTLS）；Pushplus 可选渠道（微信 / 邮件 / Webhook / 企业微信）；Webhook 新增请求超时（毫秒）和最大重试次数，模板支持 event 变量。 +- 0.6.1 及更早保存的配置自动迁移（邮件加密方式、发件人、Webhook 重试次数）。免打扰、规则选择渠道、上次发送结果保留不变。 + +使用场景 + +- 两个场景命名为 **家庭** 和 **企业**。企业场景下菜单、页面、按钮统一称“员工 / 部门”（之前是“人员”），家庭场景是“家庭成员 / 分组”。 +- 场景改为**全局设置**，在 **系统设置 → 场景与网络** 切换（原来藏在人员页的分析设置里）；切换后菜单、人员页、审计模板立即跟着变。侧边栏底部和人员、审计页头显示当前场景。 +- 浏览器记住上次的场景：企业场景刷新页面时，菜单不会再先闪出“家庭成员”。 +- 删除还有人的分组 / 部门时，提示语按场景用“成员 / 分组”或“员工 / 部门”。 + +识别顺序与自动更新 + +- **有域名时先查完所有域名库再按 IP**：国内域名列表移到官方 IP 段之前。国内网站托管在 Cloudflare / AWS 上时归「国内网站」，而不是 IP 所属的 CDN。 +- **IP 段重叠按最长前缀匹配**：AWS 中国区段落在国内 IP 段里、GitHub 段落在 AWS 段里时，更具体的小段优先；修复了小段之外、仍属于大段的地址查不到的问题。 +- 「识别库」面板按匹配顺序编号，并写明完整的匹配顺序。 +- 「自动更新」开关现在同时控制 ASN 库和补充库（国内域名列表、官方 IP 段）的定期下载；补充库每 6 小时检查一次、满一天才重新下载（之前 24 小时一次，可能因几秒误差拖到两天）。 + ## [0.6.1] - 2026-10-24 识别率（未识别占比较高，补充更强的识别数据源） diff --git a/DOCKERHUB.md b/DOCKERHUB.md index 19a11af..356f58f 100644 --- a/DOCKERHUB.md +++ b/DOCKERHUB.md @@ -1,6 +1,6 @@ # ROS-DPI -RouterOS 终端流量、应用统计与上网审计面板。按 今日 / 昨日 / 本周 / 本月 统计每台终端、每个人的流量和应用使用情况（类似 Panabit / 维盟），记录访问日志并按规则告警（参照深信服 AC）；分家用和企业两种场景。 +RouterOS 终端流量、应用统计与上网审计面板。按 今日 / 昨日 / 本周 / 本月 统计每台终端、每个人的流量和应用使用情况（类似 Panabit / 维盟），记录访问日志并按规则告警（参照深信服 AC）；分家庭和企业两种场景。 Per-device and per-person traffic, per-application (DPI-style) statistics and access auditing for MikroTik RouterOS, built on IPFIX + DNS / SNI sniffing. @@ -8,6 +8,7 @@ Source & docs: https://github.com/suyijun8182/ros-dpi ## What's new +- **0.6.2**: notification channels now match EasySub exactly — 14 channels with the same fields: Telegram (API proxy, HTTP proxy, check bot / get Chat ID), Feishu app bot, new QQ bot, Bark icon, email SSL switch, Pushplus channel, WeCom group robot **or self-built app** (CorpID / AgentId / Secret, API proxy, user / department / tag, text card, "check app"), webhook timeout and retries. The scenario (Home / Business) is now a global setting under **Settings → Scenario & network**; in Business mode the menu says "Employees" instead of "Family members", and the last scenario is cached so the menu no longer flashes the wrong label. Identification order: domain fallbacks (China domain list) are tried before IP-range fallbacks, and overlapping IP ranges use longest-prefix match (before, a small range inside a big one could hide the big one). The "Auto update" switch now also controls the ASN and fallback list downloads. Note: Feishu now uses an app bot (App ID / Secret / Chat IDs) like EasySub; an old Feishu group-webhook config has to be re-entered. - **0.6.1**: much higher identification rate. New fallback sources, all downloaded at runtime and auto-updated daily: the dnsmasq-china-list Chinese domain list (~110k domains, WTFPL) catches long-tail .cn domains as "Chinese websites" (sampling showed v2fly + built-in rules only matched 5.6% of them); official Google / Cloudflare / AWS / GitHub IP ranges plus Apple's 17.0.0.0/8 attribute domain-less traffic (DoH, ECH, raw IPs) to the owning company; China IP ranges (chnroutes2, CC-BY-SA 4.0) group domain-less domestic traffic as "China networks". Purely additive — existing rules always take priority. Disable with `CHINA_DB=false` / `IPLIST_DB=false`. - **0.6.0**: reliability & hardening. A failed database write no longer loses that minute's traffic and access log (kept in memory and retried, giving up only after 2 hours); retention cleanup deletes in batches of 5000 rows instead of one long transaction, and `domain_geo` is now covered. Security: login rate limiting (5 failures locks the IP+username for 15 min), tokens are revoked immediately when a password or role changes, setup race protected, 500 errors no longer echo exception details. Performance: destination ranking aggregated in SQL (new `base` column for merging by registered domain, auto-backfilled), the people page scans the access log once instead of twice, the overview reuses today's app ranking. Frontend: polling race fixed (stale responses can no longer overwrite fresh data) and a dozen loads that could spin forever on error now show the error. CI builds the image on PRs; the container process runs as a non-root user; the bundled-MySQL compose now requires real passwords. - **0.5.1**: people must belong to a group / department (no duplicates within a group; groups with members can't be deleted), and existing people can be edited and deleted again. Countries are shown on a flow map from your location with a ranked list; a domain's country is now stored when traffic is recorded, so it no longer turns "unlocated" after a restart. README brought up to date. @@ -90,7 +91,7 @@ Open `http://:8080`. The setup wizard has two steps. 打开网页后按两 - Change database later: delete `./data/db_config.json` and restart. 更换数据库：删除该文件后重启容器。 - Setting `DATABASE_URL` skips the wizard. 设置环境变量 `DATABASE_URL` 可跳过向导。 - RouterOS commands: **Settings → Setup & check / 系统设置 → 接入与自检**. Full guide: https://github.com/suyijun8182/ros-dpi -- Then choose the scenario under **People → Settings**, assign devices to people, and set up notifications under **Audit → Notifications**. 之后在 家庭成员 / 人员 → 分析设置 选择场景，把设备分给成员，在 审计日志 → 通知渠道 配置推送。 +- Then choose the scenario under **Settings → Scenario & network**, assign devices to people, and set up notifications under **Audit → Notifications**. 之后在 系统设置 → 场景与网络 选择场景（家庭 / 企业），把设备分给成员或员工，在 审计日志 → 通知渠道 配置推送。 ## Ports diff --git a/README.md b/README.md index edc4c17..acceb02 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # ROS-DPI -**RouterOS 终端流量、应用统计与上网审计面板**：像 Panabit / 维盟那样，按 今日 / 昨日 / 本周 / 本月 统计 RouterOS 下每台终端、每个人用了多少流量、用在了哪些应用上；像深信服 AC 那样记录访问日志、按规则告警。支持家用（孩子上网健康）和企业（效率与合规）两种场景。 +**RouterOS 终端流量、应用统计与上网审计面板**：像 Panabit / 维盟那样，按 今日 / 昨日 / 本周 / 本月 统计 RouterOS 下每台终端、每个人用了多少流量、用在了哪些应用上；像深信服 AC 那样记录访问日志、按规则告警。支持家庭（孩子上网健康）和企业（效率与合规）两种场景。 [中文](#中文) · [English](#english) @@ -16,13 +16,13 @@ - **主机监控**：在线终端 / 全部终端 / DHCP 租约 / 命名管理四个标签。每台终端的类型（手机、电脑、电视、摄像头……自动识别）、IP、MAC、主机名、归属人、在线时长、连接数、当前速率、今日 / 本周 / 本月流量，可导出 CSV；可以按 MAC 或按 IP 给设备命名。详情页有应用排行与构成、按大类的趋势、访问过的域名、访问日志和上网时间线。 - **应用统计**：参照 Panabit / 维盟分三级（应用大类 → 小类 → 应用，逐级“查看详情”下钻）。大类和小类固定列出，每级显示上传 / 下载速度和数据及占比；应用构成用矩形树图一眼看两层。共 10 个大类、43 个小类、2800 多个内置应用（抖音、B 站、微信视频号、Steam、银行、券商、ToDesk、DeepSeek……，其中约 2500 个来自 v2fly 开源域名库）。识别不出具体应用的流量，会按 CDN 的 CNAME 归到阿里云 / 腾讯云 / Akamai 等厂商。 - **流量流向**：**实时连接**（此刻哪台终端连到哪个域名 / IP、什么应用、多快；走 sing-box 的连接显示策略组和出口节点）、**访问目标排行**（流量最大的远端域名和 IP；流向图从本地飞向各国，看流量去了哪些国家 / 地区；可按主域名合并，看是哪些终端在访问）、**代理分流**（sing-box 各策略组、各节点的流量和主要应用）。 -- **家庭成员 / 人员**：把一个人的手机、平板、电脑归到一起，按人统计流量、在线时长、常用应用，以及最近 7 天的行为标签（附依据）。按使用场景： - - **家用**：关注孩子的上网健康，包括每日屏幕时间、游戏和短视频时长上限，熬夜，不良内容，学习时间玩娱乐应用。 - - **企业**：关注效率与合规，包括工作时间娱乐占比、求职倾向、上班炒股，按部门统计。 +- **家庭成员 / 员工**：把一个人的手机、平板、电脑归到一起，按人统计流量、在线时长、常用应用，以及最近 7 天的行为标签（附依据）。使用场景在 **系统设置 → 场景与网络** 全局切换，菜单、称呼、默认时间段、标签和审计模板跟着变： + - **家庭**：菜单显示“家庭成员”，按分组管理；关注孩子的上网健康，包括每日屏幕时间、游戏和短视频时长上限，熬夜，不良内容，学习时间玩娱乐应用。 + - **企业**：菜单显示“员工”，按部门管理；关注效率与合规，包括工作时间娱乐占比、求职倾向、上班炒股，按部门统计。 - **审计日志**： - **访问日志**：每台终端什么时候访问了什么、多久、多少流量，可按人员 / 部门 / 关键字筛选，导出 CSV，默认保留 30 天。 - **审计规则**：按应用、域名、时段、流量、**大流量外发**、**私接路由**、新设备设置，可限定人员 / 部门，带按场景的常用模板。命中后推送通知。 - - **通知渠道**：企业微信、钉钉、飞书、Bark、Telegram、邮件、PushPlus、Server酱、ntfy、Gotify、Discord、Slack、通用 Webhook。每条规则可以选渠道，可设免打扰。 + - **通知渠道**（与 EasySub 相同的 14 种）：Telegram Bot（可填 API 反代、HTTP 代理，一键获取 Chat ID）、飞书 Bot、QQ Bot、Bark、Email、Pushplus、Server酱、企业微信（群机器人或自建应用，支持 API 代理、指定成员 / 部门 / 标签、文本卡片）、钉钉、Discord、Slack、ntfy、Gotify、通用 Webhook。每条规则可以选渠道，可设免打扰。 - **私接检测**：根据数据包 TTL 发现后面接了路由器或多种系统共享的设备。 - **日报 / 周报**和**操作日志**。 - **管控命令**：在人员或终端详情里，按应用、时段生成 RouterOS 命令（封禁应用、定时断网、限速），附撤销命令。ROS-DPI 只生成命令，不会修改路由器。 @@ -151,7 +151,7 @@ MySQL 第一次初始化需要 20～60 秒。 接下来可以： -1. 在 **家庭成员 / 人员 → 分析设置** 选择使用场景（家用 / 企业），设置学习或工作时间、每日时长上限； +1. 在 **系统设置 → 场景与网络** 选择使用场景（家庭 / 企业），再到 **家庭成员 / 员工 → 分析设置** 设置学习或工作时间、每日时长上限； 2. 添加成员并把设备分给他们； 3. 在 **审计日志 → 通知渠道** 配置推送，在 **审计规则** 里从模板新建规则； 4. 需要只读账号时，在 **系统设置 → 用户管理** 添加普通用户。 @@ -261,13 +261,13 @@ backend/.venv/bin/python scripts/simulate.py --minutes 5 - **Hosts**: every device with its detected type (phone, computer, TV, camera…), IP, MAC, hostname, owner, online state, live rate and daily / weekly / monthly totals. Name devices by MAC or by IP. Each device has an app breakdown, visited domains, an access log and a daily timeline. - **Applications**: a three-level drill-down (app group → category → app) with fixed groups and categories, upload/download rates and totals with shares at each level, and a treemap of the mix. It covers 10 groups, 43 categories and 2800+ built-in apps (about 2500 converted from v2fly/domain-list-community). Traffic that can't be tied to an app is attributed to its CDN via CNAME. - **Flows**: live connections (sing-box connections show their policy group and outbound node), top destinations with a flow map from your location to each country, and proxy routing (traffic and top apps per sing-box policy group and node). -- **People / Family**: group each person's devices and see their traffic, online time, top apps and behaviour tags for the last 7 days, with the evidence behind each tag. +- **Family members / Employees**: pick the Home or Business scenario under **Settings → Scenario & network**; menus, wording, default hours, tags and audit templates follow it. Group each person's devices and see their traffic, online time, top apps and behaviour tags for the last 7 days, with the evidence behind each tag. - **Home**: daily screen-time, gaming and short-video limits, late nights, adult content, and leisure during homework time. - **Business**: leisure share in working hours, job hunting and trading at work, with departments. - **Audit**: - **Access log**: filter by person, department or keyword, export to CSV; kept 30 days by default. - **Audit rules**: by app, domain, time window, traffic, **large upload**, **rogue router** or new device; can be scoped to people or departments, with scenario-specific templates. - - **Notifications**: WeCom, DingTalk, Feishu, Bark, Telegram, email, PushPlus, ServerChan, ntfy, Gotify, Discord, Slack or webhooks, with per-rule channel choice and quiet hours. + - **Notifications** (the same 14 channels as EasySub): Telegram bot (API proxy, HTTP proxy, one-click Chat ID), Feishu bot, QQ bot, Bark, email, Pushplus, ServerChan, WeCom (group robot or self-built app with API proxy, user / department / tag targets and text cards), DingTalk, Discord, Slack, ntfy, Gotify or webhooks, with per-rule channel choice and quiet hours. - **Shared-access check**: TTL-based detection of devices with a router behind them. - **Daily / weekly reports** and an **operation log**. - **Control commands**: generate RouterOS commands that block apps, block internet on a schedule or set rate limits for a person or device, with undo commands. ROS-DPI never writes to the router. diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 43c4ab0..22049ab 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -1 +1 @@ -__version__ = "0.6.1" +__version__ = "0.6.2" diff --git a/backend/app/classify/__init__.py b/backend/app/classify/__init__.py index a72c70a..90daacf 100644 --- a/backend/app/classify/__init__.py +++ b/backend/app/classify/__init__.py @@ -2,7 +2,8 @@ 分三级：应用大类（group）→ 应用小类（category）→ 应用（app），与 Panabit 的应用分类方式一致。 优先级：自定义规则 > 域名（先查询名再 CNAME；精确匹配 > 最长后缀）> CIDR > 专用端口 > IP 所属 ASN -> 官方 IP 段 / 国内域名列表（app/classify/sources.py，兜底）> 通用端口（HTTPS 等）> IP 协议号 > other。 +> 国内域名列表 > 官方 IP 段（最长前缀优先）/ 国内 IP 段（app/classify/sources.py，兜底） +> 通用端口（HTTPS 等）> IP 协议号 > other。 内置规则两份：手工维护的 apps.yaml，以及从 v2fly/domain-list-community 生成的 apps_v2fly.yaml（同一域名以前者为准）。 """ import ipaddress @@ -183,13 +184,12 @@ class Classifier: if hit: app = self.asn_apps.get(hit[0]) if app is None: - # 官方 IP 段兜底：完全没有域名的流量（加密 DNS、ECH、直连 IP）按 Google/Cloudflare/Apple/国内段 认归属 - from app.classify.sources import sources as extra - app = extra.ip_app(remote_ip) - if app is None: - # 国内域名兜底：域名规则没认出、但在国内域名列表里 → 国内网站（好于落到 HTTPS/其他） from app.classify.sources import sources as extra + # 先按域名兜底、再按 IP：有域名时域名更准（国内网站托管在 Cloudflare/AWS 上时，IP 只能说明是哪家 CDN） + # 国内域名列表：域名规则没认出、但在列表里 → 国内网站 app = extra.fallback_domain(qname) or extra.fallback_domain(cname) + # 官方 IP 段：完全没有域名的流量（加密 DNS、ECH、直连 IP）按 Google/Cloudflare/AWS/Apple/国内段 认归属 + app = app or extra.ip_app(remote_ip) if app is None: app = generic or self.protos.get(proto) app = app or UNKNOWN_APP diff --git a/backend/app/classify/sources.py b/backend/app/classify/sources.py index e3e9021..0b7d8c5 100644 --- a/backend/app/classify/sources.py +++ b/backend/app/classify/sources.py @@ -11,6 +11,7 @@ 都是运行时下载到数据目录（不进镜像）、每天检查一次更新；单个来源失败不影响其它来源。 环境变量 CHINA_DB / IPLIST_DB 可分别关闭（默认开）。 """ +import heapq import ipaddress import json import os @@ -62,11 +63,16 @@ def _parse_github(text: str) -> list[str]: class _Ranges: - """CIDR → 排序区间数组（bisect 查找）。重叠时先加入的赢：官方段在前，国内段在后。""" + """CIDR → 不重叠的排序区间数组（bisect 查找）。 + + 段之间会重叠（AWS 中国区的段落在 chnroutes 的大段里、GitHub 的段落在 AWS 段里）， + 和路由表一样按最长前缀匹配：更小（更具体）的段赢，一样大时先加入的赢。 + """ def __init__(self): self._v4 = (array("Q"), array("Q"), []) self._v6 = ([], [], []) + self._raw = {4: [], 6: []} self.count = 0 def add(self, cidr: str, app: str): @@ -74,22 +80,42 @@ class _Ranges: net = ipaddress.ip_network(cidr, strict=False) except ValueError: return False - t = self._v4 if net.version == 4 else self._v6 - t[0].append(int(net.network_address)) - t[1].append(int(net.broadcast_address)) - t[2].append(app) + raw = self._raw[net.version] + raw.append((int(net.network_address), int(net.broadcast_address), net.num_addresses, len(raw), app)) self.count += 1 return True + @staticmethod + def _flatten(raw: list) -> list[tuple[int, int, str]]: + """扫描线：在每个边界点之间取覆盖它的优先级最高的段，再合并相邻同归属的区间。""" + cuts = sorted({x for lo, hi, *_ in raw for x in (lo, hi + 1)}) + by_lo = sorted(raw) + heap, out, j = [], [], 0 + for a, b in zip(cuts, cuts[1:]): + while j < len(by_lo) and by_lo[j][0] <= a: + lo, hi, size, idx, app = by_lo[j] + heapq.heappush(heap, (size, idx, hi, app)) + j += 1 + while heap and heap[0][2] < a: + heapq.heappop(heap) + if not heap: + continue + app = heap[0][3] + if out and out[-1][2] == app and out[-1][1] == a - 1: + out[-1] = (out[-1][0], b - 1, app) + else: + out.append((a, b - 1, app)) + return out + def build(self): - for t in (self._v4, self._v6): - order = sorted(range(len(t[0])), key=lambda i: t[0][i]) - s0, s1 = [t[0][i] for i in order], [t[1][i] for i in order] - if hasattr(t[0], "typecode"): # v4 是 array("Q")，切片赋值只接受 array - t[0][:], t[1][:] = array("Q", s0), array("Q", s1) - else: # v6 是 list - t[0][:], t[1][:] = s0, s1 - t[2][:] = [t[2][i] for i in order] + for ver, t in ((4, self._v4), (6, self._v6)): + segs = self._flatten(self._raw[ver]) + if ver == 4: # v4 用 array("Q") 省内存 + t[0][:], t[1][:] = array("Q", [s[0] for s in segs]), array("Q", [s[1] for s in segs]) + else: + t[0][:], t[1][:] = [s[0] for s in segs], [s[1] for s in segs] + t[2][:] = [s[2] for s in segs] + self._raw = {4: [], 6: []} def lookup(self, ip: str) -> str | None: try: diff --git a/backend/app/main.py b/backend/app/main.py index b48d892..cb622d7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -46,17 +46,30 @@ def _on_db_ready(): print("[startup] 数据库已连接", flush=True) -def _auto_update_rules(): - """识别库自动更新（默认开启，可在识别规则页关闭）；更新器内部保证一天最多检查一次""" +def _auto_update_on() -> bool: + """识别规则页的「自动更新」开关（默认开启）：管 v2fly、ASN 库和补充库的定期更新""" if not database.is_configured(): - return - from app.classify.updater import updater + return False with database.session() as db: row = db.get(Setting, "rules_auto_update") - if row is None or row.value != "0": + return row is None or row.value != "0" + + +def _auto_update_rules(): + """v2fly 规则自动更新；更新器内部保证一天最多检查一次""" + if _auto_update_on(): + from app.classify.updater import updater updater.check(reload=engine.reload_classifier) +def _auto_update_db(fn): + """ASN / 补充库的定期下载跟随同一个开关；关掉后只用已下载的数据，手动「立即检查更新」仍然可用""" + def run(): + if _auto_update_on(): + fn() + return run + + def _reconnect(): # 有配置但连接断开 / 启动时没连上：后台定期重试，不需要用户重走安装向导 if not database.is_configured() and database.configured_url(): @@ -94,12 +107,14 @@ async def lifespan(app: FastAPI): asyncio.get_running_loop().call_later(120, lambda: asyncio.get_running_loop().run_in_executor( None, _safe(_auto_update_rules, "rules-update"))) if s.asn_db: - sched.add_job(_safe(asndb.update, "asn-db"), "interval", hours=6, max_instances=1, coalesce=True) + sched.add_job(_safe(_auto_update_db(asndb.update), "asn-db"), "interval", hours=6, max_instances=1, coalesce=True) asyncio.get_running_loop().run_in_executor(None, _safe(asndb.update, "asn-db")) # 补充识别库：国内域名列表 + 官方 IP 段（sources.py，只做兜底；没配置的来源直接跳过） from app.classify.sources import sources as extra_sources if s.china_db or s.iplist_db: - sched.add_job(_safe(extra_sources.update, "extra-db"), "interval", hours=24, max_instances=1, coalesce=True) + # 每 6 小时看一次，文件满一天才重新下载（间隔和过期时间都是 24 小时会因几秒误差拖成两天） + sched.add_job(_safe(_auto_update_db(extra_sources.update), "extra-db"), "interval", hours=6, + max_instances=1, coalesce=True) asyncio.get_running_loop().run_in_executor(None, _safe(extra_sources.update, "extra-db")) if poller.enabled: sched.add_job(_safe(poller.sample, "ros-rate"), "interval", seconds=max(s.ros_rate_seconds, 2), diff --git a/backend/app/notify.py b/backend/app/notify.py index 37e6746..3b0d82b 100644 --- a/backend/app/notify.py +++ b/backend/app/notify.py @@ -1,4 +1,4 @@ -"""通知渠道（参照 EasySub：每种渠道一份配置，磁贴选择、可用未保存的配置测试）。 +"""通知渠道（照搬 EasySub：14 种渠道、字段名、发送实现都和 EasySub 一致，磁贴选择、可用未保存的配置测试）。 配置保存在 settings 表 notify_config：{渠道: {enabled, ...字段}}；免打扰在 notify_prefs：{quiet_from, quiet_to}。 返回给前端时密钥类字段只显示开头并以“…”结尾，保存时以“…”结尾的值表示沿用原值。 @@ -8,36 +8,48 @@ import hashlib import hmac import json import re +import smtplib +import ssl import threading import time import urllib.parse from datetime import datetime +from email.header import Header +from email.mime.text import MIMEText +from email.utils import formataddr import httpx from app import database from app.models import Setting +APP_NAME = "ROS-DPI" + +# 渠道与字段同 EasySub（顺序即磁贴顺序） DEFAULTS: dict[str, dict] = { - "wecom": {"url": "", "msg_type": "text"}, # 企业微信群机器人 - "dingtalk": {"url": "", "secret": ""}, # 钉钉群机器人，secret 为加签密钥 - "feishu": {"url": "", "secret": ""}, # 飞书群机器人，secret 为签名校验密钥 - "bark": {"urls": [], "group": "ROS-DPI", "level": "active"}, - "telegram": {"bot_token": "", "chat_id": "", "api_base": "https://api.telegram.org"}, - "email": {"host": "", "port": 465, "security": "ssl", "username": "", "password": "", "from_addr": "", "to": ""}, - "pushplus": {"token": "", "topic": ""}, + "telegram": {"bot_token": "", "chat_id": "", "admin_id": "", "api_base": "", "proxy": ""}, + "feishu": {"app_id": "", "app_secret": "", "chat_ids": ""}, + "qq": {"app_id": "", "app_secret": "", "group_ids": "", "user_ids": ""}, + "bark": {"urls": [], "group": APP_NAME, "level": "active", "icon": ""}, + "email": {"host": "", "port": 465, "ssl": True, "username": "", "password": "", "from": "", "to": ""}, + "pushplus": {"token": "", "topic": "", "channel": "wechat"}, "serverchan": {"sendkey": ""}, - "ntfy": {"server": "https://ntfy.sh", "topic": "", "token": ""}, - "gotify": {"server": "", "token": "", "priority": 5}, + # 企业微信：mode=webhook 群机器人；mode=app 自建应用（企业ID + 应用ID + Secret，可走 API 代理） + "wecom": {"mode": "webhook", "url": "", "corp_id": "", "agent_id": "", "secret": "", "proxy_base": "", + "to_user": "@all", "to_party": "", "to_tag": "", "msg_type": "text", "card_url": ""}, + "dingtalk": {"url": "", "secret": ""}, "discord": {"url": ""}, "slack": {"url": ""}, - "webhook": {"urls": [], "secret": "", "headers": [], "template": "", "retries": 2}, + "ntfy": {"server": "https://ntfy.sh", "topic": "", "token": ""}, + "gotify": {"server": "", "token": "", "priority": 5}, + "webhook": {"urls": [], "secret": "", "headers": [], "template": "", "timeout_ms": 5000, "max_retries": 3}, } CHANNEL_TYPES = tuple(DEFAULTS) -SECRET = {"url", "urls", "secret", "bot_token", "password", "token", "sendkey"} -REQUIRED = {"wecom": ["url"], "dingtalk": ["url"], "feishu": ["url"], "bark": ["urls"], "telegram": ["bot_token", "chat_id"], - "email": ["host", "to"], "pushplus": ["token"], "serverchan": ["sendkey"], "ntfy": ["topic"], - "gotify": ["server", "token"], "discord": ["url"], "slack": ["url"], "webhook": ["urls"]} +SECRET = {"url", "urls", "secret", "bot_token", "password", "token", "sendkey", "app_secret"} +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"], + "ntfy": ["topic"], "gotify": ["server", "token"], "webhook": ["urls"]} _last: dict[str, dict] = {} # 渠道 -> 上次发送结果 {ok, error, ts} _lock = threading.Lock() @@ -53,6 +65,21 @@ def _get(key: str, default): return default +def _migrate(t: str, c: dict) -> dict: + """0.5.x / 0.6.x 的字段名改成 EasySub 的""" + c = dict(c) + if t == "email": + if "security" in c: + c.setdefault("ssl", c.pop("security") == "ssl") + if "from_addr" in c: + c.setdefault("from", c.pop("from_addr")) + elif t == "webhook" and "retries" in c: + c.setdefault("max_retries", c.pop("retries")) + elif t == "telegram" and c.get("api_base") == "https://api.telegram.org": + c["api_base"] = "" + return c + + def load_config() -> dict[str, dict]: """各渠道配置（补齐默认字段）。兼容 0.4.0 的通道列表格式 notify_channels""" cfg = _get("notify_config", None) @@ -64,18 +91,21 @@ def load_config() -> dict[str, dict]: continue if t == "telegram": cfg[t] = {"bot_token": c.get("token", ""), "chat_id": c.get("chat_id", "")} - elif t == "bark": - cfg[t] = {"urls": [c.get("url", "")]} - elif t == "webhook": + elif t in ("bark", "webhook"): cfg[t] = {"urls": [c.get("url", "")]} elif t == "email": host, _, port = (c.get("url") or "").partition(":") - cfg[t] = {"host": host, "port": int(port or 465), "security": c.get("security", "ssl"), + cfg[t] = {"host": host, "port": int(port or 465), "ssl": c.get("security", "ssl") == "ssl", "username": c.get("user", ""), "password": c.get("token", ""), "to": c.get("to", "")} - else: + elif t != "feishu": # 飞书改成应用机器人，旧的群机器人地址用不上 cfg[t] = {"url": c.get("url", "")} - cfg[t]["enabled"] = c.get("enabled", True) - return {t: {"enabled": False, **DEFAULTS[t], **(cfg.get(t) or {})} for t in DEFAULTS} + if t in cfg: + cfg[t]["enabled"] = c.get("enabled", True) + out = {} + for t, d in DEFAULTS.items(): + saved = _migrate(t, cfg.get(t) or {}) + out[t] = {"enabled": bool(saved.get("enabled")), **d, **{k: v for k, v in saved.items() if k in d}} + return out def enabled_types() -> list[str]: @@ -83,7 +113,16 @@ def enabled_types() -> list[str]: def missing(t: str, c: dict) -> list[str]: - return [f for f in REQUIRED[t] if not (c.get(f) and (c[f] != [] and c[f] != [""]))] + def empty(v): + return not v or v == [] or v == [""] + if t == "wecom": + need = ["corp_id", "agent_id", "secret"] if c.get("mode") == "app" else ["url"] + else: + need = REQUIRED[t] + out = [f for f in need if empty(c.get(f))] + if t == "qq" and not out and empty(c.get("group_ids")) and empty(c.get("user_ids")): + out.append("group_ids / user_ids") + return out def prefs() -> dict: @@ -123,132 +162,400 @@ def unmask(new: dict, old: dict) -> dict: return out -# ---------------- 发送 ---------------- +# ---------------- 发送（各渠道实现同 EasySub，失败抛异常） ---------------- -def _check(r: httpx.Response, ok_codes=(0, 200, None)): - r.raise_for_status() - try: - j = r.json() - except ValueError: +def _split(csv: str) -> list[str]: + return [x.strip() for x in (csv or "").replace("，", ",").split(",") if x.strip()] + + +def _body(subject: str, text: str) -> str: + return f"{subject}\n\n{text}" if subject else text + + +def _tg_url(conf: dict, method: str) -> str: + base = (conf.get("api_base") or "https://api.telegram.org").strip().rstrip("/") + if not base.startswith(("http://", "https://")): + base = "https://" + base + return f"{base}/bot{conf.get('bot_token')}/{method}" + + +def _tg_client(conf: dict) -> httpx.Client: + return httpx.Client(timeout=15, proxy=conf.get("proxy") or None) + + +def _tg_json(r: httpx.Response) -> dict: + data = r.json() + if not data.get("ok"): + raise RuntimeError(data.get("description") or f"HTTP {r.status_code}") + return data + + +def _send_telegram(conf: dict, subject: str, text: str) -> None: + if not conf.get("bot_token") or not conf.get("chat_id"): + raise RuntimeError("Telegram 未配置 Bot Token 或 Chat ID") + with _tg_client(conf) as c: + _tg_json(c.post(_tg_url(conf, "sendMessage"), json={"chat_id": conf["chat_id"], "text": _body(subject, text)})) + + +def telegram_call(conf: dict, method: str) -> dict: + """「验证机器人」（getMe）和「获取 Chat ID」（getUpdates）""" + if not conf.get("bot_token"): + raise RuntimeError("请先填写 Bot Token") + with _tg_client(conf) as c: + return _tg_json(c.get(_tg_url(conf, method))) + + +def _send_feishu(conf: dict, subject: str, text: str) -> None: + app_id, secret = conf.get("app_id"), conf.get("app_secret") + chat_ids = _split(conf.get("chat_ids", "")) + if not app_id or not secret or not chat_ids: + raise RuntimeError("飞书未配置 App ID / Secret / Chat IDs") + with httpx.Client(timeout=15) as c: + r = c.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", + json={"app_id": app_id, "app_secret": secret}) + r.raise_for_status() + token = r.json().get("tenant_access_token") + if not token: + raise RuntimeError(f"飞书获取 token 失败：{r.text[:200]}") + for cid in chat_ids: + data = c.post("https://open.feishu.cn/open-apis/im/v1/messages", params={"receive_id_type": "chat_id"}, + headers={"Authorization": f"Bearer {token}"}, + json={"receive_id": cid, "msg_type": "text", + "content": json.dumps({"text": _body(subject, text)}, ensure_ascii=False)}).json() + if data.get("code", 0) != 0: + raise RuntimeError(f"飞书发送失败（{cid}）：{data.get('msg')}") + + +def _send_qq(conf: dict, subject: str, text: str) -> None: + app_id, secret = conf.get("app_id"), conf.get("app_secret") + groups, users = _split(conf.get("group_ids", "")), _split(conf.get("user_ids", "")) + if not app_id or not secret: + raise RuntimeError("QQ 未配置 App ID / Secret") + if not groups and not users: + raise RuntimeError("QQ 未填写任何群聊或私聊 OpenID") + with httpx.Client(timeout=15) as c: + r = c.post("https://bots.qq.com/app/getAppAccessToken", json={"appId": app_id, "clientSecret": secret}) + r.raise_for_status() + token = r.json().get("access_token") + if not token: + raise RuntimeError(f"QQ 获取 access_token 失败：{r.text[:200]}") + headers = {"Authorization": f"QQBot {token}", "Content-Type": "application/json"} + for kind, ids in (("groups", groups), ("users", users)): + for oid in ids: + resp = c.post(f"https://api.sgroup.qq.com/v2/{kind}/{oid}/messages", headers=headers, + json={"content": _body(subject, text), "msg_type": 0}) + if resp.status_code >= 300: + raise RuntimeError(f"QQ {'群' if kind == 'groups' else '用户'} {oid} 发送失败：{resp.text[:200]}") + + +def _send_bark(conf: dict, subject: str, text: str) -> None: + urls = [u for u in (conf.get("urls") or []) if u] + if not urls: + raise RuntimeError("Bark 未配置任何目标 URL") + payload = {"title": subject or APP_NAME, "body": text} + for k in ("group", "level", "icon"): + if conf.get(k): + payload[k] = conf[k] + with httpx.Client(timeout=15) as c: + for url in urls: + resp = c.post(url.rstrip("/"), json=payload) + resp.raise_for_status() + if resp.json().get("code", 200) not in (200, 0): + raise RuntimeError(f"Bark 推送失败：{resp.text[:200]}") + + +def _send_email(conf: dict, subject: str, text: str) -> None: + host = conf.get("host") + to_list = [x.strip() for x in re.split(r"[,，;；]", conf.get("to") or "") if x.strip()] + sender = (conf.get("from") or conf.get("username") or "").strip() + if not host or not to_list or not sender: + raise RuntimeError("Email 未配置 SMTP 主机 / 发件人 / 收件人") + port = int(conf.get("port") or (465 if conf.get("ssl") else 587)) + msg = MIMEText(text, "plain", "utf-8") + msg["Subject"] = Header(subject or f"{APP_NAME} 通知", "utf-8") + msg["From"] = formataddr((APP_NAME, sender)) + msg["To"] = ", ".join(to_list) + user, pwd = conf.get("username"), conf.get("password") + if conf.get("ssl"): + with smtplib.SMTP_SSL(host, port, timeout=20, context=ssl.create_default_context()) as s: + if user: + s.login(user, pwd) + s.sendmail(sender, to_list, msg.as_string()) + else: + with smtplib.SMTP(host, port, timeout=20) as s: + try: + s.starttls(context=ssl.create_default_context()) + except smtplib.SMTPException: + pass + if user: + s.login(user, pwd) + s.sendmail(sender, to_list, msg.as_string()) + + +def _send_pushplus(conf: dict, subject: str, text: str) -> None: + if not conf.get("token"): + raise RuntimeError("Pushplus 未配置 Token") + body = {"token": conf["token"], "title": subject or f"{APP_NAME} 通知", "content": text, "template": "txt"} + for k in ("topic", "channel"): + if conf.get(k): + body[k] = conf[k] + with httpx.Client(timeout=15) as c: + resp = c.post("https://www.pushplus.plus/send", json=body) + resp.raise_for_status() + data = resp.json() + if data.get("code") != 200: + raise RuntimeError(f"Pushplus 发送失败：{data.get('msg')}") + + +def _send_serverchan(conf: dict, subject: str, text: str) -> None: + key = (conf.get("sendkey") or "").strip() + if not key: + raise RuntimeError("Server酱未配置 SendKey") + m = re.match(r"^sctp(\d+)t", key) # Server酱³ 的 SendKey 里带 uid + url = f"https://{m.group(1)}.push.ft07.com/send/{key}.send" if m else f"https://sctapi.ftqq.com/{key}.send" + with httpx.Client(timeout=15) as c: + r = c.post(url, data={"title": subject or APP_NAME, "desp": text}) + r.raise_for_status() + if r.json().get("code", 0) != 0: + raise RuntimeError(f"Server酱发送失败：{r.text[:200]}") + + +# ---- 企业微信（群机器人 / 自建应用） ---- +WECOM_DEFAULT_BASE = "https://qyapi.weixin.qq.com" +# 自建应用的 access_token 缓存：key -> (token, 过期时间戳)。gettoken 有频率限制，必须复用（有效期 7200s） +_WECOM_TOKENS: dict[str, tuple[str, float]] = {} +_WECOM_TOKEN_ERRS = {40014, 42001, 41001} # 不合法 / 已过期 / 缺少 access_token：刷新后重试一次 + + +def _trunc_bytes(text: str, limit: int) -> str: + """按 UTF-8 字节数截断（text 上限 2048B、markdown 4096B、textcard 描述 512B）""" + raw = (text or "").encode("utf-8") + if len(raw) <= limit: + return text or "" + return raw[: limit - 3].decode("utf-8", "ignore") + "..." + + +def wecom_api_base(conf: dict) -> str: + """API 代理地址；留空用官方域名。容忍把 /cgi-bin 一起粘进来""" + base = (conf.get("proxy_base") or "").strip().rstrip("/") + if not base: + return WECOM_DEFAULT_BASE + if not base.startswith(("http://", "https://")): + base = "https://" + base + return base[: -len("/cgi-bin")] if base.endswith("/cgi-bin") else base + + +def wecom_token(conf: dict, force: bool = False) -> str: + corp_id, secret = (conf.get("corp_id") or "").strip(), (conf.get("secret") or "").strip() + if not corp_id or not secret: + raise RuntimeError("企业微信自建应用未配置企业ID / 应用Secret") + base = wecom_api_base(conf) + key = f"{base}|{corp_id}|{hashlib.sha256(secret.encode()).hexdigest()[:16]}" + hit = _WECOM_TOKENS.get(key) + if hit and not force and hit[1] > time.time(): + return hit[0] + with httpx.Client(timeout=15) as c: + r = c.get(f"{base}/cgi-bin/gettoken", params={"corpid": corp_id, "corpsecret": secret}) + r.raise_for_status() + data = r.json() + token = data.get("access_token") + if data.get("errcode", 0) != 0 or not token: + raise RuntimeError(f"企业微信获取 access_token 失败：{data.get('errcode')} {data.get('errmsg')}" + "（请检查企业ID / 应用Secret，以及是否已在「企业可信IP」放行服务器出口 IP）") + _WECOM_TOKENS[key] = (token, time.time() + max(60, int(data.get("expires_in") or 7200) - 200)) + return token + + +def wecom_agent_info(conf: dict) -> dict: + """「检查应用」：读取自建应用名与可见范围""" + agent_id = str(conf.get("agent_id") or "").strip() + if not agent_id: + raise RuntimeError("企业微信自建应用未配置应用ID（AgentId）") + base = wecom_api_base(conf) + for attempt in (0, 1): + token = wecom_token(conf, force=bool(attempt)) + with httpx.Client(timeout=15) as c: + r = c.get(f"{base}/cgi-bin/agent/get", params={"access_token": token, "agentid": agent_id}) + r.raise_for_status() + data = r.json() + code = data.get("errcode", 0) + if code in _WECOM_TOKEN_ERRS and attempt == 0: + continue + if code != 0: + raise RuntimeError(f"企业微信读取应用失败：{code} {data.get('errmsg')}") + allow = data.get("allow_userinfos") or {} + return {"agentid": data.get("agentid"), "name": data.get("name"), + "users": len(allow.get("user") or []), + "parties": len((data.get("allow_partys") or {}).get("partyid") or []), + "tags": len((data.get("allow_tags") or {}).get("tagid") or [])} + raise RuntimeError("企业微信读取应用失败：access_token 反复失效") + + +def _wecom_payload(conf: dict, subject: str, text: str) -> dict: + msg_type = (conf.get("msg_type") or "text").strip() + if msg_type == "markdown": + return {"msgtype": "markdown", "markdown": {"content": _trunc_bytes(_body(subject, text), 4096)}} + if msg_type == "textcard": + return {"msgtype": "textcard", "textcard": { + "title": _trunc_bytes(subject or APP_NAME, 128), "description": _trunc_bytes(text, 512), + "url": conf.get("card_url") or "https://github.com/suyijun8182/ros-dpi", "btntxt": "查看详情"}} + return {"msgtype": "text", "text": {"content": _trunc_bytes(_body(subject, text), 2048)}} + + +def _send_wecom_robot(conf: dict, subject: str, text: str) -> None: + url = (conf.get("url") or "").strip() + if not url: + raise RuntimeError("企业微信未配置群机器人 Webhook") + payload = _wecom_payload(conf, subject, text) + if payload["msgtype"] == "textcard": # 群机器人不支持 textcard，降级为 text + payload = {"msgtype": "text", "text": {"content": _trunc_bytes(_body(subject, text), 2048)}} + with httpx.Client(timeout=15) as c: + r = c.post(url, json=payload) + r.raise_for_status() + data = r.json() + if data.get("errcode", 0) != 0: + raise RuntimeError(f"企业微信群机器人发送失败：{data.get('errcode')} {data.get('errmsg')}") + + +def _send_wecom_app(conf: dict, subject: str, text: str) -> None: + agent_id = str(conf.get("agent_id") or "").strip() + if not agent_id: + raise RuntimeError("企业微信自建应用未配置应用ID（AgentId）") + to_user, to_party, to_tag = ((conf.get(k) or "").strip() for k in ("to_user", "to_party", "to_tag")) + if not (to_user or to_party or to_tag): + to_user = "@all" # 三者全空则发给应用可见范围内全部成员 + payload = _wecom_payload(conf, subject, text) + payload.update({"touser": to_user, "toparty": to_party, "totag": to_tag, + "agentid": int(agent_id) if agent_id.isdigit() else agent_id, + "safe": 0, "enable_duplicate_check": 0}) + base = wecom_api_base(conf) + for attempt in (0, 1): + token = wecom_token(conf, force=bool(attempt)) + with httpx.Client(timeout=15) as c: + r = c.post(f"{base}/cgi-bin/message/send", params={"access_token": token}, json=payload) + r.raise_for_status() + data = r.json() + code = data.get("errcode", 0) + if code in _WECOM_TOKEN_ERRS and attempt == 0: + continue + if code != 0: + hint = {81013: "（接收人不在应用可见范围内，请到「应用管理→可见范围」添加成员）", + 60020: "（服务器 IP 未在「企业可信IP」白名单中，或需要配置 API 代理）", + 301014: "（服务器 IP 未在「企业可信IP」白名单中，或需要配置 API 代理）", + 40056: "（应用ID AgentId 不正确）"}.get(code, "") + raise RuntimeError(f"企业微信发送失败：{code} {data.get('errmsg')}{hint}") + # errcode=0 但接收人被剔除时，消息实际没人收到 + invalid = [v for v in (str(data.get(k) or "").strip("| ") for k in ("invaliduser", "invalidparty", "invalidtag")) if v] + if invalid: + raise RuntimeError(f"企业微信部分接收人无效：{' / '.join(invalid)}（请核对成员 UserID，并确认其在应用可见范围内）") return - if isinstance(j, dict): - code = j.get("errcode", j.get("code", j.get("StatusCode"))) - if code not in ok_codes: - raise RuntimeError(str(j.get("errmsg") or j.get("msg") or j.get("message") or j)[:160]) -def _sign_dingtalk(url: str, secret: str) -> str: - ts = str(int(time.time() * 1000)) - sig = base64.b64encode(hmac.new(secret.encode(), f"{ts}\n{secret}".encode(), hashlib.sha256).digest()).decode() - return f"{url}{'&' if '?' in url else '?'}timestamp={ts}&sign={urllib.parse.quote_plus(sig)}" +def _send_wecom(conf: dict, subject: str, text: str) -> None: + (_send_wecom_app if conf.get("mode") == "app" else _send_wecom_robot)(conf, subject, text) + + +def _send_dingtalk(conf: dict, subject: str, text: str) -> None: + url = conf.get("url") + if not url: + raise RuntimeError("钉钉未配置机器人 Webhook") + secret = conf.get("secret") + if secret: + ts = str(round(time.time() * 1000)) + sign = urllib.parse.quote_plus(base64.b64encode( + hmac.new(secret.encode(), f"{ts}\n{secret}".encode(), hashlib.sha256).digest())) + url = f"{url}{'&' if '?' in url else '?'}timestamp={ts}&sign={sign}" + with httpx.Client(timeout=15) as c: + r = c.post(url, json={"msgtype": "text", "text": {"content": _body(subject, text)}}) + r.raise_for_status() + if r.json().get("errcode", 0) != 0: + raise RuntimeError(f"钉钉发送失败：{r.text[:200]}") + + +def _send_discord(conf: dict, subject: str, text: str) -> None: + if not conf.get("url"): + raise RuntimeError("Discord 未配置 Webhook URL") + content = f"**{subject}**\n{text}" if subject else text + with httpx.Client(timeout=15) as c: + c.post(conf["url"], json={"content": content[:1900]}).raise_for_status() + + +def _send_slack(conf: dict, subject: str, text: str) -> None: + if not conf.get("url"): + raise RuntimeError("Slack 未配置 Webhook URL") + with httpx.Client(timeout=15) as c: + c.post(conf["url"], json={"text": f"*{subject}*\n{text}" if subject else text}).raise_for_status() + + +def _send_ntfy(conf: dict, subject: str, text: str) -> None: + server = (conf.get("server") or "https://ntfy.sh").rstrip("/") + if not conf.get("topic"): + raise RuntimeError("ntfy 未配置 Topic") + headers = {} + if conf.get("token"): + headers["Authorization"] = f"Bearer {conf['token']}" + with httpx.Client(timeout=15) as c: + c.post(f"{server}/{conf['topic']}", content=_body(subject, text).encode("utf-8"), headers=headers).raise_for_status() + + +def _send_gotify(conf: dict, subject: str, text: str) -> None: + server = (conf.get("server") or "").rstrip("/") + if not server or not conf.get("token"): + raise RuntimeError("Gotify 未配置 Server / Token") + with httpx.Client(timeout=15) as c: + c.post(f"{server}/message", params={"token": conf["token"]}, + json={"title": subject or APP_NAME, "message": text, + "priority": int(conf.get("priority") or 5)}).raise_for_status() + + +def _render_template(tpl: str, ctx: dict) -> str: + return re.sub(r"\{\{\s*(\w+)\s*\}\}", lambda m: str(ctx.get(m.group(1).strip(), m.group(0))), tpl) + + +def _send_webhook(conf: dict, subject: str, text: str, event: str = "audit") -> None: + """POST JSON {text, subject, event, timestamp}；可选模板、HMAC 签名（X-ROSDPI-Signature）、自定义请求头、超时与重试""" + urls = [u for u in (conf.get("urls") or []) if u] + if not urls: + raise RuntimeError("Webhook 未配置任何目标 URL") + ts = int(time.time()) + ctx = {"text": text, "subject": subject, "event": event, "timestamp": ts} + rendered = _render_template(conf["template"], ctx) if conf.get("template") else text + raw = json.dumps({"text": rendered, "subject": subject, "event": event, "timestamp": ts, "source": "ros-dpi"}, + ensure_ascii=False).encode("utf-8") + headers = {"Content-Type": "application/json"} + for h in conf.get("headers") or []: + k, v = (h.get("key"), h.get("value")) if isinstance(h, dict) else (None, None) + if k and k.lower() not in ("content-type", "x-rosdpi-signature"): + headers[k] = v or "" + if conf.get("secret"): + headers["X-ROSDPI-Signature"] = hmac.new(conf["secret"].encode(), raw, hashlib.sha256).hexdigest() + timeout = max(1, int(conf.get("timeout_ms") or 5000)) / 1000 + retries = max(0, min(10, int(conf.get("max_retries") or 0))) + for url in urls: + last_err = None + for _ in range(retries + 1): + try: + with httpx.Client(timeout=timeout) as c: + c.post(url, content=raw, headers=headers).raise_for_status() + last_err = None + break + except Exception as e: # noqa: BLE001 + last_err = e + if last_err: + raise RuntimeError(f"Webhook 发送失败（{url}）：{last_err}") + + +_SENDERS = {"telegram": _send_telegram, "feishu": _send_feishu, "qq": _send_qq, "bark": _send_bark, + "email": _send_email, "pushplus": _send_pushplus, "serverchan": _send_serverchan, "wecom": _send_wecom, + "dingtalk": _send_dingtalk, "discord": _send_discord, "slack": _send_slack, "ntfy": _send_ntfy, + "gotify": _send_gotify, "webhook": _send_webhook} def send_one(t: str, c: dict, title: str, text: str): - body = f"{title}\n{text}" - with httpx.Client(timeout=15) as h: - if t == "wecom": - if c.get("msg_type") == "markdown": - _check(h.post(c["url"], json={"msgtype": "markdown", "markdown": {"content": f"**{title}**\n{text}"[:4000]}})) - else: - _check(h.post(c["url"], json={"msgtype": "text", "text": {"content": body[:2000]}})) - elif t == "dingtalk": - url = _sign_dingtalk(c["url"], c["secret"]) if c.get("secret") else c["url"] - _check(h.post(url, json={"msgtype": "text", "text": {"content": body}})) - elif t == "feishu": - p = {"msg_type": "text", "content": {"text": body}} - if c.get("secret"): - ts = str(int(time.time())) - key = f"{ts}\n{c['secret']}".encode() - p.update(timestamp=ts, sign=base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode()) - _check(h.post(c["url"], json=p)) - elif t == "bark": - for u in [x for x in c.get("urls") or [] if x]: - _check(h.post(u.rstrip("/"), json={"title": title, "body": text, "group": c.get("group") or "ROS-DPI", - "level": c.get("level") or "active"})) - elif t == "telegram": - base = (c.get("api_base") or "https://api.telegram.org").rstrip("/") - r = h.post(f"{base}/bot{c['bot_token']}/sendMessage", json={"chat_id": c["chat_id"], "text": body}) - r.raise_for_status() - if not r.json().get("ok"): - raise RuntimeError(r.json().get("description", "Telegram 返回失败")) - elif t == "email": - _send_mail(c, title, text) - elif t == "pushplus": - p = {"token": c["token"], "title": title, "content": text, "template": "txt"} - if c.get("topic"): - p["topic"] = c["topic"] - _check(h.post("https://www.pushplus.plus/send", json=p), (200,)) - elif t == "serverchan": - key = c["sendkey"].strip() - m = re.match(r"^sctp(\d+)t", key) # Server酱³ 的 SendKey 里带 uid - url = f"https://{m.group(1)}.push.ft07.com/send/{key}.send" if m else f"https://sctapi.ftqq.com/{key}.send" - _check(h.post(url, data={"title": title, "desp": text})) - elif t == "ntfy": - hd = {"Title": title.encode("utf-8").decode("latin-1", "ignore"), "Tags": "shield"} - if c.get("token"): - hd["Authorization"] = f"Bearer {c['token']}" - h.post(f"{(c.get('server') or 'https://ntfy.sh').rstrip('/')}/{c['topic']}", content=body.encode(), - headers=hd).raise_for_status() - elif t == "gotify": - h.post(f"{c['server'].rstrip('/')}/message", params={"token": c["token"]}, - json={"title": title, "message": text, "priority": int(c.get("priority") or 5)}).raise_for_status() - elif t == "discord": - h.post(c["url"], json={"content": body[:1900]}).raise_for_status() - elif t == "slack": - h.post(c["url"], json={"text": f"*{title}*\n{text}"}).raise_for_status() - elif t == "webhook": - _send_webhook(h, c, title, text) - - -def _send_mail(c: dict, title: str, text: str): - import smtplib - from email.header import Header - from email.mime.text import MIMEText - from email.utils import formataddr - sec = c.get("security") or "ssl" - port = int(c.get("port") or (465 if sec == "ssl" else 587 if sec == "starttls" else 25)) - sender = (c.get("from_addr") or c.get("username") or "").strip() - to = [x.strip() for x in (c.get("to") or "").replace("；", ";").replace(",", ";").replace("，", ";").split(";") if x.strip()] - msg = MIMEText(text, "plain", "utf-8") - msg["Subject"] = Header(title, "utf-8") - msg["From"] = formataddr(("ROS-DPI", sender)) - msg["To"] = ", ".join(to) - cls = smtplib.SMTP_SSL if sec == "ssl" else smtplib.SMTP - with cls(c["host"], port, timeout=15) as s: - if sec == "starttls": - s.starttls() - if c.get("username") and c.get("password"): - s.login(c["username"], c["password"]) - s.sendmail(sender, to, msg.as_string()) - - -def _send_webhook(h: httpx.Client, c: dict, title: str, text: str): - """通用 Webhook：POST JSON {subject, text, event, timestamp}；可选模板替换 text、HMAC 签名、自定义请求头、重试""" - ts = int(time.time()) - msg = text - if c.get("template"): - msg = c["template"].replace("{{text}}", text).replace("{{subject}}", title).replace("{{timestamp}}", str(ts)) - raw = json.dumps({"subject": title, "text": msg, "event": "audit", "timestamp": ts, "source": "ros-dpi"}, - ensure_ascii=False).encode() - hd = {"Content-Type": "application/json"} - for x in c.get("headers") or []: - k = (x.get("key") or "").strip() - if k and k.lower() not in ("content-type", "x-rosdpi-signature"): - hd[k] = x.get("value") or "" - if c.get("secret"): - hd["X-ROSDPI-Signature"] = hmac.new(c["secret"].encode(), raw, hashlib.sha256).hexdigest() - for u in [x for x in c.get("urls") or [] if x]: - err = None - for _ in range(int(c.get("retries") or 0) + 1): - try: - h.post(u, content=raw, headers=hd).raise_for_status() - err = None - break - except Exception as e: # noqa: BLE001 - err = e - time.sleep(1) - if err: - raise err + fn = _SENDERS.get(t) + if not fn: + raise RuntimeError(f"未知渠道：{t}") + fn(c, title, text) def send(title: str, text: str, types: list[str] | None = None, wait: bool = False, config: dict | None = None) -> list[dict]: diff --git a/backend/app/routers/audit.py b/backend/app/routers/audit.py index e69cc5e..4fd35e4 100644 --- a/backend/app/routers/audit.py +++ b/backend/app/routers/audit.py @@ -266,7 +266,9 @@ def _clean(t: str, c: dict, old: dict) -> dict: out = {"enabled": bool(c.get("enabled"))} for k, dv in base.items(): v = c.get(k, dv) - if isinstance(dv, list): + if isinstance(dv, bool): + v = bool(v) + elif isinstance(dv, list): v = [x for x in (v or []) if (x.get("key") if isinstance(x, dict) else str(x).strip())] if k != "headers": v = [str(x).strip() for x in v] @@ -278,7 +280,9 @@ def _clean(t: str, c: dict, old: dict) -> dict: else: v = str(v or "").strip() out[k] = v - for k in ("url", "server", "api_base"): + if t == "wecom" and out["mode"] not in ("webhook", "app"): + out["mode"] = "webhook" + for k in ("url", "server", "card_url", "icon"): if out.get(k) and not str(out[k]).startswith(("http://", "https://")): raise HTTPException(400, f"{notify_name(t)}：地址要以 http:// 或 https:// 开头") for u in out.get("urls") or []: @@ -290,9 +294,9 @@ def _clean(t: str, c: dict, old: dict) -> dict: def notify_name(t: str) -> str: - return {"wecom": "企业微信", "dingtalk": "钉钉", "feishu": "飞书", "bark": "Bark", "telegram": "Telegram", "email": "邮件", - "pushplus": "PushPlus", "serverchan": "Server酱", "ntfy": "ntfy", "gotify": "Gotify", "discord": "Discord", - "slack": "Slack", "webhook": "Webhook"}.get(t, t) + return {"wecom": "企业微信", "dingtalk": "钉钉", "feishu": "飞书", "qq": "QQ", "bark": "Bark", "telegram": "Telegram", + "email": "Email", "pushplus": "Pushplus", "serverchan": "Server酱", "ntfy": "ntfy", "gotify": "Gotify", + "discord": "Discord", "slack": "Slack", "webhook": "Webhook"}.get(t, t) @router.get("/notify") @@ -332,6 +336,36 @@ def test_notify(req: NotifyTest): return {"ok": True} +def _draft(channel: str, config: dict | None) -> dict: + """前端还没保存的配置：只保留认识的字段，密钥以“…”结尾时沿用已保存的值（不校验必填）""" + old = notify.load_config()[channel] + return _clean(channel, {**(config or old), "enabled": False}, old) + + +@router.post("/notify/telegram/{action}") +def telegram_helper(action: str, req: NotifyTest): + """同 EasySub：「验证机器人」= getMe，「获取 Chat ID」= getUpdates（用户先给机器人发一条消息）""" + method = {"me": "getMe", "updates": "getUpdates"}.get(action) + if not method: + raise HTTPException(404, "未知操作") + try: + return notify.telegram_call(_draft("telegram", req.config), method) + except Exception as e: # noqa: BLE001 + raise HTTPException(502, f"Telegram {method} 失败：{e}") + + +@router.post("/notify/wecom/check") +def wecom_check(req: NotifyTest): + """校验企业微信自建应用：拉取 access_token 并读取应用名与可见范围""" + conf = _draft("wecom", req.config) + if conf.get("mode") != "app": + raise HTTPException(400, "该校验仅适用于「自建应用」模式") + try: + return {"ok": True, "agent": notify.wecom_agent_info(conf)} + except Exception as e: # noqa: BLE001 + raise HTTPException(502, str(e)) + + # ---------------- 报表 ---------------- class ReportSettings(BaseModel): diff --git a/backend/app/routers/people.py b/backend/app/routers/people.py index 6a951b0..e62abcc 100644 --- a/backend/app/routers/people.py +++ b/backend/app/routers/people.py @@ -221,7 +221,9 @@ def delete_dept(did: int, request: Request, user: str = Depends(current_user), d raise HTTPException(404, "部门不存在") n = db.scalar(select(func.count()).select_from(Person).where(Person.dept_id == did)) or 0 if n: - raise HTTPException(400, f"“{d.name}”下还有 {n} 个成员，先把他们移到别的分组或删除") + home = insight.get_config(db)["scenario"] == "home" + who, where = ("成员", "分组") if home else ("员工", "部门") + raise HTTPException(400, f"“{d.name}”下还有 {n} 个{who}，先把他们移到别的{where}或删除") name = d.name db.delete(d) oplog.log(db, request, user, "dept_delete", {"id": did, "name": name}) diff --git a/backend/tests/test_notify.py b/backend/tests/test_notify.py index 4946200..aa9a450 100644 --- a/backend/tests/test_notify.py +++ b/backend/tests/test_notify.py @@ -39,7 +39,7 @@ def test_payloads_and_signatures(monkeypatch): cfg = { "dingtalk": {"url": "https://oapi.dingtalk.com/robot/send?access_token=x", "secret": "SECabc"}, "webhook": {"urls": ["https://example.com/hook"], "secret": "s3", "headers": [{"key": "X-A", "value": "1"}], - "template": "[{{subject}}] {{text}}", "retries": 0}, + "template": "[{{subject}}] {{text}}", "max_retries": 0}, "bark": {"urls": ["https://api.day.app/KEY/"], "group": "ROS-DPI", "level": "timeSensitive"}, "serverchan": {"sendkey": "sctp123tABC"}, } @@ -48,7 +48,7 @@ def test_payloads_and_signatures(monkeypatch): by = {c["url"].split("?")[0]: c for c in Recorder.calls} d = next(c for c in Recorder.calls if "dingtalk" in c["url"]) q = urllib.parse.parse_qs(urllib.parse.urlparse(d["url"]).query) - assert "timestamp" in q and "sign" in q and d["json"]["text"]["content"] == "标题\n正文" + assert "timestamp" in q and "sign" in q and d["json"]["text"]["content"] == "标题\n\n正文" w = by["https://example.com/hook"] body = json.loads(w["content"]) assert body["text"] == "[标题] 正文" and w["headers"]["X-A"] == "1" @@ -109,3 +109,93 @@ def test_rule_channels_and_quiet_hours(monkeypatch): assert "免打扰" in mine[0]["json"]["text"] c.put("/api/audit/notify", json={"config": {"wecom": {"enabled": False}, "slack": {"enabled": False}}, "quiet_from": "", "quiet_to": ""}, headers=h) + + +class Recorder2(Recorder): + """在 Recorder 的基础上支持 GET，并按地址返回各家 API 的成功应答""" + def post(self, url, json=None, content=None, headers=None, data=None, params=None): + Recorder.calls.append({"url": str(url), "json": json, "content": content, "headers": headers or {}, "data": data, "params": params}) + body = {"errcode": 0, "code": 0, "ok": True} + if "tenant_access_token" in str(url): + body = {"code": 0, "tenant_access_token": "FT"} + elif "getAppAccessToken" in str(url): + body = {"access_token": "QT"} + elif "pushplus" in str(url): + body = {"code": 200} + return httpx.Response(200, json=body, request=httpx.Request("POST", str(url))) + + def get(self, url, params=None): + Recorder.calls.append({"url": str(url), "params": params}) + body = {"errcode": 0, "access_token": "WT", "expires_in": 7200} if "gettoken" in str(url) else \ + {"errcode": 0, "agentid": 1000002, "name": "ROS-DPI", "allow_userinfos": {"user": [{"userid": "a"}]}} + return httpx.Response(200, json=body, request=httpx.Request("GET", str(url))) + + +def test_easysub_channels(monkeypatch): + """照搬 EasySub 的渠道：飞书应用机器人、QQ 机器人、企业微信自建应用、Pushplus 渠道、Bark 图标""" + monkeypatch.setattr(notify.httpx, "Client", Recorder2) + notify._WECOM_TOKENS.clear() + Recorder.calls = [] + cfg = { + "feishu": {"app_id": "cli_1", "app_secret": "S", "chat_ids": "oc_a，oc_b"}, + "qq": {"app_id": "1", "app_secret": "S", "group_ids": "G1", "user_ids": ""}, + "wecom": {"mode": "app", "corp_id": "ww1", "agent_id": "1000002", "secret": "S", "to_user": "", "msg_type": "textcard"}, + "pushplus": {"token": "P", "channel": "mail"}, + "bark": {"urls": ["https://api.day.app/K"], "icon": "https://x/i.png"}, + } + res = notify.send("标题", "正文", wait=True, config={t: {**notify.DEFAULTS[t], **c} for t, c in cfg.items()}) + assert [r for r in res if not r["ok"]] == [] and len(res) == 5 + urls = [c["url"] for c in Recorder.calls] + assert sum("im/v1/messages" in u for u in urls) == 2 # 两个飞书群 + assert any(u.endswith("/v2/groups/G1/messages") for u in urls) + send = next(c for c in Recorder.calls if c["url"].endswith("/cgi-bin/message/send")) + assert send["params"] == {"access_token": "WT"} and send["json"]["touser"] == "@all" + assert send["json"]["msgtype"] == "textcard" and send["json"]["agentid"] == 1000002 + assert next(c for c in Recorder.calls if "pushplus" in c["url"])["json"]["channel"] == "mail" + assert next(c for c in Recorder.calls if "day.app" in c["url"])["json"]["icon"] == "https://x/i.png" + # token 缓存：再发一次不会再调 gettoken + n = sum("gettoken" in c["url"] for c in Recorder.calls) + notify.send("x", "y", wait=True, config={"wecom": {**notify.DEFAULTS["wecom"], **cfg["wecom"]}}) + assert sum("gettoken" in c["url"] for c in Recorder.calls) == n + assert notify.wecom_agent_info({**notify.DEFAULTS["wecom"], **cfg["wecom"]})["users"] == 1 + # 必填项随模式变化 + assert notify.missing("wecom", {**notify.DEFAULTS["wecom"], "mode": "app"}) == ["corp_id", "agent_id", "secret"] + assert notify.missing("wecom", dict(notify.DEFAULTS["wecom"])) == ["url"] + assert notify.missing("qq", {"app_id": "1", "app_secret": "s"}) == ["group_ids / user_ids"] + + +def test_old_field_names_migrate(): + """0.6.x 存的 email.security / from_addr、webhook.retries 读出来改成 EasySub 的字段""" + with database.session() as db: + row = db.get(Setting, "notify_config") or Setting(key="notify_config") + old = row.value + row.value = json.dumps({"email": {"enabled": True, "host": "smtp.x", "security": "starttls", "from_addr": "a@x", "to": "b@x"}, + "webhook": {"urls": ["https://h"], "retries": 1}}) + db.merge(row) + db.commit() + try: + cfg = notify.load_config() + assert cfg["email"]["ssl"] is False and cfg["email"]["from"] == "a@x" and "security" not in cfg["email"] + assert cfg["webhook"]["max_retries"] == 1 and "retries" not in cfg["webhook"] + assert list(cfg) == list(notify.DEFAULTS)[:] and list(cfg)[0] == "telegram" + finally: + with database.session() as db: + row = db.get(Setting, "notify_config") + if old is None: + db.delete(row) + else: + row.value = old + db.commit() + + +def test_notify_api_bool_and_helpers(monkeypatch): + monkeypatch.setattr(notify.httpx, "Client", Recorder2) + notify._WECOM_TOKENS.clear() + with TestClient(app) as c: + h = _h(c) + r = c.put("/api/audit/notify", json={"config": {"email": {"enabled": False, "host": "smtp.x", "ssl": False, "to": "a@x"}}}, headers=h) + assert r.status_code == 200 and r.json()["config"]["email"]["ssl"] is False + r = c.post("/api/audit/notify/wecom/check", json={"channel": "wecom", "config": {"mode": "app", "corp_id": "ww", "agent_id": "1000002", "secret": "S"}}, headers=h) + assert r.status_code == 200 and r.json()["agent"]["name"] == "ROS-DPI" + assert c.post("/api/audit/notify/wecom/check", json={"channel": "wecom", "config": {"mode": "webhook"}}, headers=h).status_code == 400 + assert c.post("/api/audit/notify/telegram/nope", json={"channel": "telegram"}, headers=h).status_code == 404 diff --git a/backend/tests/test_sources.py b/backend/tests/test_sources.py index bc2a0b1..f5fa6e7 100644 --- a/backend/tests/test_sources.py +++ b/backend/tests/test_sources.py @@ -70,3 +70,39 @@ def test_sources_rows_for_library_panel(loaded_sources): assert rows["github"]["ranges"] == 2 assert rows["cn"]["ranges"] == 1 assert all(r["license"] and r["url"] for r in rows.values()) + + +def test_domain_fallback_before_ip_fallback(loaded_sources): + """有域名时先按域名兜底：国内网站托管在 Google/国内段的 IP 上，仍归「国内网站」而不是 IP 所属的 CDN。""" + c = Classifier() + assert c.classify("192.0.2.1", 6, 443, "www.example-test.cn", None) == "cn_web" + assert c.classify("198.51.100.7", 6, 443, "www.example-test.cn", None) == "cn_web" + # 域名不在任何库里时才按 IP + assert c.classify("192.0.2.1", 6, 443, "not-in-list.example.org", None) == "google" + + +def test_overlapping_ranges_longest_prefix(): + """IP 段重叠时按最长前缀：小段赢；小段之外仍落在大段里（旧实现在这里会漏查返回 None）。""" + from app.classify.sources import _Ranges + r = _Ranges() + r.add("10.0.0.0/8", "cn_net") # 先加入的大段 + r.add("10.1.0.0/16", "aws") # 后加入、更具体 + r.add("10.1.2.0/24", "github") # 再嵌套一层 + r.add("10.200.0.0/16", "cn_net") # 同归属的嵌套，合并后不影响 + r.add("2001:db8::/32", "google") + r.add("2001:db8:1::/48", "cloudflare") + r.build() + assert r.lookup("10.1.2.3") == "github" + assert r.lookup("10.1.3.1") == "aws" + assert r.lookup("10.2.0.1") == "cn_net" # 在小段后面、仍属于大段 + assert r.lookup("10.255.255.255") == "cn_net" + assert r.lookup("10.0.0.0") == "cn_net" + assert r.lookup("11.0.0.1") is None + assert r.lookup("2001:db8:1::5") == "cloudflare" + assert r.lookup("2001:db8:2::5") == "google" + # 一样大时先加入的赢 + r2 = _Ranges() + r2.add("192.0.2.0/24", "google") + r2.add("192.0.2.0/24", "aws") + r2.build() + assert r2.lookup("192.0.2.9") == "google" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 53066d8..be46e4b 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -23,11 +23,13 @@
@@ -151,7 +153,8 @@ function logout() { .mini .nav a.on::before { left: -8px; } .side-foot { margin-top: auto; display: flex; align-items: center; gap: 8px; font: 500 12px/1 var(--cond); color: var(--faint); padding: 0 14px; } .mini .side-foot { justify-content: center; padding: 0; } -.mini .side-foot span:last-child { display: none; } +.mini .side-foot span:last-child, .mini .side-foot .scene { display: none; } +.side-foot .scene { margin-left: auto; text-decoration: none; font: inherit; } .main { margin-left: var(--side); padding: 16px 18px 12px; min-width: 0; } .mini .main { margin-left: 64px; } diff --git a/frontend/src/components/NotifyChannels.vue b/frontend/src/components/NotifyChannels.vue index 4f21887..d4574f0 100644 --- a/frontend/src/components/NotifyChannels.vue +++ b/frontend/src/components/NotifyChannels.vue @@ -1,117 +1,244 @@