271 lines
11 KiB
Python
271 lines
11 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""s-file 交付验收冒烟测试：登录 -> 上传送检 -> 不通过拦截 -> 跨网摆渡 -> 外网解包再送检 -> 人工复核。
|
|||
|
|
|
|||
|
|
在能同时访问内网机与外网机前端的机器上运行（需要 python3 + requests）。
|
|||
|
|
全部取值走环境变量，脚本内不含任何口令：
|
|||
|
|
|
|||
|
|
SFILE_IW_BASE=http://10.0.0.11/api SFILE_OW_BASE=http://10.0.0.12/api \\
|
|||
|
|
SFILE_ADMIN_USER=admin SFILE_ADMIN_PWD='<现场口令>' python3 smoke_test.py
|
|||
|
|
|
|||
|
|
退出码：0 全通过，1 有失败项，2 参数缺失。
|
|||
|
|
"""
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
import uuid
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
import requests
|
|||
|
|
except ImportError:
|
|||
|
|
print("缺少 requests：pip3 install requests")
|
|||
|
|
sys.exit(2)
|
|||
|
|
|
|||
|
|
IW = os.environ.get("SFILE_IW_BASE", "http://127.0.0.1:28180/api").rstrip("/")
|
|||
|
|
OW = os.environ.get("SFILE_OW_BASE", "http://127.0.0.1:28280/api").rstrip("/")
|
|||
|
|
USER = os.environ.get("SFILE_ADMIN_USER", "admin")
|
|||
|
|
PWD = os.environ.get("SFILE_ADMIN_PWD", "")
|
|||
|
|
# 光闸发出目录（内网机侧本地路径）。跨机验收时留空，只断言接口成功
|
|||
|
|
GATE_IW2OW = os.environ.get("SFILE_GATE_SEND_DIR", "")
|
|||
|
|
|
|||
|
|
# EICAR 标准测试串；后端的 MeshSafe 结构引擎会识别「内嵌 EICAR」，
|
|||
|
|
# 因此允许在尾部追加唯一标记（见 main 里的载荷说明）
|
|||
|
|
EICAR = "X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
|
|||
|
|
STATUS = {"1": "检测中", "2": "检测通过", "3": "检测不通过", "4": "检测失败"}
|
|||
|
|
|
|||
|
|
results = []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def rec(name, ok, detail=""):
|
|||
|
|
results.append((name, bool(ok), detail))
|
|||
|
|
print((" [OK] " if ok else " [FAIL] ") + name + (" " + detail if detail else ""), flush=True)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Side:
|
|||
|
|
def __init__(self, base, label):
|
|||
|
|
self.base = base
|
|||
|
|
self.label = label
|
|||
|
|
self.s = requests.Session()
|
|||
|
|
self.token = None
|
|||
|
|
|
|||
|
|
def _hdr(self, csrf=False):
|
|||
|
|
# 后端 MyLoginSuccessHandler 靠 x-requested-with 决定是否回 JSON，否则 302 跳转
|
|||
|
|
h = {"x-requested-with": "XMLHttpRequest"}
|
|||
|
|
if self.token:
|
|||
|
|
h["Authorization"] = "Bearer " + self.token
|
|||
|
|
if csrf:
|
|||
|
|
t = self.s.cookies.get("XSRF-TOKEN")
|
|||
|
|
if t:
|
|||
|
|
h["X-XSRF-TOKEN"] = t
|
|||
|
|
return h
|
|||
|
|
|
|||
|
|
def login(self, user, pwd):
|
|||
|
|
r = self.s.post(self.base + "/login",
|
|||
|
|
data={"username": user, "password": pwd, "userType": 1},
|
|||
|
|
headers=self._hdr(), timeout=30)
|
|||
|
|
j = self._json(r)
|
|||
|
|
if str(j.get("code")) != "200":
|
|||
|
|
raise RuntimeError("登录失败 HTTP %s：%s" % (r.status_code, json.dumps(j, ensure_ascii=False)[:300]))
|
|||
|
|
self.token = j["data"]["token"]
|
|||
|
|
# 触发 XSRF-TOKEN cookie（POST/PUT 需要带 X-XSRF-TOKEN）
|
|||
|
|
self.s.get(self.base + "/user/info", headers=self._hdr(), timeout=30)
|
|||
|
|
return j
|
|||
|
|
|
|||
|
|
def get(self, path, params=None):
|
|||
|
|
r = self.s.get(self.base + path, params=params, headers=self._hdr(), timeout=60)
|
|||
|
|
return r.status_code, self._json(r)
|
|||
|
|
|
|||
|
|
def post(self, path, payload=None, files=None):
|
|||
|
|
kw = {"files": files} if files is not None else {"json": payload if payload is not None else {}}
|
|||
|
|
r = self.s.post(self.base + path, headers=self._hdr(csrf=True), timeout=180, **kw)
|
|||
|
|
return r.status_code, self._json(r)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _json(r):
|
|||
|
|
try:
|
|||
|
|
return r.json()
|
|||
|
|
except Exception:
|
|||
|
|
return {"_status": r.status_code, "_raw": r.text[:400]}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def ok(j):
|
|||
|
|
return str(j.get("code")) == "200"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def msg(j):
|
|||
|
|
return j.get("msg") or j.get("_raw") or json.dumps(j, ensure_ascii=False)[:200]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def page_find(side, name, root_type, parent_id=0):
|
|||
|
|
"""按文件名在指定列表里找记录。root_type：1 我的文件 / 2 内穿外 / 3 外传内"""
|
|||
|
|
_, j = side.get("/user_file/page",
|
|||
|
|
{"parentId": parent_id, "pageNo": 1, "pageSize": 50, "rootType": root_type})
|
|||
|
|
if not ok(j):
|
|||
|
|
return None
|
|||
|
|
for e in (j.get("data") or {}).get("entityList", []):
|
|||
|
|
if e.get("fileName") == name:
|
|||
|
|
return e
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def wait_scan(side, file_code, timeout=240):
|
|||
|
|
"""轮询安全检测结论；检测中(1)继续等，其余状态返回"""
|
|||
|
|
deadline = time.time() + timeout
|
|||
|
|
detail = None
|
|||
|
|
while time.time() < deadline:
|
|||
|
|
_, j = side.get("/file_security/detail", {"fileCode": file_code})
|
|||
|
|
if ok(j):
|
|||
|
|
detail = j.get("data")
|
|||
|
|
st = str((detail or {}).get("scanStatus") or "")
|
|||
|
|
if st and st != "1":
|
|||
|
|
return detail
|
|||
|
|
time.sleep(3)
|
|||
|
|
return detail
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
if not PWD:
|
|||
|
|
print("必须设置 SFILE_ADMIN_PWD")
|
|||
|
|
return 2
|
|||
|
|
tag = uuid.uuid4().hex[:8]
|
|||
|
|
clean_name = "sfile-smoke-clean-%s.txt" % tag
|
|||
|
|
bad_name = "sfile-smoke-eicar-%s.txt" % tag
|
|||
|
|
|
|||
|
|
iw, ow = Side(IW, "内网"), Side(OW, "外网")
|
|||
|
|
print("\n目标：内网 %s / 外网 %s" % (IW, OW))
|
|||
|
|
|
|||
|
|
print("\n== 1. 登录两侧 ==")
|
|||
|
|
for side in (iw, ow):
|
|||
|
|
try:
|
|||
|
|
side.login(USER, PWD)
|
|||
|
|
rec("%s 登录" % side.label, True, "token %d 字符" % len(side.token))
|
|||
|
|
except Exception as e:
|
|||
|
|
rec("%s 登录" % side.label, False, str(e))
|
|||
|
|
return 1
|
|||
|
|
_, j = iw.get("/file_security/enabled")
|
|||
|
|
ms_on = ok(j) and j.get("data") is True
|
|||
|
|
rec("MeshSafe 开关", True, "enabled=%s（false 时跳过 3/4/7 段）" % ms_on)
|
|||
|
|
|
|||
|
|
print("\n== 2. 上传 + 建档 ==")
|
|||
|
|
# fileCode 按内容寻址：同一份字节会复用上一条检测结论（含人工复核），
|
|||
|
|
# 载荷必须每轮唯一，否则「不通过拦截」这类断言会被上一次的放行结论污染
|
|||
|
|
uploaded = {}
|
|||
|
|
for name, content in ((clean_name, ("s-file smoke clean payload %s\n" % tag) * 80),
|
|||
|
|
(bad_name, EICAR + "\n# smoke-tag %s\n" % tag)):
|
|||
|
|
_, j = iw.post("/file/upload", files={"file": (name, content.encode(), "text/plain")})
|
|||
|
|
if not ok(j):
|
|||
|
|
rec("上传 %s" % name, False, msg(j))
|
|||
|
|
return 1
|
|||
|
|
f = j["data"]
|
|||
|
|
_, j2 = iw.post("/normal_user_file/create",
|
|||
|
|
{"fileCode": f.get("fileCode"), "fileName": name, "fileType": "1", "parentId": 0})
|
|||
|
|
rec("上传并建档 %s" % name, ok(j2), msg(j2) if not ok(j2) else "fileCode=%s" % f.get("fileCode"))
|
|||
|
|
if not ok(j2):
|
|||
|
|
return 1
|
|||
|
|
uploaded[name] = f.get("fileCode")
|
|||
|
|
|
|||
|
|
print("\n== 3. 检测结论 ==")
|
|||
|
|
ids = {}
|
|||
|
|
verdicts = {}
|
|||
|
|
for name, code in uploaded.items():
|
|||
|
|
found = page_find(iw, name, "1")
|
|||
|
|
if found:
|
|||
|
|
ids[name] = found["id"]
|
|||
|
|
d = wait_scan(iw, code)
|
|||
|
|
st = str((d or {}).get("scanStatus") or "")
|
|||
|
|
verdicts[name] = st
|
|||
|
|
print(" %s -> %s | action=%s reason=%s"
|
|||
|
|
% (name, STATUS.get(st, "未知"), (d or {}).get("action"), ((d or {}).get("reason") or "")[:80]))
|
|||
|
|
if ms_on:
|
|||
|
|
rec("干净文件检测通过", verdicts.get(clean_name) == "2", "状态=%s" % STATUS.get(verdicts.get(clean_name)))
|
|||
|
|
rec("EICAR 检测不通过", verdicts.get(bad_name) == "3", "状态=%s" % STATUS.get(verdicts.get(bad_name)))
|
|||
|
|
|
|||
|
|
print("\n== 4. 不通过文件禁止传输 ==")
|
|||
|
|
if len(ids) != len(uploaded):
|
|||
|
|
rec("内网「我的文件」可见上传文件", False, "只找到 %s" % list(ids))
|
|||
|
|
return 1
|
|||
|
|
if ms_on:
|
|||
|
|
_, j = iw.get("/file_security/share_check", {"userFileId": ids[bad_name]})
|
|||
|
|
d = j.get("data") or {}
|
|||
|
|
rec("share_check 命中不通过", ok(j) and d.get("blockedCount", 0) >= 1,
|
|||
|
|
"blocked=%s scanning=%s" % (d.get("blockedCount"), d.get("scanningCount")))
|
|||
|
|
_, j = iw.post("/transfer_user_file/create", {"userFileId": ids[bad_name]})
|
|||
|
|
rec("不通过文件传输被拦截", not ok(j), msg(j)[:160])
|
|||
|
|
|
|||
|
|
print("\n== 5. 干净文件跨网摆渡 ==")
|
|||
|
|
before = set(os.listdir(GATE_IW2OW)) if GATE_IW2OW and os.path.isdir(GATE_IW2OW) else None
|
|||
|
|
_, j = iw.post("/transfer_user_file/create", {"userFileId": ids[clean_name]})
|
|||
|
|
rec("内网发起传输", ok(j), msg(j) if not ok(j) else "外发文件夹ID=%s" % j.get("data"))
|
|||
|
|
if not ok(j):
|
|||
|
|
return 1
|
|||
|
|
if before is None:
|
|||
|
|
print(" 未设 SFILE_GATE_SEND_DIR（跨机验收常见），跳过摆渡包落盘检查")
|
|||
|
|
else:
|
|||
|
|
new_zip = None
|
|||
|
|
for _ in range(30):
|
|||
|
|
delta = set(os.listdir(GATE_IW2OW)) - before
|
|||
|
|
if delta:
|
|||
|
|
new_zip = sorted(delta)[0]
|
|||
|
|
break
|
|||
|
|
time.sleep(2)
|
|||
|
|
rec("光闸发出目录出现摆渡包", new_zip is not None, new_zip or GATE_IW2OW)
|
|||
|
|
|
|||
|
|
print("\n== 6. 外网解包 + 再送检（监听任务默认每分钟一次） ==")
|
|||
|
|
# 接收侧落在「内穿外」列表(root_type=2)，目录名带时间戳后缀，需进目录取子文件
|
|||
|
|
folder = None
|
|||
|
|
deadline = time.time() + 300
|
|||
|
|
while time.time() < deadline and folder is None:
|
|||
|
|
_, j = ow.get("/user_file/page", {"parentId": 0, "pageNo": 1, "pageSize": 50, "rootType": "2"})
|
|||
|
|
for e in (j.get("data") or {}).get("entityList", []):
|
|||
|
|
if (e.get("fileName") or "").startswith(clean_name.rsplit(".", 1)[0]):
|
|||
|
|
folder = e
|
|||
|
|
break
|
|||
|
|
if folder is None:
|
|||
|
|
time.sleep(10)
|
|||
|
|
rec("外网「内穿外」出现接收目录", folder is not None,
|
|||
|
|
str(folder.get("fileName")) if folder else "超时未出现（检查光闸与 FILE_TRANSFER_LOCATION）")
|
|||
|
|
if folder:
|
|||
|
|
child = None
|
|||
|
|
_, j = ow.get("/user_file/page",
|
|||
|
|
{"parentId": folder["id"], "pageNo": 1, "pageSize": 50, "rootType": "2"})
|
|||
|
|
for e in (j.get("data") or {}).get("entityList", []):
|
|||
|
|
if e.get("fileName") == clean_name:
|
|||
|
|
child = e
|
|||
|
|
break
|
|||
|
|
rec("接收目录内可见文件", child is not None, str(child.get("fileCode")) if child else "目录为空")
|
|||
|
|
if child:
|
|||
|
|
d = wait_scan(ow, child["fileCode"], timeout=180)
|
|||
|
|
st = str((d or {}).get("scanStatus") or "")
|
|||
|
|
rec("外网解包后自动送检", (not ms_on) or st == "2",
|
|||
|
|
"状态=%s reason=%s" % (STATUS.get(st, "未知"), ((d or {}).get("reason") or "")[:60]))
|
|||
|
|
|
|||
|
|
print("\n== 7. 人工复核改判 ==")
|
|||
|
|
if not ms_on:
|
|||
|
|
print(" MeshSafe 未启用，跳过")
|
|||
|
|
else:
|
|||
|
|
_, j = iw.post("/file_security/review",
|
|||
|
|
{"fileCode": uploaded[bad_name], "decision": "pass",
|
|||
|
|
"comment": "交付验收复核放行 %s" % tag})
|
|||
|
|
rec("复核改判接口", ok(j), msg(j) if not ok(j) else "")
|
|||
|
|
d = j.get("data") or {}
|
|||
|
|
rec("复核后状态=通过", str(d.get("scanStatus")) == "2",
|
|||
|
|
"状态=%s action=%s" % (STATUS.get(str(d.get("scanStatus"))), d.get("action")))
|
|||
|
|
_, j = iw.get("/file_security/detail", {"fileCode": uploaded[bad_name]})
|
|||
|
|
rec("复核说明落库", ok(j) and tag in ((j.get("data") or {}).get("reviewComment") or ""),
|
|||
|
|
str((j.get("data") or {}).get("reviewComment"))[:80])
|
|||
|
|
_, j = iw.get("/file_security/share_check", {"userFileId": ids[bad_name]})
|
|||
|
|
d = j.get("data") or {}
|
|||
|
|
rec("复核后不再拦截", ok(j) and d.get("blockedCount", 0) == 0, "blocked=%s" % d.get("blockedCount"))
|
|||
|
|
|
|||
|
|
print("\n== 汇总 ==")
|
|||
|
|
passed = sum(1 for _, o, _ in results if o)
|
|||
|
|
for name, o, _ in results:
|
|||
|
|
print(" %s %s" % ("OK " if o else "FAIL", name))
|
|||
|
|
print(" %d/%d 通过" % (passed, len(results)))
|
|||
|
|
return 0 if passed == len(results) else 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|