315 lines
14 KiB
Python
315 lines
14 KiB
Python
"""社区共享库（debug16 → 0.6.21 重做）：项目自带社区主站——注册 / 令牌、心跳 + 投票、按来源网络聚合防刷票、
|
|
ETag 304、部署量；普通实例的完整同步流程走真实的主站端点（TestClient），不打桩协议本身。"""
|
|
import gzip
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import database, share
|
|
from app.classify.local import local
|
|
from app.config import get_settings
|
|
from app.engine import engine
|
|
from app.main import app
|
|
from app.models import HostDomainDay, Setting, ShareDeployment, ShareDomain, ShareEntry
|
|
from app.routers import share as share_router
|
|
|
|
DAY = 1792900000
|
|
HOST = "IP:192.168.88.41"
|
|
HUB = "http://testserver"
|
|
|
|
|
|
def _clean(db):
|
|
for m in (ShareEntry, ShareDomain, ShareDeployment):
|
|
db.query(m).delete(synchronize_session=False)
|
|
db.query(HostDomainDay).filter(HostDomainDay.host == HOST).delete(synchronize_session=False)
|
|
for k in ("share_client_instance", "share_client_token", "share_client_hub", "share_hub_instance"):
|
|
row = db.get(Setting, k)
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture()
|
|
def hub(monkeypatch, tmp_path):
|
|
"""本进程同时当社区主站和普通实例：主站端点由 TestClient 提供；X-Test-IP 头模拟不同实例的来源地址"""
|
|
s = get_settings()
|
|
monkeypatch.setattr(s, "share_hub", True)
|
|
monkeypatch.setattr(s, "share_hub_url", HUB)
|
|
monkeypatch.setattr(share, "LIB_REBUILD_GAP", 0)
|
|
monkeypatch.setattr(share, "VOTE_MIN_AGE", 0)
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(share_router, "_source", lambda req: req.headers.get("x-test-ip") or "203.0.113.9")
|
|
database.connect_from_saved(force=True)
|
|
with database.session() as db:
|
|
_clean(db)
|
|
share.reset_library_cache()
|
|
local.reset()
|
|
c = TestClient(app)
|
|
state = {"ip": "198.51.100.1"}
|
|
|
|
class _Client:
|
|
"""client_cycle 用的 httpx.Client 替身：请求转给 TestClient，并带上当前模拟的来源地址"""
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def post(self, url, **kw):
|
|
kw["headers"] = {**(kw.get("headers") or {}), "X-Test-IP": state["ip"]}
|
|
return c.post(url, **kw)
|
|
|
|
def get(self, url, **kw):
|
|
kw["headers"] = {**(kw.get("headers") or {}), "X-Test-IP": state["ip"]}
|
|
return c.get(url, **kw)
|
|
|
|
monkeypatch.setattr(share, "_http", lambda: _Client())
|
|
yield c, state
|
|
with database.session() as db:
|
|
_clean(db)
|
|
share.reset_library_cache()
|
|
local.reset()
|
|
|
|
|
|
def _as_new_instance(state, ip):
|
|
"""换成另一台实例：清掉本机保存的注册信息，换来源地址"""
|
|
with database.session() as db:
|
|
for k in ("share_client_instance", "share_client_token", "share_client_hub"):
|
|
row = db.get(Setting, k)
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.commit()
|
|
share._client.update(votes_ts=0, etag="")
|
|
state["ip"] = ip
|
|
|
|
|
|
def _seed(db, rows):
|
|
for dom, app, b in rows:
|
|
db.add(HostDomainDay(ts=DAY, host=HOST, domain=dom, app=app, base=None, bytes=b))
|
|
db.commit()
|
|
|
|
|
|
def _client_cycle():
|
|
with database.session() as db:
|
|
res = share.client_cycle(engine, db)
|
|
db.commit()
|
|
return res
|
|
|
|
|
|
def test_export_filters_private():
|
|
database.connect_from_saved(force=True)
|
|
with database.session() as db:
|
|
_clean(db)
|
|
_seed(db, [("93.184.216.34", "douyin", 800), ("192.168.8.5", "douyin", 999),
|
|
("239.255.255.250", "douyin", 999), ("v.douyin.com", "douyin", 50)])
|
|
entries, domains = share.export_entries(engine, db)
|
|
_clean(db)
|
|
ips = {e["ip"] for e in entries}
|
|
assert "93.184.216.34" in ips and not ips & {"192.168.8.5", "239.255.255.250"} # 私网 / 组播不出门
|
|
assert "v.douyin.com" in {d["domain"] for d in domains}
|
|
|
|
|
|
def test_client_ip_behind_proxy():
|
|
# 直连公网：用 TCP 对端，伪造的 X-Forwarded-For 无效
|
|
assert share.client_ip("8.8.8.8", "1.1.1.1", None) == "8.8.8.8"
|
|
# 反向代理在内网：取代理追加在最右边的公网地址
|
|
assert share.client_ip("192.168.8.1", "1.1.1.1, 9.9.9.9", None) == "9.9.9.9"
|
|
# 最右一跳不是公网地址：不再往左找（左边的值客户端可以随便写），退回代理地址
|
|
assert share.client_ip("127.0.0.1", "9.9.9.9, 10.0.0.5", None) == "127.0.0.1"
|
|
assert share.client_ip("127.0.0.1", None, "9.9.9.9") == "9.9.9.9"
|
|
|
|
|
|
def test_two_networks_publish_end_to_end(hub):
|
|
c, state = hub
|
|
with database.session() as db:
|
|
_seed(db, [("1.2.3.4", "douyin", 500), ("v.douyin.com", "douyin", 300)])
|
|
|
|
# 第一台：注册 + 上报；只有一个网络见过，社区库为空
|
|
res = _client_cycle()
|
|
assert res["total"] == 1 and res["online"] == 1 and res["community_ips"] == 0
|
|
assert share.sync_state()["ok"] is True and local.ip_app("1.2.3.4") is None
|
|
|
|
# 第二台（另一个网络）投了同样的结论：两网一致 → 进社区库，本机自学习的社区层生效
|
|
_as_new_instance(state, "203.0.113.50")
|
|
res = _client_cycle()
|
|
assert res["total"] == 2 and res["community_ips"] >= 1 and res["community_domains"] >= 1
|
|
assert local.ip_app("1.2.3.4") == "douyin"
|
|
assert local.domain_app("v.douyin.com") == "douyin"
|
|
|
|
# 社区库没变化：带 ETag 再拉是 304，结果沿用；缓存文件已写，重启后可直接加载
|
|
etag = share._client["etag"]
|
|
assert etag and c.get(HUB + share.API + "/library", headers={"If-None-Match": etag}).status_code == 304
|
|
local.reset()
|
|
share.load_cached_library()
|
|
assert local.ip_app("1.2.3.4") == "douyin"
|
|
|
|
# 公开的部署量
|
|
assert c.get(HUB + share.API + "/stats").json() == {"total": 2, "online": 2, "offline": 0}
|
|
|
|
|
|
def test_same_network_counts_once(hub):
|
|
"""一个人在同一网络里伪造多个实例：只算一票，进不了社区库"""
|
|
c, state = hub
|
|
with database.session() as db:
|
|
_seed(db, [("5.6.7.8", "bilibili", 100)])
|
|
_client_cycle()
|
|
_as_new_instance(state, "198.51.100.77") # 同一个 /24
|
|
res = _client_cycle()
|
|
assert res["total"] == 2 and res["community_ips"] == 0 and local.ip_app("5.6.7.8") is None
|
|
|
|
|
|
def test_majority_and_whitelist(hub):
|
|
c, _state = hub
|
|
with database.session() as db:
|
|
rows = []
|
|
for i, ip in enumerate(("1.0.0.1", "2.0.0.1", "3.0.0.1")):
|
|
iid, _t = share.register(db, share.net_of(db, ip), "t")
|
|
db.flush()
|
|
rows.append(db.get(ShareDeployment, iid))
|
|
cls = engine.classifier
|
|
share.report(db, rows[0], rows[0].net, {"entries": [{"ip": "8.8.8.8", "app": "douyin", "bytes": 1}]}, cls)
|
|
share.report(db, rows[1], rows[1].net, {"entries": [{"ip": "8.8.8.8", "app": "douyin", "bytes": 1}]}, cls)
|
|
share.report(db, rows[2], rows[2].net, {"entries": [{"ip": "8.8.8.8", "app": "bilibili", "bytes": 10 ** 9},
|
|
{"ip": "9.9.9.9", "app": "made_up_app", "bytes": 1},
|
|
{"ip": "10.0.0.1", "app": "douyin", "bytes": 1}]}, cls)
|
|
db.commit()
|
|
lib = share.build_library(db)
|
|
n = db.query(ShareEntry).filter(ShareEntry.reporter == rows[2].instance).count()
|
|
assert lib["entries"] == [{"ip": "8.8.8.8", "app": "douyin"}] # 两个网络对一个网络：按网络数，不按字节
|
|
assert n == 1 # 编造的应用、私网地址被拒收
|
|
|
|
|
|
def test_auth_and_limits(hub):
|
|
c, _state = hub
|
|
url = HUB + share.API
|
|
assert c.post(url + "/report", json={"version": "x"}).status_code == 401
|
|
reg = c.post(url + "/register", json={"version": "x"}, headers={"X-Test-IP": "192.0.2.1"}).json()
|
|
bad = {"Authorization": f"Bearer {reg['instance']}:wrong"}
|
|
assert c.post(url + "/report", json={"version": "x"}, headers=bad).status_code == 401
|
|
ok = {"Authorization": f"Bearer {reg['instance']}:{reg['token']}", "Content-Encoding": "gzip"}
|
|
assert c.post(url + "/report", content=gzip.compress(json.dumps({"version": "x"}).encode()), headers=ok).json()["ok"]
|
|
# 解压后超限：拒收（防压缩炸弹）
|
|
bomb = gzip.compress(b" " * (share.MAX_INFLATED + 10))
|
|
assert c.post(url + "/report", content=bomb, headers=ok).status_code == 413
|
|
# 同一网络一天注册超限
|
|
for i in range(share.REGISTER_PER_NET):
|
|
assert c.post(url + "/register", json={}, headers={"X-Test-IP": f"100.20.30.{i}"}).status_code == 200
|
|
assert c.post(url + "/register", json={}, headers={"X-Test-IP": "100.20.30.200"}).status_code == 429
|
|
|
|
|
|
def test_hub_counts_itself_in_process(hub):
|
|
"""主站自己的投票在进程内计入（不经网络）：主站 + 一个普通实例两网一致即发布"""
|
|
c, state = hub
|
|
with database.session() as db:
|
|
_seed(db, [("4.4.4.4", "douyin", 100)])
|
|
res = share.hub_cycle(engine, db)
|
|
db.commit()
|
|
assert res["total"] == 1 and res["community_ips"] == 0
|
|
assert local.ip_app("4.4.4.4") is None
|
|
_client_cycle()
|
|
with database.session() as db:
|
|
share._lib["dirty"] = True
|
|
res = share.hub_cycle(engine, db)
|
|
db.commit()
|
|
assert res["total"] == 2 and res["community_ips"] >= 1 and local.ip_app("4.4.4.4") == "douyin"
|
|
|
|
|
|
def test_non_hub_404(monkeypatch):
|
|
monkeypatch.setattr(get_settings(), "share_hub", False)
|
|
c = TestClient(app)
|
|
for path in ("/stats", "/library"):
|
|
assert c.get(share.API + path).status_code == 404
|
|
assert c.post(share.API + "/register", json={}).status_code == 404
|
|
|
|
|
|
def test_client_error_is_reported(monkeypatch):
|
|
"""主站不可达：设置页能看到原因，社区库沿用缓存"""
|
|
import httpx
|
|
|
|
database.connect_from_saved(force=True)
|
|
monkeypatch.setattr(get_settings(), "share_hub", False)
|
|
monkeypatch.setattr(get_settings(), "share_hub_url", "http://127.0.0.1:9")
|
|
|
|
class _Down:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def post(self, url, **kw):
|
|
raise httpx.ConnectError("connection refused")
|
|
get = post
|
|
|
|
monkeypatch.setattr(share, "_http", lambda: _Down())
|
|
with database.session() as db:
|
|
with pytest.raises(httpx.ConnectError):
|
|
share.cycle(engine, db)
|
|
st = share.sync_state()
|
|
assert st["ok"] is False and "ConnectError" in st["error"]
|
|
|
|
|
|
def test_no_ipv6_egress_message():
|
|
"""实例不需要公网 IP，但主站只有 IPv6：容器没有 IPv6 出口时给出能照着做的提示"""
|
|
import httpx
|
|
msg = share._err(httpx.ConnectError("[Errno 101] Network is unreachable"))
|
|
assert "IPv6" in msg and "Docker" in msg
|
|
|
|
|
|
def test_hub_url_from_settings(monkeypatch):
|
|
"""debug23：社区主站地址可在网页填写（优先于环境变量），空字符串恢复内置默认；地址格式校验"""
|
|
monkeypatch.setattr(get_settings(), "share_hub", False)
|
|
monkeypatch.setattr(get_settings(), "share_hub_url", "https://builtin.example:8443")
|
|
with TestClient(app) as c:
|
|
if not c.get("/api/auth/status").json()["initialized"]:
|
|
c.post("/api/auth/setup", json={"username": "admin", "password": "secret1"})
|
|
h = {"Authorization": "Bearer " + c.post("/api/auth/login", json={"username": "admin", "password": "secret1"}).json()["token"]}
|
|
assert c.put("/api/settings", json={"share_hub_url": "hub.example.com"}, headers=h).status_code == 400
|
|
r = c.put("/api/settings", json={"share_hub_url": "https://hub.example.com/"}, headers=h).json()
|
|
assert r["share_hub_url"] == "https://hub.example.com"
|
|
st = c.get("/api/system/deployments", headers=h).json()
|
|
assert st["hub_url"] == "https://hub.example.com" and st["custom_hub"]
|
|
assert st["default_hub_url"] == "https://builtin.example:8443"
|
|
c.put("/api/settings", json={"share_hub_url": ""}, headers=h)
|
|
st = c.get("/api/system/deployments", headers=h).json()
|
|
assert st["hub_url"] == "https://builtin.example:8443" and not st["custom_hub"]
|
|
|
|
|
|
def test_hub_abuse_limits(hub, monkeypatch):
|
|
"""新注册的实例满 1 天才计票；同一网络最多 2 个投票实例；未认证的上报不读请求体；条目数粗算超限拒收"""
|
|
c, _state = hub
|
|
url = HUB + share.API
|
|
cls = engine.classifier
|
|
with database.session() as db:
|
|
rows = []
|
|
for ip in ("1.0.0.1", "2.0.0.1"):
|
|
iid, _t = share.register(db, share.net_of(db, ip), "t")
|
|
db.flush()
|
|
rows.append(db.get(ShareDeployment, iid))
|
|
for r in rows:
|
|
share.report(db, r, r.net, {"entries": [{"ip": "8.8.8.8", "app": "douyin", "bytes": 1}]}, cls)
|
|
db.commit()
|
|
monkeypatch.setattr(share, "VOTE_MIN_AGE", 86400)
|
|
assert share.build_library(db)["entries"] == [] # 刚注册：不计票
|
|
for r in rows:
|
|
r.first_seen -= 2 * 86400
|
|
db.commit()
|
|
assert share.build_library(db)["entries"] == [{"ip": "8.8.8.8", "app": "douyin"}]
|
|
# 同一网络第 3 个实例的投票不收
|
|
net = rows[0].net
|
|
extra = []
|
|
for _ in range(2):
|
|
iid, _t = share.register(db, net, "t")
|
|
db.flush()
|
|
extra.append(db.get(ShareDeployment, iid))
|
|
assert share.report(db, extra[0], net, {"entries": []}, cls)["accepted"] == 0
|
|
db.commit()
|
|
res = share.report(db, extra[1], net, {"entries": [{"ip": "9.9.9.9", "app": "douyin", "bytes": 1}]}, cls)
|
|
assert res.get("skipped") == "同一网络的投票实例已满"
|
|
# 没有令牌：直接 401，不会去解压一个巨大的请求体
|
|
bomb = gzip.compress(b"{}," * 3_000_000)
|
|
assert c.post(url + "/report", content=bomb, headers={"Content-Encoding": "gzip"}).status_code == 401
|
|
with pytest.raises(share.ShareError):
|
|
share.inflate(b"[" + b"{}," * (2 * share.MAX_UPLOAD + 100) + b"{}]", None)
|