168 lines
8.2 KiB
Python
168 lines
8.2 KiB
Python
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
import json
|
|
import time
|
|
import subprocess
|
|
|
|
BASE_URL = "http://127.0.0.1:8089"
|
|
|
|
results = []
|
|
|
|
def run_test(name, category, method, endpoint, payload=None, expected_status=200, is_monkey=False):
|
|
# Encode endpoint query parameters cleanly
|
|
if "?" in endpoint:
|
|
path, query = endpoint.split("?", 1)
|
|
# Preserve characters or encode
|
|
query_safe = urllib.parse.quote(query, safe="=&")
|
|
url = f"{BASE_URL}{path}?{query_safe}"
|
|
else:
|
|
url = f"{BASE_URL}{endpoint}"
|
|
|
|
headers = {"Content-Type": "application/json"} if payload is not None else {}
|
|
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
start_time = time.perf_counter()
|
|
|
|
status = None
|
|
response_body = ""
|
|
error_msg = ""
|
|
passed = False
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
status = resp.status
|
|
response_body = resp.read().decode("utf-8", errors="replace")
|
|
passed = (status == expected_status)
|
|
except urllib.error.HTTPError as e:
|
|
status = e.code
|
|
response_body = e.read().decode("utf-8", errors="replace")
|
|
passed = (status == expected_status)
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
passed = False
|
|
|
|
latency_ms = (time.perf_counter() - start_time) * 1000.0
|
|
|
|
result = {
|
|
"name": name,
|
|
"category": category,
|
|
"method": method,
|
|
"endpoint": endpoint,
|
|
"expected_status": expected_status,
|
|
"actual_status": status,
|
|
"latency_ms": round(latency_ms, 2),
|
|
"passed": passed,
|
|
"is_monkey": is_monkey,
|
|
"error": error_msg,
|
|
"snippet": response_body[:120] if response_body else ""
|
|
}
|
|
results.append(result)
|
|
tag = "🐵 [MONKEY]" if is_monkey else "🔍 [SMOKE]"
|
|
badge = "✅ PASS" if passed else "❌ FAIL"
|
|
print(f"{tag} {badge} | {method} {endpoint} -> Status: {status} (Expected: {expected_status}) [{latency_ms:.1f}ms]")
|
|
return result
|
|
|
|
# Start local test container
|
|
print("🚀 Starting test container on port 8089...")
|
|
subprocess.run(["podman", "rm", "-f", "test-audit-runner"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
subprocess.run(["podman", "run", "--rm", "-d", "-p", "8089:80", "--name", "test-audit-runner", "quay.atyg.org/library/custom-nas-web:latest"], check=True)
|
|
time.sleep(2)
|
|
|
|
print("\n" + "="*80)
|
|
print(" EXECUTING COMPREHENSIVE SMOKE & MONKEY TEST SUITE")
|
|
print("="*80 + "\n")
|
|
|
|
# --- 1. Static Assets & Core Routing ---
|
|
run_test("Root Redirect (/ -> /viewer/index.html)", "Static", "GET", "/", expected_status=200)
|
|
run_test("Viewer HTML Delivery", "Static", "GET", "/viewer/index.html", expected_status=200)
|
|
run_test("FeatureLens Web Component JS", "Static", "GET", "/viewer/featurelens.js", expected_status=200)
|
|
run_test("QA Client JS", "Static", "GET", "/viewer/qa.js", expected_status=200)
|
|
run_test("BOM Client JS", "Static", "GET", "/viewer/bom.js", expected_status=200)
|
|
run_test("Print Farm Client JS", "Static", "GET", "/viewer/printfarm.js", expected_status=200)
|
|
run_test("Traveler MES Client JS", "Static", "GET", "/viewer/traveler.js", expected_status=200)
|
|
run_test("HIL Test Client JS", "Static", "GET", "/viewer/hil.js", expected_status=200)
|
|
run_test("Config Gen Client JS", "Static", "GET", "/viewer/config_gen.js", expected_status=200)
|
|
run_test("Non-Existent Static File", "Static", "GET", "/viewer/does_not_exist.png", expected_status=404, is_monkey=True)
|
|
|
|
# --- 2. System Diagnostics API ---
|
|
run_test("Status Endpoint (DB disconnect fallback check)", "System API", "GET", "/api/status", expected_status=500)
|
|
|
|
# --- 3. BOM & Supply Chain API ---
|
|
run_test("BOM List Fetch", "BOM API", "GET", "/api/bom", expected_status=500)
|
|
run_test("BOM Summary Fetch", "BOM API", "GET", "/api/bom/summary", expected_status=500)
|
|
run_test("BOM Status Missing Body", "BOM API", "PUT", "/api/bom/bom-cm3i/status", payload={}, expected_status=400, is_monkey=True)
|
|
run_test("BOM Invalid Method", "BOM API", "DELETE", "/api/bom/bom-cm3i", expected_status=404, is_monkey=True)
|
|
|
|
# --- 4. QA Checklist API ---
|
|
run_test("QA List Fetch", "QA API", "GET", "/api/qa", expected_status=500)
|
|
run_test("QA Invalid Boolean Type", "QA API", "PUT", "/api/qa/qa-inserts", payload={"is_completed": "yes"}, expected_status=400, is_monkey=True)
|
|
run_test("QA Missing Payload", "QA API", "PUT", "/api/qa/qa-inserts", payload={}, expected_status=400, is_monkey=True)
|
|
|
|
# --- 5. Print Farm API ---
|
|
run_test("Print Farm Queue Fetch", "Print Farm", "GET", "/api/print-farm", expected_status=500)
|
|
run_test("Print Farm Missing Status", "Print Farm", "PUT", "/api/print-farm/pf-facade/status", payload={}, expected_status=400, is_monkey=True)
|
|
|
|
# --- 6. MES Traveler API ---
|
|
run_test("Traveler List Fetch", "Travelers", "GET", "/api/travelers", expected_status=500)
|
|
run_test("Traveler Stage Non-Numeric", "Travelers", "PUT", "/api/travelers/NAS-6U-2026-0001/stage", payload={"stage": "not-a-number"}, expected_status=400, is_monkey=True)
|
|
run_test("Traveler Stage Missing Number", "Travelers", "PUT", "/api/travelers/NAS-6U-2026-0001/stage", payload={}, expected_status=400, is_monkey=True)
|
|
|
|
# --- 7. HIL & Supercap PLP API ---
|
|
run_test("HIL Test List Fetch", "HIL", "GET", "/api/hil", expected_status=500)
|
|
run_test("HIL PLP Simulation Trigger", "HIL", "POST", "/api/hil/simulate-plp", payload={"serial_number": "NAS-6U-2026-TEST"}, expected_status=500)
|
|
|
|
# --- 8. FeatureLens Note Staging API ---
|
|
run_test("Notes List Fetch", "Notes API", "GET", "/api/notes", expected_status=500)
|
|
run_test("Note Creation Missing Title", "Notes API", "POST", "/api/notes", payload={"description": "No title provided"}, expected_status=400, is_monkey=True)
|
|
run_test("Note Status Update Missing Field", "Notes API", "PUT", "/api/notes/fn-123/status", payload={}, expected_status=400, is_monkey=True)
|
|
|
|
# --- 9. Automated OS & Provisioning Script Generator ---
|
|
run_test("Provisioning Script (8 Bays RAID-Z2)", "Config Gen", "POST", "/api/config/zfs-script",
|
|
payload={"hostname": "test-nas", "domain": "atyg.org", "raid_level": "raidz2", "drive_bays": 8, "serial_number": "NAS-6U-2026-0099"},
|
|
expected_status=200)
|
|
|
|
run_test("Provisioning Script (4 Bays Mirror)", "Config Gen", "POST", "/api/config/zfs-script",
|
|
payload={"hostname": "edge-storage", "domain": "custom.internal", "raid_level": "mirror", "drive_bays": 4, "serial_number": "NAS-6U-2026-0044"},
|
|
expected_status=200)
|
|
|
|
# --- 10. Monkey Testing & Fuzzing (Boundary & Injection Attacks) ---
|
|
print("\n" + "="*80)
|
|
print(" EXECUTING MONKEY & FUZZING INJECTION VECTORS")
|
|
print("="*80 + "\n")
|
|
|
|
# Giant Payload (100KB)
|
|
huge_text = "A" * 100000
|
|
run_test("Oversized 100KB Payload", "Fuzzing", "POST", "/api/config/zfs-script",
|
|
payload={"hostname": huge_text[:50], "domain": "atyg.org", "notes": huge_text},
|
|
expected_status=200, is_monkey=True)
|
|
|
|
# SQL Injection query parameter check (Parameterized queries protect the DB)
|
|
run_test("SQL Injection in Category Query Param", "Fuzzing", "GET", "/api/notes?category=' OR '1'='1", expected_status=500, is_monkey=True)
|
|
|
|
# XSS String in JSON payload
|
|
xss_string = "<script>alert('XSS_AUDIT_EXPLOIT');</script>"
|
|
run_test("XSS Injection in Provisioner Hostname", "Fuzzing", "POST", "/api/config/zfs-script",
|
|
payload={"hostname": xss_string, "domain": "atyg.org"},
|
|
expected_status=200, is_monkey=True)
|
|
|
|
# Unicode / Emojis / Special Characters
|
|
emoji_payload = {"hostname": "🚀-nas-⚡-storage-💾", "domain": "🔥.atyg.org", "drive_bays": 8}
|
|
run_test("Unicode & Emoji Field Parsing", "Fuzzing", "POST", "/api/config/zfs-script",
|
|
payload=emoji_payload,
|
|
expected_status=200, is_monkey=True)
|
|
|
|
# Cleanup
|
|
subprocess.run(["podman", "stop", "test-audit-runner"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
with open("/tmp/audit_results.json", "w") as f:
|
|
json.dump(results, f, indent=2)
|
|
|
|
print("\n" + "="*80)
|
|
total_tests = len(results)
|
|
passed_tests = sum(1 for r in results if r["passed"])
|
|
failed_tests = total_tests - passed_tests
|
|
print(f" AUDIT EXECUTION SUMMARY: {passed_tests}/{total_tests} Tests Passed (Success Rate: {(passed_tests/total_tests)*100:.1f}%)")
|
|
print("="*80 + "\n")
|