""" Adversarial Stress-Testing and Empirical Security Challenge for Milestone M1: - Tier 1 Synchronous Trie Profanity Filter Evasion (Homoglyphs, Leetspeak, Diacritics, Zero-width chars) - Tier 2 Asynchronous Sentinel RMT & Black Market Pattern Detection - Comprehensive Auto-Mute Enforcement across all 8 Channels - Cryptographic HMAC-SHA256 Anti-Spoofing and Forgery Resistance - Strict SLA Performance Benchmarking (< 0.1ms filter, < 2ms item query) - Empirical Proof of Edge-Case Evasion Vectors (for M4 Hardening) """ import asyncio import hashlib import hmac import time import unittest from typing import Any, Dict, List, Tuple from server.chat.channel_manager import ChannelManager from server.chat.chat_service import ChatService from server.chat.chat_types import ( ChatChannelType, ItemAffixDTO, ItemSnapshotDTO, SendChatRequestDTO, ) from server.chat.item_link_service import ItemLinkService from server.chat.moderation import ( AsyncSentinelRMTDetector, ChatModerationPipeline, SynchronousTrieFilter, ) class TestChatModerationAdversarial(unittest.IsolatedAsyncioTestCase): """Adversarial stress-test suite challenging the Chat and Moderation subsystem.""" def setUp(self) -> None: self.secret_key = b"FREEEXILE_SECRET_SIGNING_KEY_2026" self.trie_filter = SynchronousTrieFilter() self.pipeline = ChatModerationPipeline() self.item_link_service = ItemLinkService(secret_signing_key=self.secret_key) self.chat_service = ChatService(item_secret_key=self.secret_key, cluster_shards=16) def test_adversarial_homoglyphs_and_leetspeak(self) -> None: """Adversarial homoglyph substitutions (@, 0, 1, !, 3, $, 4, 5, 7, 8).""" attack_payloads: List[Tuple[str, str]] = [ ("thang d.m may", "dot"), ("thang d-k-m may", "dash"), ("thang d_m may", "underscore"), ("thang d...m may", "multi-dot"), ("con l0n", "0 for o"), ("thang c4c", "4 for a"), ("thang c@c", "@ for a"), ("do b!tch", "! for i"), ("do b1tch", "1 for i"), ("thang $h1t", "$ and 1"), ("thang 5h1t", "5 and 1"), ("thang di7", "7 for t"), ("thang 8uoi", "8 for b"), ("what the f.u.c.k", "all dots"), ("f_u_c_k you", "underscores"), ("f*u*c*k this", "asterisks"), ("f/u/c/k that", "slashes"), ] for payload, desc in attack_payloads: has_prof, censored = self.trie_filter.scan_and_censor(payload) self.assertTrue(has_prof, f"Failed profanity detection for '{payload}' ({desc})") self.assertIn("***", censored, f"Missing '***' masking in '{censored}' for '{payload}'") def test_adversarial_vietnamese_diacritics_and_precomposed_unicode(self) -> None: """Adversarial Vietnamese diacritics, precomposed vs decomposed Unicode.""" attack_payloads: List[Tuple[str, str]] = [ ("thang đ.m may", "đ with dot"), ("thang đ_k_m may", "đ with underscore"), ("thang đ.ị.t may", "địt with separated tone marks"), ("thang đ-ị-t may", "đ-ị-t with dash"), ("thang đ ị t may", "spaced characters"), ("con l.ồ.n may", "lồn with accents"), ("thang c.ặ.c may", "cặc with breve and dot"), ("thang b.u.ồ.i may", "buồi with grave"), ("thang d.c.m may", "dcm compound"), ("thang v.c.l may", "vcl compound"), ("thang v.k.l may", "vkl compound"), ] for payload, desc in attack_payloads: has_prof, censored = self.trie_filter.scan_and_censor(payload) self.assertTrue(has_prof, f"Failed profanity detection for '{payload}' ({desc})") self.assertIn("***", censored, f"Missing '***' masking in '{censored}' for '{payload}'") def test_adversarial_zero_width_character_injection(self) -> None: """Attackers inject zero-width spaces (\\u200b, \\ufeff) to split tokens.""" has_prof1, censored1 = self.trie_filter.scan_and_censor("thang d\u200bm may") self.assertTrue(has_prof1, "Failed to detect zero-width space \\u200b injection") self.assertIn("***", censored1) has_prof2, censored2 = self.trie_filter.scan_and_censor("con l\ufeffon") self.assertTrue(has_prof2, "Failed to detect zero-width space \\ufeff injection") self.assertIn("***", censored2) def test_legitimate_game_lexicon_not_false_positively_censored(self) -> None: """Verify non-profane words containing substrings (e.g. 'class', 'account') are NOT masked.""" safe_messages = [ "chao moi nguoi, toi chon class Kiem Tu", "he thong account rat tot", "toi can mua classic sword", "di danh boss bang hoi luc 8 gio toi nhe", "co ai muon vao party farm map khong?", "nang cap trang bi tai tho ren", ] for safe_msg in safe_messages: has_prof, result_text = self.trie_filter.scan_and_censor(safe_msg) self.assertFalse(has_prof, f"False positive triggered on: '{safe_msg}'") self.assertEqual(result_text, safe_msg) def test_adversarial_rmt_phone_and_payment_combinations(self) -> None: """Adversarial RMT inputs combining phones, social tags, and payment keywords.""" # Phone + Payment -> Risk 90 -> Auto-mute res1 = self.pipeline.process("Ban vang gia cuc re chuyen khoan atm qua so 0987654321 nhe") self.assertTrue(res1.is_rmt and res1.should_auto_mute and res1.risk_score >= 60) self.assertTrue({"PHONE_VN", "PAYMENT_KEYWORD"}.issubset(set(res1.violation_reasons))) # Zalo handle + MoMo payment -> Risk 85 -> Auto-mute res2 = self.pipeline.process("Shopacc gia re lien he zalo: shop_game88 thanh toan momo") self.assertTrue(res2.is_rmt and res2.should_auto_mute and res2.risk_score >= 60) self.assertTrue({"SOCIAL_HANDLE", "PAYMENT_KEYWORD"}.issubset(set(res2.violation_reasons))) # Telegram handle + Crypto USDT -> Risk 80 -> Auto-mute res3 = self.pipeline.process("Thu mua ngoc so luong lon inbox telegram @crypto_king nhan usdt") self.assertTrue(res3.is_rmt and res3.should_auto_mute and res3.risk_score >= 60) self.assertTrue({"SOCIAL_HANDLE", "CRYPTO_PAYMENT"}.issubset(set(res3.violation_reasons))) # Clean in-game trade talk -> Risk 0 -> No auto-mute res_safe = self.pipeline.process("Toi can ban cay kiem pham cap 50 doi 100 Hon Nguyen Thach trong game") self.assertFalse(res_safe.is_rmt or res_safe.should_auto_mute) self.assertEqual(res_safe.risk_score, 0) async def test_auto_mute_enforcement_blocks_all_eight_channels(self) -> None: """Triggering auto-mute persists and shuts down chat on ALL 8 channels.""" sender_id = 99901 rmt_req = SendChatRequestDTO( sender_id=sender_id, sender_name="RMTBot99", sender_level=50, channel=ChatChannelType.ZONE, zone_id="zone_sanctuary", content="Ban vang SLL chuyen khoan atm lien he zalo: @boss88 qua so 0912345678" ) res_rmt = await self.chat_service.handle_send_chat(rmt_req) self.assertFalse(res_rmt.success) self.assertIn("khóa chat", res_rmt.error_message) is_muted, rem_sec = self.chat_service.is_player_muted(sender_id) self.assertTrue(is_muted and rem_sec > 0) channels_to_test: List[Tuple[ChatChannelType, Dict[str, Any]]] = [ (ChatChannelType.WORLD, {"content": "Hello world"}), (ChatChannelType.ZONE, {"zone_id": "zone_sanctuary", "content": "Hello zone"}), (ChatChannelType.RECRUIT, {"content": "Recruit party"}), (ChatChannelType.GUILD, {"guild_id": "guild_01", "content": "Hello guild"}), (ChatChannelType.PARTY, {"party_id": "party_01", "content": "Hello party"}), (ChatChannelType.WHISPER, {"target_id": 12345, "content": "Secret message"}), (ChatChannelType.FEEDBACK, {"content": "GM please help"}), (ChatChannelType.SYSTEM, {"content": "Fake system announcement"}), ] for ch_type, extra_kwargs in channels_to_test: attempt = SendChatRequestDTO( sender_id=sender_id, sender_name="RMTBot99", sender_level=50, channel=ch_type, **extra_kwargs ) res = await self.chat_service.handle_send_chat(attempt) self.assertFalse(res.success, f"Auto-muted player sent message on {ch_type.name}") self.assertIn("tạm khóa chat", res.error_message) # Honest player is unaffected clean_req = SendChatRequestDTO( sender_id=99902, sender_name="HonestPlayer", sender_level=50, channel=ChatChannelType.ZONE, zone_id="zone_sanctuary", content="Toi di danh boss day" ) res_clean = await self.chat_service.handle_send_chat(clean_req) self.assertTrue(res_clean.success) # Admin un-mutes self.chat_service.unmute_player(sender_id) is_muted_after, _ = self.chat_service.is_player_muted(sender_id) self.assertFalse(is_muted_after) def test_hmac_tamper_attacks_rejected(self) -> None: """Adversarial tampering of item attributes and signatures.""" uuid_val, name_key, legit_rarity = "item_godly_sword_001", "Thai Co Long Kiem", 1 snapshot = self.item_link_service.create_item_snapshot( item_uuid=uuid_val, item_name_key=name_key, rarity=legit_rarity, element=1, quality=0, item_level=10, crafter_name="Tan Thu", affixes=(ItemAffixDTO(affix_id="aff_1", stat_name="Cong", stat_value=10.0),) ) valid_sig = snapshot.hmac_signature self.assertTrue(self.item_link_service.verify_item_signature(uuid_val, name_key, legit_rarity, valid_sig)) # Tampered rarity (1 -> 5) self.assertFalse(self.item_link_service.verify_item_signature(uuid_val, name_key, 5, valid_sig)) # Tampered item_uuid self.assertFalse(self.item_link_service.verify_item_signature("item_stolen_sword_999", name_key, legit_rarity, valid_sig)) # Tampered item_name_key self.assertFalse(self.item_link_service.verify_item_signature(uuid_val, "Huyen Vu Trong Kiem", legit_rarity, valid_sig)) # Bit-flipping attack on signature corrupted_sig = ("0" if valid_sig[0] != "0" else "1") + valid_sig[1:] query_tampered = self.item_link_service.query_item_snapshot(uuid_val, corrupted_sig) self.assertFalse(query_tampered.is_valid) self.assertIn("mismatch", query_tampered.error_message.lower()) # Non-existent item UUID self.assertFalse(self.item_link_service.query_item_snapshot("fake_uuid_none", valid_sig).is_valid) # Expired snapshot rejection expired_snap = self.item_link_service.create_item_snapshot( item_uuid="item_expired_001", item_name_key="Kiem Cu", rarity=1, element=0, quality=0, item_level=1, crafter_name="Vo Danh", ttl_seconds=-10 ) self.assertFalse(self.item_link_service.query_item_snapshot("item_expired_001", expired_snap.hmac_signature).is_valid) def test_sla_tier1_moderation_latency(self) -> None: """Measure Tier 1 moderation latency over 1,000 iterations (SLA: < 0.1ms).""" import tracemalloc was_tracing = tracemalloc.is_tracing() if was_tracing: tracemalloc.stop() benchmark_payloads = [ "thang d.m may c_a_c f-u-c-k", "chao moi nguoi trong bang hoi, toi la thanh vien moi", "con l_o_n đ.ị.t b.u.ồ.i", "ai muon vao party danh boss Thien Hoang khong?", "thang đ_k_m may scam tao ha?", ] iterations = 1000 start = time.perf_counter() for i in range(iterations): self.trie_filter.scan_and_censor(benchmark_payloads[i % len(benchmark_payloads)]) avg_latency_ms = ((time.perf_counter() - start) / iterations) * 1000 if was_tracing: tracemalloc.start() print(f"\n[BENCHMARK] Tier 1 Trie Moderation Avg Latency: {avg_latency_ms:.4f}ms (Target: < 0.1ms)") self.assertLess(avg_latency_ms, 0.1) def test_sla_item_snapshot_retrieval_latency(self) -> None: """Measure item snapshot retrieval latency over 1,000 iterations (SLA: < 2.0ms).""" snap = self.item_link_service.create_item_snapshot( item_uuid="bench_sword_001", item_name_key="Loi Dinh Kiem", rarity=4, element=3, quality=20, item_level=95, crafter_name="Loi Than" ) iterations = 1000 start = time.perf_counter() for _ in range(iterations): res = self.item_link_service.query_item_snapshot("bench_sword_001", snap.hmac_signature) assert res.is_valid avg_latency_ms = ((time.perf_counter() - start) / iterations) * 1000 print(f"[BENCHMARK] Item Snapshot Query Avg Latency: {avg_latency_ms:.4f}ms (Target: < 2.0ms)") self.assertLess(avg_latency_ms, 2.0) def test_empirical_edge_case_evasions_documented(self) -> None: """Empirically documents 4 adversarial evasion vectors discovered during stress testing.""" # Evasion 1: ZWNJ (\\u200c), ZWJ (\\u200d), Word Joiner (\\u2060) has_prof_zwnj, _ = self.trie_filter.scan_and_censor("thang d\u200cm may") has_prof_zwj, _ = self.trie_filter.scan_and_censor("thang d\u200dm may") has_prof_wj, _ = self.trie_filter.scan_and_censor("thang d\u2060m may") self.assertFalse(has_prof_zwnj, "Evidence: ZWNJ currently evades Trie filter") self.assertFalse(has_prof_zwj, "Evidence: ZWJ currently evades Trie filter") self.assertFalse(has_prof_wj, "Evidence: Word Joiner currently evades Trie filter") # Evasion 2: Cyrillic homoglyphs look identical to Latin letters but bypass ASCII Trie cyrillic_cac = "thang " + chr(0x441) + chr(0x430) + chr(0x441) + " may" has_prof_cyrillic, _ = self.trie_filter.scan_and_censor(cyrillic_cac) self.assertFalse(has_prof_cyrillic, "Evidence: Cyrillic homoglyphs currently evade Trie filter") # Evasion 3: Separated phone numbers corrupted by leetspeak substitution table res_dot_phone = self.pipeline.process("lien he qua so 098.765.4321") res_space_phone = self.pipeline.process("lien he qua so 098 765 4321") self.assertFalse(res_dot_phone.is_rmt, "Evidence: Dot-separated phone corrupted by leetspeak table") self.assertFalse(res_space_phone.is_rmt, "Evidence: Space-separated phone corrupted by leetspeak table") # Evasion 4: Negative lookahead (?![a-zA-Z0-9]) blocks masking when digit is suffixed has_prof_digit_suffix, _ = self.trie_filter.scan_and_censor("thang dm1 may") self.assertFalse(has_prof_digit_suffix, "Evidence: Digit-suffixed profanity currently evades regex masking") if __name__ == "__main__": unittest.main()