""" TDD Unit & Integration Test Suite for Independent Security & Anti-Cheat Division. Validates: 1. Security Audit Gate & Veto Power enforcement on Critical/High vulnerabilities. 2. Automated Red Team Fuzzer attack scenarios (Speedhack, Dupe, Replay, Tamper, Bot). 3. Economic Anomaly Detector & Emergency Quarantine trigger. 4. Binary Integrity Attestor & Anti-Hooking detection. 5. Independent Security Audit Runner & Consolidated Audit Report generation. """ import unittest import sys import os import json # Add server directory to sys.path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../server"))) from security.independent_security_auditor import ( SeverityLevel, SecurityVulnerability, SecurityAuditGate, SecurityVetoException, AutomatedRedTeamFuzzer, EconomicAnomalyDetector, BinaryIntegrityAttestor, IndependentSecurityAuditor, AuditReportStatus, ) class TestIndependentSecurityDivision(unittest.TestCase): def setUp(self): self.auditor = IndependentSecurityAuditor() def test_security_gate_veto_on_critical_vulnerability(self): """Security Gate MUST raise SecurityVetoException when CRITICAL vulnerability is unpatched.""" gate = SecurityAuditGate() crit_vuln = SecurityVulnerability( vuln_id="SEC-001", title="Two-Phase Commit Race Condition Dupe Exploit", severity=SeverityLevel.CRITICAL, component="trade.two_phase_commit", description="Concurrent buyer requests can bypass mutex if timeout occurs", proof_of_concept="Simultaneous buyout triggers negative balance", remediation="Enforce atomic Redis lock with distributed fencing token" ) gate.record_vulnerability(crit_vuln) # Must fail release check and trigger VETO self.assertTrue(gate.has_veto_condition()) with self.assertRaises(SecurityVetoException) as ctx: gate.enforce_release_gate(build_id="v2026.1-rc1") self.assertIn("CRITICAL", str(ctx.exception)) self.assertIn("SEC-001", str(ctx.exception)) def test_security_gate_passes_when_clean_or_low_severity_only(self): """Security Gate passes when only LOW/INFO findings exist and no CRITICAL/HIGH.""" gate = SecurityAuditGate() low_vuln = SecurityVulnerability( vuln_id="SEC-002", title="Verbose Debug Log in Staging", severity=SeverityLevel.LOW, component="gateway.network_gateway", description="Log outputs partial session identifier", proof_of_concept="Check server.log", remediation="Mask session token in logger" ) gate.record_vulnerability(low_vuln) self.assertFalse(gate.has_veto_condition()) result = gate.enforce_release_gate(build_id="v2026.1-rc2") self.assertTrue(result) def test_automated_red_team_fuzzer_speedhack_defense(self): """Red team fuzzer attacks movement authority with speedhack injection -> Defense blocks.""" fuzzer = AutomatedRedTeamFuzzer() report = fuzzer.run_speedhack_injection_attack(player_id=101, base_speed=6.0, claimed_leap=150.0) self.assertTrue(report.is_blocked) self.assertGreater(report.suspicion_score, 0) self.assertEqual(report.reconciled_coords, (10.0, 10.0)) def test_automated_red_team_fuzzer_dupe_attack(self): """Red team fuzzer executes concurrent double-spend race condition attack -> Defense blocks.""" fuzzer = AutomatedRedTeamFuzzer() report = fuzzer.run_concurrent_dupe_attack( item_uuid="primal_ring_99", seller_id="trader_a", buyer1_id="exploiter_b", buyer2_id="exploiter_c", price=100 ) self.assertTrue(report.is_blocked) self.assertEqual(report.successful_purchases, 1) self.assertEqual(report.blocked_purchases, 1) self.assertFalse(report.item_duplicated) def test_automated_red_team_fuzzer_tampered_packet(self): """Red team fuzzer injects bit-flipped ciphertext -> AEAD Poly1305 authentication fails.""" fuzzer = AutomatedRedTeamFuzzer() report = fuzzer.run_bitflip_tampering_attack(payload=b"ADD_CURRENCY_HON_NGUYEN_999999") self.assertTrue(report.is_blocked) self.assertTrue("tamper" in report.defense_trigger.lower() or "signature" in report.defense_trigger.lower()) def test_economic_anomaly_detector_triggers_emergency_quarantine(self): """Economic Auditor detects abnormal velocity of Primal Stone inflow -> Triggers Quarantine.""" detector = EconomicAnomalyDetector(max_hourly_currency_inflow=5000) # Normal transaction detector.record_inflow(account_id="acc_fair_player", currency="HonNguyen", amount=200) self.assertFalse(detector.is_quarantined("acc_fair_player")) # Exploit inflow spike (e.g. 50,000 stones in 10 minutes) detector.record_inflow(account_id="acc_suspect", currency="HonNguyen", amount=60000) self.assertTrue(detector.is_quarantined("acc_suspect")) # Verify quarantine status & event log event = detector.get_quarantine_record("acc_suspect") self.assertIsNotNone(event) self.assertEqual(event["status"], "LOCKED_FOR_AUDIT") self.assertGreater(event["inflow_amount"], 5000) def test_binary_integrity_attestor_detects_jailbreak_and_repack(self): """Binary Integrity Attestor verifies client signature and flags cracked/jailbroken client.""" attestor = BinaryIntegrityAttestor(expected_bundle_id="com.freeexile.game.ios") # Valid client valid_res = attestor.verify_client_environment( bundle_id="com.freeexile.game.ios", has_jailbreak_artifacts=False, is_debugger_attached=False, signature_fingerprint="SHA256:OFFICIAL_APPLE_CERT_2026" ) self.assertTrue(valid_res.is_trusted) # Attacked client: Cracked bundle + Frida detected attacked_res = attestor.verify_client_environment( bundle_id="com.freeexile.cracked.mod", has_jailbreak_artifacts=True, is_debugger_attached=True, signature_fingerprint="UNKNOWN_REPACK_CERT" ) self.assertFalse(attacked_res.is_trusted) self.assertIn("BUNDLE_ID_MISMATCH", attacked_res.violation_flags) self.assertIn("JAILBREAK_DETECTED", attacked_res.violation_flags) self.assertIn("DEBUGGER_ATTACHED", attacked_res.violation_flags) def test_consolidated_independent_security_audit_report(self): """Auditor runs end-to-end full audit, executes Red Team drills, and compiles report.""" audit_res = self.auditor.execute_full_security_audit( build_version="2026.1-SEC-AUDIT", environment="STAGING" ) self.assertIn(audit_res.status, [AuditReportStatus.PASSED, AuditReportStatus.VETOED]) self.assertGreater(audit_res.tests_executed, 0) self.assertIsNotNone(audit_res.audit_id) markdown_summary = audit_res.to_markdown() self.assertIn("BÁO CÁO KIỂM TOÁN AN NINH ĐỘC LẬP", markdown_summary.upper()) self.assertIn("KHỐI BẢO MẬT & CHỐNG GIAN LẬN", markdown_summary.upper()) if __name__ == "__main__": unittest.main()