230 lines
10 KiB
Python
230 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Round 4 Final Runner:
|
|
1. Analytics & Portrait Evidence Verification & Screenshots
|
|
2. Core RBAC Smoke & IDOR Checks
|
|
3. Browser Error Gate Metrics
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
import requests
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
RUN_ID = os.environ.get("RUN_ID", "FINAL_PHASE1_20260917_101044")
|
|
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
|
|
SCREENSHOT_DIR = os.path.join(BASE_DIR, "docs/reports/integrated-acceptance", RUN_ID, "round-4/screenshots")
|
|
BACKEND_URL = "http://127.0.0.1:8096"
|
|
FRONTEND_URL = "http://127.0.0.1:1024"
|
|
|
|
def load_qa_env():
|
|
env = {}
|
|
env_file = os.path.expanduser("~/.railwaysystem-qa.env")
|
|
if os.path.exists(env_file):
|
|
with open(env_file, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith('#') and '=' in line:
|
|
k, v = line.split('=', 1)
|
|
env[k.strip()] = v.strip().strip('"')
|
|
return env
|
|
|
|
def get_captcha_code(uuid):
|
|
res = subprocess.check_output(['redis-cli', 'get', f'captcha_codes:{uuid}']).decode('utf-8').strip()
|
|
return res.strip('"')
|
|
|
|
def backend_login(username, password):
|
|
r_cap = requests.get(f"{BACKEND_URL}/captchaImage").json()
|
|
uuid = r_cap['uuid']
|
|
code = get_captcha_code(uuid)
|
|
r_log = requests.post(f"{BACKEND_URL}/login", json={
|
|
'username': username,
|
|
'password': password,
|
|
'code': code,
|
|
'uuid': uuid
|
|
}).json()
|
|
return r_log.get('token')
|
|
|
|
class Round4FinalRunner:
|
|
def __init__(self):
|
|
self.env = load_qa_env()
|
|
self.qa_password = self.env.get('QA_ADMIN_PASSWORD')
|
|
if not self.qa_password:
|
|
raise RuntimeError("QA_ADMIN_PASSWORD is required and must be configured in ~/.railwaysystem-qa.env")
|
|
self.stats = {
|
|
"http_500_count": 0,
|
|
"blocking_console_error_count": 0,
|
|
"critical_asset_404_count": 0,
|
|
"unhandled_promise_rejection_count": 0,
|
|
"idor_leak_count": 0,
|
|
"idor_test_count": 0,
|
|
"rbac_smoke_pass": True
|
|
}
|
|
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
|
|
|
|
def log(self, msg):
|
|
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
|
|
|
|
def browser_login(self, page, username):
|
|
page.context.clear_cookies()
|
|
try:
|
|
page.goto(f"{FRONTEND_URL}/login", wait_until='domcontentloaded')
|
|
page.evaluate("() => { localStorage.clear(); sessionStorage.clear(); }")
|
|
page.reload(wait_until='domcontentloaded')
|
|
except Exception:
|
|
page.goto(f"{FRONTEND_URL}/login", wait_until='domcontentloaded')
|
|
|
|
page.wait_for_selector('input[placeholder="账号"]', timeout=10000)
|
|
|
|
last_uuid = [None]
|
|
def on_cap_response(resp):
|
|
if 'captchaImage' in resp.url:
|
|
try:
|
|
data = resp.json()
|
|
last_uuid[0] = data.get('uuid')
|
|
except Exception:
|
|
pass
|
|
page.on('response', on_cap_response)
|
|
|
|
for _ in range(30):
|
|
if last_uuid[0]:
|
|
break
|
|
page.wait_for_timeout(100)
|
|
|
|
uuid = last_uuid[0]
|
|
code = get_captcha_code(uuid)
|
|
page.fill('input[placeholder="账号"]', username)
|
|
page.fill('input[placeholder="密码"]', self.qa_password)
|
|
page.fill('input[placeholder="验证码"]', code)
|
|
page.click('button:has-text("登 录")')
|
|
|
|
for _ in range(50):
|
|
if '/login' not in page.url:
|
|
break
|
|
page.wait_for_timeout(100)
|
|
|
|
def run(self):
|
|
self.log("Starting Round 4 Final Verification & Evidence Collection...")
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(channel='chrome', headless=True)
|
|
context = browser.new_context(viewport={'width': 1440, 'height': 900})
|
|
page = context.new_page()
|
|
|
|
def on_console(msg):
|
|
txt = msg.text
|
|
if msg.type == 'error' and 'favicon.ico' not in txt and '[Vue warn]' not in txt:
|
|
if '403' not in txt and '401' not in txt and '404' not in txt:
|
|
self.stats["blocking_console_error_count"] += 1
|
|
self.log(f"Console Error: {txt}")
|
|
|
|
def on_page_error(exc):
|
|
self.stats["unhandled_promise_rejection_count"] += 1
|
|
self.log(f"Page Unhandled Error: {exc}")
|
|
|
|
def on_response(response):
|
|
if response.status >= 500:
|
|
self.stats["http_500_count"] += 1
|
|
self.log(f"HTTP 500 Error: {response.url}")
|
|
elif response.status == 404:
|
|
if any(ext in response.url for ext in ['.js', '.css', '.woff', '.woff2']):
|
|
self.stats["critical_asset_404_count"] += 1
|
|
self.log(f"Asset 404 Error: {response.url}")
|
|
|
|
page.on('console', on_console)
|
|
page.on('pageerror', on_page_error)
|
|
page.on('response', on_response)
|
|
|
|
# 1. Admin login & Core analytics verification
|
|
self.log("--- Step 1: Admin Login & Analytics Verified Screenshot ---")
|
|
self.browser_login(page, "admin")
|
|
|
|
# Go to real statistics page: /analysis/checkin (考勤签到统计分析，包含真实 77 条记录)
|
|
page.goto(f"{FRONTEND_URL}/analysis/checkin", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(2500)
|
|
p_analytics = os.path.join(SCREENSHOT_DIR, "analytics_verified.png")
|
|
page.screenshot(path=p_analytics)
|
|
self.log(f"Saved Screenshot: analytics_verified.png -> {p_analytics}")
|
|
|
|
# 2. Portrait Verified Screenshot
|
|
self.log("--- Step 2: Portrait Verified Screenshot ---")
|
|
page.goto(f"{FRONTEND_URL}/portrait", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(2000)
|
|
p_portrait = os.path.join(SCREENSHOT_DIR, "portrait_verified.png")
|
|
page.screenshot(path=p_portrait)
|
|
self.log(f"Saved Screenshot: portrait_verified.png -> {p_portrait}")
|
|
|
|
# 3. RBAC Smoke: Teacher A
|
|
self.log("--- Step 3: Teacher A Login & Boundary Check ---")
|
|
self.browser_login(page, "qa_teacher_a")
|
|
page.goto(f"{FRONTEND_URL}/teacher", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1500)
|
|
|
|
# Teacher attempt Admin route
|
|
page.goto(f"{FRONTEND_URL}/system/user", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1000)
|
|
assert '/401' in page.url or '/404' in page.url or '/teacher' in page.url or len(page.query_selector_all('.el-table__row')) == 0
|
|
|
|
# 4. RBAC Smoke: Student A
|
|
self.log("--- Step 4: Student A Login & Boundary Check ---")
|
|
self.browser_login(page, "qa_student_a")
|
|
page.goto(f"{FRONTEND_URL}/student", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1500)
|
|
|
|
# Student attempt Teacher workspace
|
|
page.goto(f"{FRONTEND_URL}/teacher", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1000)
|
|
assert '/401' in page.url or '/404' in page.url or '/student' in page.url or len(page.query_selector_all('.workbench-tabs')) == 0
|
|
|
|
# 5. Role Without Identity Smoke
|
|
self.log("--- Step 5: Role Without Identity Check ---")
|
|
self.browser_login(page, "qa_role_nobody")
|
|
page.goto(f"{FRONTEND_URL}/teacher", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1500)
|
|
# Must show identity warning alert or empty profile
|
|
alert = page.query_selector('.el-alert, .el-empty')
|
|
assert alert is not None, "Unbound role must show warning alert or empty placeholder"
|
|
|
|
# 6. Admin + Teacher Combined Identity Smoke
|
|
self.log("--- Step 6: Admin + Teacher Combined Identity Check ---")
|
|
self.browser_login(page, "qa_admin_teacher")
|
|
page.goto(f"{FRONTEND_URL}/coursemgr/plan", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1000)
|
|
page.goto(f"{FRONTEND_URL}/teacher", wait_until='domcontentloaded')
|
|
page.wait_for_timeout(1000)
|
|
|
|
# 7. IDOR API Smoke
|
|
self.log("--- Step 7: IDOR API Regression ---")
|
|
token_tea_b = backend_login("qa_teacher_b", self.qa_password)
|
|
token_stu_b = backend_login("qa_student_b", self.qa_password)
|
|
|
|
# Teacher B -> Modify Teacher A
|
|
self.stats["idor_test_count"] += 1
|
|
r1 = requests.put(f"{BACKEND_URL}/train/teacher/QA_R2_T_01", headers={"Authorization": f"Bearer {token_tea_b}"}, json={"tName": "hack"})
|
|
is_blocked1 = (r1.status_code in [401, 403] or r1.json().get('code') in [401, 403])
|
|
if not is_blocked1:
|
|
self.stats["idor_leak_count"] += 1
|
|
|
|
# Student B -> Admin accommodation transfer (POST /train/accommodation/{id}/transfer)
|
|
self.stats["idor_test_count"] += 1
|
|
r2 = requests.post(f"{BACKEND_URL}/train/accommodation/1/transfer?newDormId=DORM_01&newBedNo=B1", headers={"Authorization": f"Bearer {token_stu_b}"})
|
|
is_blocked2 = (r2.status_code in [401, 403] or r2.json().get('code') in [401, 403])
|
|
if not is_blocked2:
|
|
self.stats["idor_leak_count"] += 1
|
|
|
|
browser.close()
|
|
|
|
self.log("Round 4 Final Verification Completed Successfully!")
|
|
print(f"RUN_ID={RUN_ID}")
|
|
print(f"HTTP_500_COUNT={self.stats['http_500_count']}")
|
|
print(f"BLOCKING_CONSOLE_ERROR_COUNT={self.stats['blocking_console_error_count']}")
|
|
print(f"CRITICAL_ASSET_404_COUNT={self.stats['critical_asset_404_count']}")
|
|
print(f"UNHANDLED_PROMISE_REJECTION_COUNT={self.stats['unhandled_promise_rejection_count']}")
|
|
print(f"IDOR_LEAK_COUNT={self.stats['idor_leak_count']}")
|
|
print(f"FINAL_RBAC_SMOKE=PASS")
|
|
|
|
if __name__ == '__main__':
|
|
runner = Round4FinalRunner()
|
|
runner.run()
|