""" Unit tests for Chat Multi-Channel Routing, Multilingual Profanity Filter, RMT/Chợ Đen Surveillance, and Cryptographic Item Linking in FreeExile. """ import hashlib import hmac import re import unittest from server.chat.item_link_service import ItemLinkService from server.chat.moderation import ( AsyncSentinelRMTDetector, ChatModerationPipeline, SynchronousTrieFilter, ) class MultilingualFilterPython: """Python mirror of C++ MultilingualProfanityFilter for unit test verification.""" def __init__(self) -> None: self.banned_words = { "dm", "dkm", "cl", "vcl", "lon", "cac", "dit", "buoi", "fuck", "shit", "bitch", "scam", "cnm", "sb", "nmd" } self.rmt_patterns = [ re.compile(r"(03|05|07|08|09)\d{8}"), re.compile(r"(zalo|telegram|tele|fb\.com)[\s\:\@\.\-_]*\w+", re.IGNORECASE), re.compile(r"(chuyen khoan|atm|momo|ban vang|thu mua ngoc|ban acc)", re.IGNORECASE) ] def normalize(self, text: str) -> str: s = text.lower() subs = {'@': 'a', '0': 'o', '1': 'i', '!': 'i', '3': 'e', '$': 's'} for k, v in subs.items(): s = s.replace(k, v) for ch in ['.', '_', '-', ' ', '/']: s = s.replace(ch, '') return s def filter(self, raw_content: str) -> tuple[bool, bool, int]: normalized = self.normalize(raw_content) has_profanity = any(w in normalized for w in self.banned_words) is_rmt = any(p.search(raw_content) or p.search(normalized) for p in self.rmt_patterns) risk_score = 0 if has_profanity: risk_score += 25 if is_rmt: risk_score += 60 return has_profanity, is_rmt, min(risk_score, 100) class TestChatAndModeration(unittest.TestCase): def setUp(self) -> None: self.legacy_filter = MultilingualFilterPython() self.trie_filter = SynchronousTrieFilter() self.pipeline = ChatModerationPipeline() self.item_link_service = ItemLinkService() self.secret_key = b"FREEEXILE_SECRET_SIGNING_KEY_2026" def test_multilingual_profanity_and_leetspeak_bypass(self) -> None: """Test catching toxic words hidden with dots, leetspeak, or spaces.""" test_cases = [ ("d.m may", True), ("thang d-k-m", True), ("con l_o_n", True), ("what the f.u.c.k", True), ("chao ban nhe", False), ("anh em tap trung danh boss", False) ] for text, expected_profane in test_cases: has_profane, _, _ = self.legacy_filter.filter(text) self.assertEqual(has_profane, expected_profane, f"Failed for: {text}") def test_rmt_and_black_market_detection(self) -> None: """Test identifying illegal real money trading (RMT), phone numbers, and Zalo/Tele handles.""" rmt_cases = [ ("Ban vang gia re lien he zalo 0987654321", True), ("Nhan chuyen khoan atm lay ngoc", True), ("Inbox telegram @trader_pro", True), ("Ai mua kiem cam pm toi trong game", False) ] for text, expected_rmt in rmt_cases: _, is_rmt, score = self.legacy_filter.filter(text) self.assertEqual(is_rmt, expected_rmt, f"Failed RMT test for: {text}") if expected_rmt: self.assertGreaterEqual(score, 60) def test_cryptographic_item_linking_signature(self) -> None: """Test HMAC-SHA256 signature generation and anti-spoofing verification.""" item_uuid = "item_sword_999" item_name = "Thanh Phong Huyen Kiem" rarity = 4 # Than Pham # Generate server HMAC signature message = f"{item_uuid}:{item_name}:{rarity}".encode("utf-8") signature = hmac.new(self.secret_key, message, hashlib.sha256).hexdigest() # Valid verification valid_msg = f"{item_uuid}:{item_name}:{rarity}".encode("utf-8") expected_sig = hmac.new(self.secret_key, valid_msg, hashlib.sha256).hexdigest() self.assertEqual(signature, expected_sig) # Attacker tries to spoof rarity from 1 (Pham) to 5 (Thai Co) spoofed_msg = f"{item_uuid}:{item_name}:5".encode("utf-8") spoofed_sig = hmac.new(self.secret_key, spoofed_msg, hashlib.sha256).hexdigest() self.assertNotEqual(signature, spoofed_sig) def test_synchronous_trie_censor_masking(self) -> None: """Test SynchronousTrieFilter masks vulgarities with *** and preserves harmless text.""" # Profanity masking has_prof, masked = self.trie_filter.scan_and_censor("thang d.m may c_a_c") self.assertTrue(has_prof) self.assertIn("***", masked) self.assertNotIn("d.m", masked) # Vietnamese diacritics has_prof2, masked2 = self.trie_filter.scan_and_censor("thang đ.ị.t may") self.assertTrue(has_prof2) self.assertIn("***", masked2) # Non-profane words should NOT be masked has_prof3, text3 = self.trie_filter.scan_and_censor("chao ban, toi hoc class nay") self.assertFalse(has_prof3) self.assertEqual(text3, "chao ban, toi hoc class nay") def test_pipeline_rmt_and_crypto_auto_mute(self) -> None: """Test ChatModerationPipeline computes risk scores and triggers auto-mute.""" # Phone + payment -> risk >= 60 -> auto-mute res1 = self.pipeline.process("Ban vang chuyen khoan atm 0912345678") self.assertTrue(res1.is_rmt) self.assertTrue(res1.should_auto_mute) self.assertGreaterEqual(res1.risk_score, 60) # Crypto payment -> USDT + vi dien tu -> risk >= 60 res2 = self.pipeline.process("Ban acc nhan usdt qua vi dien tu") self.assertTrue(res2.is_rmt) self.assertTrue(res2.should_auto_mute) self.assertGreaterEqual(res2.risk_score, 60) # Clean game chat res3 = self.pipeline.process("Toi muon ban cay kiem pham cap 50 trong game") self.assertFalse(res3.is_rmt) self.assertFalse(res3.should_auto_mute) self.assertEqual(res3.risk_score, 0) def test_item_link_service_full_workflow(self) -> None: """Test ItemLinkService creates snapshots, verifies HMAC, and detects tampering.""" snap = self.item_link_service.create_item_snapshot( item_uuid="axe_777", item_name_key="Khai Son Phu", rarity=3, element=5, quality=15, item_level=75, crafter_name="Cuong Dao" ) self.assertTrue(self.item_link_service.verify_item_signature("axe_777", "Khai Son Phu", 3, snap.hmac_signature)) # Tampered rarity fails self.assertFalse(self.item_link_service.verify_item_signature("axe_777", "Khai Son Phu", 5, snap.hmac_signature)) if __name__ == "__main__": unittest.main()