""" Tier 5 White-Box Adversarial Coverage Hardening Suite. Covers extreme edge cases, race conditions, division-by-zero, SQL injection, mid-cast state mutations, and PoE2 anti-exploit quarantine security gates. """ from __future__ import annotations import math import threading import time import pytest from server.party.party_types import LootAllocationMode, PartyMemberState, PartyRole from server.party.party_scaling_calculator import PartyScalingCalculator from server.party.party_service import PartyService from server.social.social_types import GuildMemberPresenceDTO, PresenceStatus from server.social.social_repository import SocialRepository from server.social.social_service import SocialService from server.social.guild_roster_service import GuildRosterService from server.world.town_portal_types import InterruptReason, TargetType from server.world.town_portal_engine import TownPortalEngine from server.world.teleport_service import TeleportService class TestTier5AdversarialHardening: """Tier 5 Adversarial and Boundary Stress Test Suite.""" def test_party_scaling_extreme_level_gap_and_zero_exp(self) -> None: """Stress-tests EXP distribution with level 1 vs level 100 disparity and zero base exp.""" p_low = PartyMemberState("p_low", "LowLevel", level=1, zone_id="z1", instance_id="i1") p_high = PartyMemberState("p_high", "HighLevel", level=100, zone_id="z1", instance_id="i1") # Zero and negative base exp protection (division-by-zero and empty pool) assert PartyScalingCalculator.calculate_party_exp(0, [p_low, p_high]) == {} assert PartyScalingCalculator.calculate_party_exp(-500, [p_low, p_high]) == {} # Extreme level gap scaling base_exp = 1000 shares = PartyScalingCalculator.calculate_party_exp(base_exp, [p_low, p_high]) assert shares["p_low"] >= 0 assert shares["p_high"] > shares["p_low"] assert sum(shares.values()) <= int(base_exp * 1.30) # Level gap penalty with monster level 100 (player 1 vs monster 100) gap_shares = PartyScalingCalculator.calculate_party_exp(base_exp, [p_low, p_high], monster_level=100) assert gap_shares["p_low"] <= shares["p_low"] assert gap_shares["p_high"] == shares["p_high"] def test_party_scaling_empty_party_and_negative_inputs(self) -> None: """Verifies boundary edge cases in party scaling math.""" assert PartyScalingCalculator.calculate_party_exp(1000, []) == {} assert PartyScalingCalculator.calculate_loot_multipliers(0) == (1.0, 1.0, 1.0) assert PartyScalingCalculator.calculate_loot_multipliers(-10) == (1.0, 1.0, 1.0) # Zero or negative party count in monster scaling s0 = PartyScalingCalculator.calculate_monster_scaling(base_hp=1000, rank="BOSS", party_count=0) assert s0["scaled_hp"] == 1000 assert s0["hp_mult"] == 1.0 s_neg = PartyScalingCalculator.calculate_monster_scaling(base_hp=2000, rank="RARE", party_count=-5) assert s_neg["scaled_hp"] == 2000 assert s_neg["hp_mult"] == 1.0 def test_party_service_rapid_lifecycle_concurrency(self) -> None: """Exercises thread-safe RLock concurrency with rapid invites, leaves, and promotions.""" svc = PartyService() errors: list[Exception] = [] def worker(worker_id: int) -> None: try: pid = f"player_{worker_id}" p = svc.create_party(pid, f"Name_{worker_id}", 10) party_id = p.party_id target_id = f"guest_{worker_id}" svc.invite_player(party_id, pid, target_id) svc.accept_invite(party_id, target_id, f"Guest_{worker_id}", 10) svc.promote_leader(party_id, pid, target_id) svc.leave_party(pid) svc.leave_party(target_id) except Exception as exc: # pylint: disable=broad-except errors.append(exc) threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrency errors occurred: {errors}" assert len(svc.parties) == 0 assert len(svc.player_party_map) == 0 def test_party_disband_with_pending_invites_cleanup(self) -> None: """Asserts disbanding party permanently invalidates all outstanding pending invitations.""" svc = PartyService() p = svc.create_party("lead", "Leader", 50) party_id = p.party_id svc.invite_player(party_id, "lead", "invitee_1") svc.invite_player(party_id, "lead", "invitee_2") assert len(svc.pending_invites.get(party_id, {})) == 2 # Leader disbands assert svc.disband_party(party_id, "lead") is True assert party_id not in svc.parties assert party_id not in svc.pending_invites # Former invitee cannot accept disbanded phantom party assert svc.accept_invite(party_id, "invitee_1", "Invitee 1", 10) is False def test_social_sql_injection_and_unicode_in_friend_note(self) -> None: """Tests SQL injection defense and 64-char cap in friend notes.""" repo = SocialRepository(db_path=":memory:") repo.add_friend_request("alice", "bob") repo.accept_friend_request("bob", "alice") # SQL Injection attempt in note sqli_payload = "'; DROP TABLE player_friendships; --" assert repo.set_friend_note("alice", "bob", sqli_payload) is True assert repo.get_friend_note("alice", "bob") == sqli_payload # Table should remain intact assert repo.is_friend("alice", "bob") is True # Unicode & 64-character limit truncation long_unicode_note = "⚔️ Cổ Võ Hoang Vực Bất Diệt Huyết Thạch Lưu Đày " * 5 assert repo.set_friend_note("alice", "bob", long_unicode_note) is True retrieved = repo.get_friend_note("alice", "bob") assert len(retrieved) <= 64 assert retrieved == long_unicode_note[:64] def test_social_self_interaction_adversarial(self) -> None: """Ensures players cannot invite, block, or set notes on themselves.""" repo = SocialRepository(db_path=":memory:") service = SocialService(repository=repo) assert repo.add_friend_request("self_p", "self_p") is False assert repo.block_player("self_p", "self_p") is False assert service.send_friend_request("self_p", "self_p") is False assert service.set_friend_note("self_p", "self_p", "myself") is False assert repo.is_blocked("self_p", "self_p") is False def test_social_rapid_block_unblock_cycle_and_mutual_block_invariants(self) -> None: """Verifies mutual block invariants and friendship dissolution upon blocking.""" repo = SocialRepository(db_path=":memory:") repo.add_friend_request("u1", "u2") repo.accept_friend_request("u2", "u1") assert repo.is_friend("u1", "u2") is True # U1 blocks U2 -> dissolves friendship repo.block_player("u1", "u2") assert repo.is_blocked("u1", "u2") is True assert repo.is_blocked("u2", "u1") is True assert repo.is_friend("u1", "u2") is False # U2 also blocks U1 (mutual block) repo.block_player("u2", "u1") assert repo.is_blocked("u1", "u2") is True # U1 unblocks U2, but U2 STILL blocks U1 -> block must persist assert repo.unblock_player("u1", "u2") is True assert repo.is_blocked("u1", "u2") is True assert repo.is_blocked("u2", "u1") is True # U2 also unblocks U1 -> now fully unblocked assert repo.unblock_player("u2", "u1") is True assert repo.is_blocked("u1", "u2") is False # Friendship must NOT be magically restored assert repo.is_friend("u1", "u2") is False def test_guild_roster_restricted_zones_and_offline_filtering(self) -> None: """Verifies teleport targetability filtering across restricted zones and presence states.""" roster_svc = GuildRosterService() roster_svc.register_member("g1", GuildMemberPresenceDTO("m1", "OnlineWild", role="MEMBER", level=20, presence_status=PresenceStatus.ONLINE, zone_id="zone_tang_kiem_nhai")) roster_svc.register_member("g1", GuildMemberPresenceDTO("m2", "InHideout", role="MEMBER", level=20, presence_status=PresenceStatus.ONLINE, zone_id="zone_player_hideout")) roster_svc.register_member("g1", GuildMemberPresenceDTO("m3", "InSecret", role="MEMBER", level=20, presence_status=PresenceStatus.ONLINE, zone_id="zone_secret_chamber")) roster_svc.register_member("g1", GuildMemberPresenceDTO("m4", "InBoss", role="MEMBER", level=20, presence_status=PresenceStatus.ONLINE, zone_id="zone_boss_arena")) roster_svc.register_member("g1", GuildMemberPresenceDTO("m5", "OfflineGuy", role="MEMBER", level=20, presence_status=PresenceStatus.OFFLINE, zone_id="zone_tang_kiem_nhai")) roster = roster_svc.get_guild_roster_with_presence("g1") target_map = {m.player_id: m.is_teleport_targetable for m in roster} assert target_map["m1"] is True assert target_map["m2"] is False assert target_map["m3"] is False assert target_map["m4"] is False assert target_map["m5"] is False def test_teleport_frame_zero_interrupts_and_subthreshold_jitter(self) -> None: """Verifies instant interrupts on frame 0 and deadzone joystick tolerance.""" party_svc = PartyService() party_svc.create_party("caster", "Caster", 10) p = party_svc.get_player_party("caster") assert p is not None party_svc.invite_player(p.party_id, "caster", "target") party_svc.accept_invite(p.party_id, "target", "Target", 10) tp_svc = TeleportService(party_service=party_svc) tp_svc.set_zone_state("zone_wild", is_active=True, player_count=1, max_capacity=6) # 1. Damage at frame 0 (damage = 0.001) res = tp_svc.start_teleport_channel("caster", TargetType.PARTY_MEMBER, "target", "zone_wild") assert res.success is True int_reason = tp_svc.on_player_damaged("caster", damage=0.001) assert int_reason == InterruptReason.INTERRUPT_DAMAGE assert "caster" not in tp_svc.active_channels # 2. Sub-threshold movement jitter tolerance (mag <= 0.05) res2 = tp_svc.start_teleport_channel("caster", TargetType.PARTY_MEMBER, "target", "zone_wild") assert res2.success is True assert tp_svc.on_player_moved("caster", mag=0.049) is None assert "caster" in tp_svc.active_channels # 3. Movement displacement interrupt (mag > 0.05) assert tp_svc.on_player_moved("caster", mag=0.051) == InterruptReason.INTERRUPT_MOVEMENT assert "caster" not in tp_svc.active_channels def test_teleport_mid_channel_destination_quarantine_mutation(self) -> None: """Tests race condition where destination zone triggers boss encounter mid-channel.""" party_svc = PartyService() party_svc.create_party("alice", "Alice", 10) p = party_svc.get_player_party("alice") assert p is not None party_svc.invite_player(p.party_id, "alice", "bob") party_svc.accept_invite(p.party_id, "bob", "Bob", 10) tp_svc = TeleportService(party_service=party_svc) tp_svc.set_zone_state("zone_dungeon", is_boss_active=False, is_active=True) res = tp_svc.start_teleport_channel("alice", TargetType.PARTY_MEMBER, "bob", "zone_dungeon") assert res.success is True # Mid-channel mutation: Boss spawns in destination zone tp_svc.set_zone_state("zone_dungeon", is_boss_active=True, is_active=True) # Completion post-flight check must catch boss quarantine and reject ok, reason = tp_svc.complete_channel("alice", elapsed_time=3.5) assert ok is False assert "GATE_5_QUARANTINE_BOSS" in reason assert "alice" not in tp_svc.active_channels def test_town_portal_double_quarantine_and_boss_lockout(self) -> None: """Verifies quarantine priority and unlock transitions for town portals.""" engine = TownPortalEngine() engine.set_encounter_state("inst_1", boss_active=True, secret_active=True) # Blocked when both active is_locked, err = engine.check_quarantine_lock("zone_dungeon", "inst_1") assert is_locked is True assert "QUARANTINE_BOSS_ENGAGED" in err res = engine.request_town_portal("p1", "zone_dungeon", "inst_1") assert res.success is False # Defeat boss, but secret chamber still active engine.set_encounter_state("inst_1", boss_active=False, secret_active=True) is_locked, err = engine.check_quarantine_lock("zone_dungeon", "inst_1") assert is_locked is True assert "QUARANTINE_SECRET_CHAMBER" in err # Clear secret chamber -> unlocked engine.set_encounter_state("inst_1", boss_active=False, secret_active=False) assert engine.check_quarantine_lock("zone_dungeon", "inst_1") == (False, "UNLOCKED") res_ok = engine.request_town_portal("p1", "zone_dungeon", "inst_1") assert res_ok.success is True assert res_ok.portal_id is not None def test_map_device_zero_charge_reentry_and_depletion_boundary(self) -> None: """Tests map device exhaustion, death consumption, and zero-charge re-entry lockout.""" engine = TownPortalEngine() engine.register_astral_map_device("map_inst", initial_charges=0) ok, msg, rem = engine.enter_map_portal("map_inst") assert ok is False assert "MAP_SEALED" in msg assert rem == 0 # Register 2 charges engine.register_astral_map_device("map_inst_2", initial_charges=2) ok1, _, rem1 = engine.enter_map_portal("map_inst_2") assert ok1 is True assert rem1 == 1 # Death in map consumes 1 charge death_ok, death_msg = engine.on_player_death("p1", "zone_map", "map_inst_2", is_map_instance=True) assert death_ok is True assert "MAP_DEATH_CONSUMED_PORTAL" in death_msg assert engine.map_charges["map_inst_2"] == 0 # Attempting to re-enter sealed map ok2, msg2, rem2 = engine.enter_map_portal("map_inst_2") assert ok2 is False assert "MAP_SEALED" in msg2 assert rem2 == 0 def test_town_portal_unauthorized_party_hijack_prevention(self) -> None: """Prevents non-party members from traversing personal town portals.""" engine = TownPortalEngine() party_svc = PartyService() party_svc.create_party("owner", "Owner", 50) p = party_svc.get_player_party("owner") assert p is not None party_svc.invite_player(p.party_id, "owner", "ally") party_svc.accept_invite(p.party_id, "ally", "Ally", 50) # Stranger creates their own party party_svc.create_party("stranger", "Stranger", 50) res = engine.request_town_portal("owner", "zone_tang_kiem_nhai", "inst_1", wx=10.0, wy=20.0) assert res.success is True portal_id = res.portal_id assert portal_id is not None # Owner traversal allowed ok_owner, _, _ = engine.traverse_town_portal("owner", portal_id, party_svc) assert ok_owner is True # Ally in same party allowed ok_ally, _, _ = engine.traverse_town_portal("ally", portal_id, party_svc) assert ok_ally is True # Stranger rejected ok_stranger, reason, _ = engine.traverse_town_portal("stranger", portal_id, party_svc) assert ok_stranger is False assert "PORTAL_UNAUTHORIZED" in reason