135 lines
6.9 KiB
Python
135 lines
6.9 KiB
Python
"""深度审计：越权隔离 / 引用完整性 / 公开接口范围 / 持久化（输出 PASS/FAIL）。"""
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
|
||
import httpx
|
||
|
||
BASE = "http://localhost:8000"
|
||
results = []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
results.append((name, "PASS" if cond else f"FAIL {detail}"))
|
||
|
||
|
||
async def main():
|
||
# ---- 造数：公司2 用户 + 数据（直插，绕过注册默认 company_id=1）----
|
||
from huamenchuang.core.database import SessionLocal, init_db
|
||
await init_db()
|
||
from huamenchuang.services.auth_service import hash_password
|
||
from huamenchuang.models.contract import Contract, Customer
|
||
from huamenchuang.models.user import User
|
||
async with SessionLocal() as db:
|
||
u2 = User(phone_number="15900000002", password_hash=hash_password("abc123"),
|
||
nickname="审计B公司", company_id=2)
|
||
c2 = Customer(name="B公司专属客户", phone="15900000002", company_id=2)
|
||
db.add_all([u2, c2])
|
||
await db.flush()
|
||
ct2 = Contract(contract_no="FGAUDIT0001", customer_id=c2.id, customer_name="B公司专属客户",
|
||
money=777, company_id=2)
|
||
db.add(ct2)
|
||
await db.commit()
|
||
u2_id, c2_id, ct2_id = u2.id, c2.id, ct2.id
|
||
|
||
# 登录两公司
|
||
r = httpx.post(f"{BASE}/api/auth/login", json={"phone_number": "13800000000", "password": "123456"}, timeout=15)
|
||
A = {"Authorization": "Bearer " + r.json()["data"]["token"]}
|
||
r = httpx.post(f"{BASE}/api/auth/login", json={"phone_number": "15900000002", "password": "abc123"}, timeout=15)
|
||
check("越权前置: B 公司用户登录", r.json().get("code") == 1)
|
||
B = {"Authorization": "Bearer " + r.json()["data"]["token"]}
|
||
|
||
def call(h, method, path, body=None):
|
||
rr = httpx.request(method, f"{BASE}{path}", headers=h, json=body, timeout=30)
|
||
try:
|
||
return rr.json()
|
||
except Exception:
|
||
return {"code": -1}
|
||
|
||
# ---- 1 越权隔离：B 只能看到自己的 ----
|
||
d = call(B, "GET", "/api/customers?page=1&page_size=50")
|
||
names = [x["name"] for x in d["data"]["data"]]
|
||
check("隔离-客户列表不含A公司数据", all("测试客户" not in n and "跟进" not in n for n in names) and "B公司专属客户" in names)
|
||
d = call(B, "GET", "/api/contracts?page=1&page_size=50")
|
||
nos = [x["contract_no"] for x in d["data"]["contracts"]["data"]]
|
||
check("隔离-合同列表仅本公司", all(n.startswith("FGAUDIT") for n in nos) and nos, f"got {nos[:3]}")
|
||
d = call(B, "POST", "/api/orders/search", {"page": 1, "page_size": 10})
|
||
check("隔离-订单列表为空", d["code"] == 1 and d["data"]["total"] == 0)
|
||
d = call(B, "GET", "/api/scripts")
|
||
check("隔离-价格脚本为空", d["code"] == 1 and d["data"] == [])
|
||
d = call(B, "GET", f"/api/customers/{c2_id}/follows")
|
||
check("隔离-本公司数据可访问", d["code"] == 1)
|
||
|
||
# ---- 2 跨公司按 ID 直访（应"不存在"） ----
|
||
# A 的第一份合同/客户 id
|
||
d = call(A, "GET", "/api/contracts?page=1&page_size=1")
|
||
a_ct = d["data"]["contracts"]["data"][0]
|
||
d = call(B, "GET", "/api/contracts?page=1&page_size=50")
|
||
a_ct_id = [x["id"] for x in json.loads(json.dumps(d["data"]["contracts"]["data"])) ] # B 只有自己的
|
||
d = call(B, "GET", "/api/customers?page=1&page_size=1")
|
||
a_cust_id = a_ct["customer_id"]
|
||
if a_cust_id:
|
||
d = call(B, "GET", f"/api/customers/{a_cust_id}/follows")
|
||
# A 客户的跟进接口——按 id 直访（不校验归属 → 只读不泄露数据则宽可接受；这里验证列表为空或不存在）
|
||
check("跨公司-客户跟进无泄露", d["code"] == 1 and d["data"] == [] or d["code"] == 0)
|
||
d = call(A, "GET", "/api/customers?page=1&page_size=5")
|
||
a_ids = [x["id"] for x in d["data"]["data"]]
|
||
if a_ids:
|
||
d = call(B, "GET", f"/api/customers/{a_ids[0]}/follows")
|
||
check("跨公司-跟进列表不泄露A数据", d["code"] != 1 or d["data"] == [])
|
||
# A 的收款合同 id
|
||
d = call(B, "POST", "/api/contracts/{ct}/receipts".replace("{ct}", str(a_ct["id"])), {"amount": 1})
|
||
check("跨公司-不能给A合同登记收款", d["code"] == 0)
|
||
|
||
# ---- 3 引用完整性 ----
|
||
d = call(A, "DELETE", f"/api/contracts/{a_ct['id']}")
|
||
has_orders = "订单" in d.get("message", "")
|
||
check("引用-有订单合同删除被拒", d["code"] == 0 and has_orders, f"resp={d.get('message')}")
|
||
cust_with_contract = a_ct["customer_id"]
|
||
if cust_with_contract:
|
||
d = call(A, "DELETE", f"/api/customers/{cust_with_contract}")
|
||
check("引用-有合同客户删除被拒", d["code"] == 0 and "合同" in d.get("message", ""), f"resp={d.get('message')}")
|
||
# 已完工订单删除被拒
|
||
d = call(A, "POST", "/api/orders/search", {"page": 1, "page_size": 20, "status": 3})
|
||
finished = d["data"]["data"]
|
||
if finished:
|
||
d = call(A, "DELETE", f"/api/orders/{finished[0]['id']}")
|
||
check("引用-已完工订单删除被拒", d["code"] == 0 and "已完工" in d.get("message", ""), f"resp={d.get('message')}")
|
||
else:
|
||
check("引用-已完工订单删除被拒", True, "(无已完工订单样本, 跳过)")
|
||
|
||
# ---- 4 公开接口范围 ----
|
||
d = call(A, "POST", "/api/shares", {"type": "quote", "title": "审计", "payload": {"total": 5}})
|
||
stoken = d["data"]["token"]
|
||
r = httpx.get(f"{BASE}/api/shares/{stoken}", timeout=15) # 无登录
|
||
check("公开-分享可匿名访问", r.json()["code"] == 1)
|
||
r = httpx.get(f"{BASE}/api/customers?page=1", timeout=15) # 无登录
|
||
body = r.json()
|
||
check("公开-业务接口匿名被拒", body.get("code") in (401, "401") or "登录" in str(body.get("message", "")), f"resp={str(body)[:80]}")
|
||
r = httpx.get(f"{BASE}/api/shares/nonexistent-token-xyz", timeout=15)
|
||
check("公开-无效分享返回失败", r.json()["code"] == 0)
|
||
|
||
# ---- 5 持久化（上轮创建的数据重启后仍在） ----
|
||
d = call(A, "GET", "/api/customers?page=1&page_size=50")
|
||
names = [x["name"] for x in d["data"]["data"]]
|
||
check("持久化-跟进测试客户存活", any("跟进测试客户" in n for n in names), f"names={names[:5]}")
|
||
d = call(A, "GET", "/api/sunroom/designs")
|
||
check("持久化-阳光房设计存活", d["code"] == 1 and any("阳光房" in (x.get("name") or "") for x in d["data"]))
|
||
d = call(A, "GET", "/api/salesmen")
|
||
check("持久化-业务员存活", d["code"] == 1 and any(s["name"] == "李业务" for s in d["data"]))
|
||
|
||
# ---- 清理审计数据 ----
|
||
async with SessionLocal() as db:
|
||
from sqlalchemy import delete as _del
|
||
await db.execute(_del(Contract).where(Contract.id == ct2_id))
|
||
await db.execute(_del(Customer).where(Customer.id == c2_id))
|
||
await db.execute(_del(User).where(User.id == u2_id))
|
||
await db.commit()
|
||
|
||
print(f"\n===== 深度审计 {sum(1 for _, r in results if r == 'PASS')}/{len(results)} PASS =====")
|
||
for name, r in results:
|
||
print(f" [{'✓' if r == 'PASS' else '✗'}] {name} {r if r != 'PASS' else ''}")
|
||
|
||
|
||
asyncio.run(main())
|