#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 价格查询接口 · 真 HTTP 端到端验证（注入内存会话 + 临时实例，不碰生产端口） ======================================================================= 为什么必须这么验：价格是敏感信息，静态审读证明不了分支可达、更证明不了隔离生效。 本脚本给不同身份注入内存会话，起一个临时实例，真打 /api/price/query， 逐条断言「谁能看到什么」，并把隔离证据打印出来。 覆盖： ① 普通销售（默认 own）→ 只能看到 销售员==本人 的明细；响应全文不得出现其他客户名 ② admin → 能看到全量明细（records_total == 全库成交次数） ③ 权限面板置「不可见」→ 403 ④ 无 token → 401 ⑤ 权限面板改「全量」→ 立即可见全量（改完还原） ⑥ 客户维度模式（sku 空 + customer 有）→ 只给该客户在我可见范围内的货号 ⑨ 年份筛选（2024/2025/2026）：逐年条数 == 索引里该年条数、year_options 与 meta.by_year 一致、 三年之和 == 全部（无重叠）、own 口径在任一年都不出现其他客户名、非法年份回落到「全部」 用法：/usr/bin/python3 脚本/verify_price_api.py 退出码：0 全通过；1 有断言失败 """ import hashlib import importlib import json import os import sys import threading import time import urllib.error import urllib.parse import urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) sys.path.insert(0, HERE) sys.path.insert(0, ROOT) PORT = 8799 INDEX_FILE = os.path.join(ROOT, '数据', 'price_index.json') PERM_FILE = os.path.join(ROOT, '数据', 'user_permissions.json') SALES = '刘子研' # 普通销售（大客户部） SALES2 = '陈溪子' # 普通销售（拓展部） ADMIN = '刘新元' # admin NONEU = '测试—不可见' # 临时注入的「不可见」身份（验证后从权限文件删掉） FAILS = [] EVIDENCE = [] def sha(p): return hashlib.sha256(open(p, 'rb').read()).hexdigest() if os.path.exists(p) else '' def check(cond, msg, detail=''): print(' %s %s%s' % ('✅' if cond else '❌', msg, (' | ' + detail) if detail else '')) if not cond: FAILS.append(msg + (' | ' + detail if detail else '')) def http(path, token=None): rq = urllib.request.Request('http://127.0.0.1:%d%s' % (PORT, path), headers={'Authorization': 'Bearer ' + token} if token else {}) try: r = urllib.request.urlopen(rq, timeout=30) return r.status, r.read().decode('utf-8', 'replace') except urllib.error.HTTPError as e: return e.code, e.read().decode('utf-8', 'replace') except Exception as e: # 请求没发出去也算失败（如 token 含非 ASCII 导致头编码失败） return 0, '%s: %s' % (type(e).__name__, e) def main(): idx = json.load(open(INDEX_FILE, encoding='utf-8')) svc = importlib.import_module('销售工作台服务') import price_api # ── 注入会话（内存，不写盘、不用真实凭据）── # ⚠️ token 必须纯 ASCII：HTTP 头按 latin-1 编码，含中文名会抛 UnicodeEncodeError、 # 请求根本发不出去（旧版就是这么静默失败的）。用户名放 session 里，不放 token 里。 toks = {} for _i, (nm, role, dept) in enumerate( ((ADMIN, 'admin', '诊断原料销售部'), (SALES, 'sales', '大客户部'), (SALES2, 'sales', '销售拓展部'), (NONEU, 'sales', '销售拓展部'))): t = 'probe%d-%d' % (_i, int(time.time() * 1000) % 1000000) toks[nm] = t svc.AUTH_SESSIONS[t] = {'name': nm, 'role': role, 'dept': dept, 'line': 'da', 'email': nm + '@probe.local', 'expire_at': time.time() + 900} # ── 找一个「该销售有成交、别人也有成交」的货号当靶子 ── def own_customers(sku, who): return set(r['customer'] for r in idx['skus'][sku]['records'] if (r.get('sales') or '') == who) target = None for sku, rec in sorted(idx['skus'].items(), key=lambda kv: -(kv[1].get('stats') or {}).get('deal_count', 0))[:60]: n_mine = len(own_customers(sku, SALES)) tot = len(rec['records']) if n_mine >= 1 and tot >= n_mine + 2: target = sku break check(bool(target), '找到用于隔离验证的靶子货号', target or '未找到') if not target: return 1 rec = idx['skus'][target] mine_c = own_customers(target, SALES) other_c = set(r['customer'] for r in rec['records'] if r['customer']) - mine_c market_n = rec['stats']['deal_count'] mine_n = len([r for r in rec['records'] if (r.get('sales') or '') == SALES]) print('\n 靶子货号 %s：全库 %d 次成交 / %s 自己 %d 次 / %d 个客户' % (target, market_n, SALES, mine_n, len(mine_c))) perm_before = sha(PERM_FILE) perm_raw = open(PERM_FILE, 'rb').read() if os.path.exists(PERM_FILE) else b'{}' perms_backup = json.loads(perm_raw.decode('utf-8') or '{}') srv = svc.ThreadingHTTPServer(('127.0.0.1', PORT), svc.Handler) threading.Thread(target=srv.serve_forever, daemon=True).start() time.sleep(1.0) try: # ═══ ① 普通销售：own ═══ print('\n① 普通销售 %s（默认 own）' % SALES) code, txt = http('/api/price/query?sku=%s' % target, toks[SALES]) d = json.loads(txt) check(code == 200 and d.get('ok'), 'HTTP 200', 'code=%s' % code) check(d.get('scope') == 'own', 'scope=own', str(d.get('scope'))) check(d.get('records_total') == mine_n, '明细条数 == 本人成交条数', '%s vs %s' % (d.get('records_total'), mine_n)) got_c = set(r['customer'] for r in d.get('records', [])) check(got_c == mine_c, '返回客户集 == 本人客户集', '返回 %d 家 / 期望 %d 家' % (len(got_c), len(mine_c))) check(bool(d.get('market')), '全库区间可见（含客户数）', '全库客户数 %s（不点名）' % (d.get('market') or {}).get('customer_count')) leaks = [c for c in other_c if c and c in txt] real_leaks = [c for c in leaks if not any(c in m for m in mine_c)] check(not real_leaks, '响应全文无其他客户名（%d 个候选）' % len(other_c), ('泄漏：%s' % real_leaks[:3]) if real_leaks else '干净') check(all('sales' not in r for r in (d.get('records') or [])), 'own 口径明细记录不含销售员字段（%d 条）' % len(d.get('records') or []), '注意：不能对全文搜 "sales" —— 响应里 role 字段本身就等于 "sales"，会误判') EVIDENCE.append('① %s 查 %s：records_total=%s（全库 %s），客户 %s；其他客户名 %d 个全部未出现。' % (SALES, target, d.get('records_total'), market_n, '、'.join(sorted(got_c)) or '无', len(other_c))) # ═══ ② admin：all ═══ print('\n② admin %s（全量）' % ADMIN) code, txt = http('/api/price/query?sku=%s' % target, toks[ADMIN]) d2 = json.loads(txt) check(code == 200 and d2.get('ok'), 'HTTP 200', 'code=%s' % code) check(d2.get('scope') == 'all', 'scope=all', str(d2.get('scope'))) check(d2.get('records_total') == market_n, '明细条数 == 全库成交次数', '%s vs %s' % (d2.get('records_total'), market_n)) check(len(set(r['customer'] for r in d2.get('records', []))) > len(mine_c), '看到多于自己的客户数', '%s > %s' % (len(set(r['customer'] for r in d2.get('records', []))), len(mine_c))) check('"sales"' in txt, 'all 口径含销售员字段（便于追责）') EVIDENCE.append('② %s 查同一货号：records_total=%s（含 %d 家客户，含 %s 自己的 %d 条）。' % (ADMIN, d2.get('records_total'), len(set(r['customer'] for r in d2.get('records', []))), SALES, mine_n)) # ═══ ⑨ 年份筛选（2024/2025/2026）═══ # ⚠️ 必须放在 ⑤ 之前：⑤ 会把 SALES 的 price_scope 改成 all，之后再验 own 口径就失真。 print('\n⑨ 年份筛选：逐年查同一货号，数字必须对得上索引（在 ⑤ 改权限之前跑）') yopt = d2.get('year_options') or [] check(len(yopt) == 4, 'year_options 含 全部/2024/2025/2026', '、'.join('%s=%s' % (o['year'], o.get('record_count')) for o in yopt)) meta_by_year = (idx.get('meta') or {}).get('by_year') or {} for o in yopt: if o['year'] == 'all': continue exp = (meta_by_year.get(o['year']) or {}).get('record_count') check(o.get('record_count') == exp, '年份 %s 的 year_options.record_count == meta.by_year' % o['year'], '%s vs %s' % (o.get('record_count'), exp)) all_exp = rec['records'] ysum = 0 ycnt = {} for y in ('2024', '2025', '2026'): exp_recs = [r for r in all_exp if str(r.get('year')) == y] exp_own = [r for r in exp_recs if (r.get('sales') or '') == SALES] ysum += len(exp_recs) ycnt[y] = len(exp_recs) code, txt = http('/api/price/query?sku=%s&year=%s' % (target, y), toks[ADMIN]) dy = json.loads(txt) if txt.strip().startswith('{') else {} check(code == 200 and dy.get('year') == y, 'admin 查 %s 返回 year=%s' % (y, y), 'code=%s year=%s' % (code, dy.get('year'))) check(dy.get('records_total') == len(exp_recs), '%s 年 records_total == 索引里该年条数' % y, '%s vs %s' % (dy.get('records_total'), len(exp_recs))) check(all(str(r.get('year')) == y for r in (dy.get('records') or [])), '%s 年返回明细全部属该年（%d 条）' % (y, len(dy.get('records') or []))) mk = dy.get('market') or {} check(mk.get('deal_count') == len(exp_recs), '%s 年 market.deal_count == 该年条数' % y, '%s vs %s' % (mk.get('deal_count'), len(exp_recs))) check(bool(dy.get('sku_by_year')), '%s 响应含 sku_by_year 年度分布' % y, '年份键 %s' % sorted((dy.get('sku_by_year') or {}).keys())) # own 口径 + 年份：条数对得上，且不得出现其他客户名 code2, txt2 = http('/api/price/query?sku=%s&year=%s' % (target, y), toks[SALES]) d9 = json.loads(txt2) if txt2.strip().startswith('{') else {} check(d9.get('scope') == 'own', '%s 年 SALES 仍是 own 口径（未被 ⑤ 污染）' % y, str(d9.get('scope'))) check(d9.get('records_total') == len(exp_own), '%s 年 own 口径条数 == 本人该年条数' % y, '%s vs %s' % (d9.get('records_total'), len(exp_own))) lk = [c for c in other_c if c and c in txt2 and not any(c in mc for mc in mine_c)] check(not lk, '%s 年 own 口径响应无其他客户名' % y, (str(lk[:2]) if lk else '干净')) check(ysum == len(all_exp), '三年条数之和 == 全部条数（无重叠/无遗漏）', '%s vs %s' % (ysum, len(all_exp))) code, txt = http('/api/price/query?sku=%s&year=1999' % target, toks[ADMIN]) dX = json.loads(txt) if txt.strip().startswith('{') else {} check(dX.get('year') == 'all', '非法年份参数回落到 全部', str(dX.get('year'))) EVIDENCE.append('⑨ 年份筛选：%s 2024/2025/2026 明细 %d/%d/%d 条，逐年之和 == 全部 %d 条；' 'own 口径逐年均无其他客户名、均未出现销售员字段。' % ( target, ycnt['2024'], ycnt['2025'], ycnt['2026'], len(all_exp))) # ═══ ⑤ 权限面板改「全量」后立即生效（改完还原）══ print('\n⑤ 权限面板把 %s 调成「价格查询=全量」' % SALES) body = json.dumps({'name': SALES, 'price_scope': 'all'}).encode('utf-8') rq = urllib.request.Request('http://127.0.0.1:%d/api/permissions' % PORT, data=body, method='POST', headers={'Authorization': 'Bearer ' + toks[ADMIN], 'Content-Type': 'application/json'}) rc = json.loads(urllib.request.urlopen(rq, timeout=30).read().decode('utf-8')) check(rc.get('ok'), 'POST /api/permissions 保存成功', str(rc.get('error') or '')) code, txt = http('/api/price/query?sku=%s' % target, toks[SALES]) d5 = json.loads(txt) check(d5.get('scope') == 'all' and d5.get('records_total') == market_n, '改完全量立即生效', 'scope=%s records_total=%s' % (d5.get('scope'), d5.get('records_total'))) EVIDENCE.append('⑤ %s 被授权「全量」后同一请求 records_total 由 %s → %s（无缓存、即时生效）。' % (SALES, mine_n, d5.get('records_total'))) # ═══ ③ 未授权 → 403 ═══ print('\n③ 未授权身份（权限面板=不可见）') perms = json.load(open(PERM_FILE, encoding='utf-8')) perms[NONEU] = dict(perms.get(NONEU) or {}, price_scope='none') json.dump(perms, open(PERM_FILE, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) code, txt = http('/api/price/query?sku=%s' % target, toks[NONEU]) check(code == 403, 'HTTP 403', 'code=%s %s' % (code, txt[:120])) EVIDENCE.append('③ 权限面板勾「不可见」的账号查同一货号：HTTP 403 → %s' % txt[:90]) # ═══ ④ 无 token → 401 ═══ print('\n④ 无 token') code, txt = http('/api/price/query?sku=%s' % target) check(code == 401, 'HTTP 401', 'code=%s' % code) # ═══ ⑥ 客户维度模式 ═══ print('\n⑥ 客户维度模式（只给可见客户的货号清单）') cust = sorted(mine_c)[0] code, txt = http('/api/price/query?customer=%s' % urllib.parse.quote(cust), toks[SALES]) d6 = json.loads(txt) check(code == 200 and d6.get('ok') and 'customer_skus' in d6, 'HTTP 200 且返回货号清单', '货号数 %s' % len(d6.get('customer_skus') or [])) check(all(r['deal_count'] >= 1 for r in (d6.get('customer_skus') or [])), '清单内每条都有成交') EVIDENCE.append('⑥ %s 查自己客户「%s」：返回 %d 个货号的价格清单。' % (SALES, cust, len(d6.get('customer_skus') or []))) # ═══ ⑦ 权限接口下发口径 ═══ print('\n⑦ GET /api/permissions 含 price_scope') code, txt = http('/api/permissions', toks[ADMIN]) dp = json.loads(txt) mp = {u['name']: u.get('price_scope') for u in dp.get('users', [])} check(mp.get(ADMIN) == 'all', 'admin 默认全量', 'admin=%s' % mp.get(ADMIN)) check(mp.get(SALES2) == 'own', 'sales 默认仅自己客户', '%s=%s' % (SALES2, mp.get(SALES2))) EVIDENCE.append('⑦ 权限面板下发：%s=%s（默认全量）、%s=%s（默认仅自己客户）' % (ADMIN, mp.get(ADMIN), SALES2, mp.get(SALES2))) # ═══ ⑧ 第二个销售身份交叉验证 ═══ print('\n⑧ 第二个销售 %s 交叉验证' % SALES2) m2 = own_customers(target, SALES2) code, txt = http('/api/price/query?sku=%s' % target, toks[SALES2]) d8 = json.loads(txt) got2 = set(r['customer'] for r in d8.get('records', [])) check(got2 == m2, '%s 只看到自己的客户' % SALES2, '返回 %d 家 / 期望 %d 家' % (len(got2), len(m2))) EVIDENCE.append('⑧ %s 查同一货号：records_total=%s（客户 %d 家，与 %s 互不重叠=%s）。' % (SALES2, d8.get('records_total'), len(got2), SALES, not (got2 & mine_c))) finally: srv.shutdown() # 还原权限文件（逐字节还原快照） open(PERM_FILE, 'wb').write(perm_raw) print('\n 权限文件还原：%s（改动前 sha=%s）' % ('✅ 一致' if sha(PERM_FILE) == perm_before else '❌ 不一致', perm_before[:12])) print('\n──────── 实测证据 ────────') for e in EVIDENCE: print(' ' + e) print('\n%s' % ('✅ 全部断言通过（%d 条证据）' % len(EVIDENCE) if not FAILS else '❌ 失败 %d 条：%s' % (len(FAILS), FAILS))) return 1 if FAILS else 0 if __name__ == '__main__': sys.exit(main())