diff --git a/guard_core/handlers/_dynamic_rule_application.py b/guard_core/handlers/_dynamic_rule_application.py index 7fa9d73f..85a39ad6 100644 --- a/guard_core/handlers/_dynamic_rule_application.py +++ b/guard_core/handlers/_dynamic_rule_application.py @@ -159,6 +159,8 @@ async def _apply_user_agent_rules(self, user_agents: list[str]) -> None: async def _apply_pattern_rules(self, patterns: list[str]) -> None: from guard_core.handlers.suspatterns_handler import sus_patterns_handler + known = set(sus_patterns_handler.patterns) + patterns = [p for p in patterns if p not in known] added = [p for p in patterns if await sus_patterns_handler.add_pattern(p)] if added: self.logger.info( diff --git a/guard_core/handlers/dynamic_rule_handler.py b/guard_core/handlers/dynamic_rule_handler.py index 57b872b9..e504ab5c 100644 --- a/guard_core/handlers/dynamic_rule_handler.py +++ b/guard_core/handlers/dynamic_rule_handler.py @@ -225,7 +225,7 @@ def _snapshot_config(self) -> dict[str, object]: def _restore_config(self, snapshot: dict[str, object]) -> None: for field, value in snapshot.items(): - self.config._set_prevalidated(field, value) + self.config._set_prevalidated(field, deepcopy(value)) def _capture_active_base_snapshot(self, snapshot: dict[str, object]) -> None: if self.current_rules is None and self._active_base_snapshot is None: @@ -233,6 +233,9 @@ def _capture_active_base_snapshot(self, snapshot: dict[str, object]) -> None: async def _apply_rules(self, rules: DynamicRules) -> None: async with self._lock: + previous = self._snapshot_config() + if self._active_base_snapshot is not None: + self._restore_config(self._active_base_snapshot) snapshot = self._snapshot_config() try: await self._apply_ip_rules(rules) @@ -248,7 +251,7 @@ async def _apply_rules(self, rules: DynamicRules) -> None: await self._activate_emergency_mode(rules.emergency_whitelist) except Exception as e: - self._restore_config(snapshot) + self._restore_config(previous) self.logger.error(f"Failed to apply dynamic rules: {e}") raise diff --git a/guard_core/sync/handlers/_dynamic_rule_application.py b/guard_core/sync/handlers/_dynamic_rule_application.py index ce680945..03fbef82 100644 --- a/guard_core/sync/handlers/_dynamic_rule_application.py +++ b/guard_core/sync/handlers/_dynamic_rule_application.py @@ -157,6 +157,8 @@ def _apply_user_agent_rules(self, user_agents: list[str]) -> None: def _apply_pattern_rules(self, patterns: list[str]) -> None: from guard_core.sync.handlers.suspatterns_handler import sus_patterns_handler + known = set(sus_patterns_handler.patterns) + patterns = [p for p in patterns if p not in known] added = [p for p in patterns if sus_patterns_handler.add_pattern(p)] if added: self.logger.info( diff --git a/guard_core/sync/handlers/dynamic_rule_handler.py b/guard_core/sync/handlers/dynamic_rule_handler.py index 09e9d856..6b63f6e3 100644 --- a/guard_core/sync/handlers/dynamic_rule_handler.py +++ b/guard_core/sync/handlers/dynamic_rule_handler.py @@ -224,7 +224,7 @@ def _snapshot_config(self) -> dict[str, object]: def _restore_config(self, snapshot: dict[str, object]) -> None: for field, value in snapshot.items(): - self.config._set_prevalidated(field, value) + self.config._set_prevalidated(field, deepcopy(value)) def _capture_active_base_snapshot(self, snapshot: dict[str, object]) -> None: if self.current_rules is None and self._active_base_snapshot is None: @@ -232,6 +232,9 @@ def _capture_active_base_snapshot(self, snapshot: dict[str, object]) -> None: def _apply_rules(self, rules: DynamicRules) -> None: with self._lock: + previous = self._snapshot_config() + if self._active_base_snapshot is not None: + self._restore_config(self._active_base_snapshot) snapshot = self._snapshot_config() try: self._apply_ip_rules(rules) @@ -247,7 +250,7 @@ def _apply_rules(self, rules: DynamicRules) -> None: self._activate_emergency_mode(rules.emergency_whitelist) except Exception as e: - self._restore_config(snapshot) + self._restore_config(previous) self.logger.error(f"Failed to apply dynamic rules: {e}") raise diff --git a/tests/test_dynamic_rule_atomicity.py b/tests/test_dynamic_rule_atomicity.py index cfca6044..834b30ad 100644 --- a/tests/test_dynamic_rule_atomicity.py +++ b/tests/test_dynamic_rule_atomicity.py @@ -86,7 +86,8 @@ async def slow_blocking(rules: DynamicRules) -> None: assert len(observed) == 2 assert observed[0] in ([], ["AA"], ["BB"]) assert observed[1] in ([], ["AA"], ["BB"]) - assert observed[0] != observed[1] + assert observed == [[], []] + assert config.blocked_countries in (frozenset({"AA"}), frozenset({"BB"})) @pytest.mark.asyncio diff --git a/tests/test_dynamic_rule_overlay.py b/tests/test_dynamic_rule_overlay.py new file mode 100644 index 00000000..27ea3cd6 --- /dev/null +++ b/tests/test_dynamic_rule_overlay.py @@ -0,0 +1,209 @@ +from collections.abc import Generator +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from guard_core.handlers.dynamic_rule_handler import DynamicRuleManager +from guard_core.models import DynamicRules, SecurityConfig +from guard_core.protocols.geo_ip_protocol import GeoIPHandler + + +def _rules(version: int = 1, rule_id: str = "r1", **kwargs: object) -> DynamicRules: + base = { + "rule_id": rule_id, + "version": version, + "timestamp": datetime.now(timezone.utc), + } + base.update(kwargs) + return DynamicRules(**base) + + +@pytest.fixture(autouse=True) +def reset_singleton() -> Generator[None, None, None]: + DynamicRuleManager._instance = None + yield + DynamicRuleManager._instance = None + + +async def _apply(manager: DynamicRuleManager, rules: DynamicRules) -> None: + await manager._apply_rules(rules) + manager.current_rules = rules + + +@pytest.mark.asyncio +async def test_emergency_halving_does_not_compound_across_versions() -> None: + config = SecurityConfig(auto_ban_threshold=16) + manager = DynamicRuleManager(config) + + seen = [] + for version in (1, 2, 3): + await _apply( + manager, _rules(version, emergency_mode=True, emergency_whitelist=[]) + ) + seen.append(config.auto_ban_threshold) + + assert seen == [8, 8, 8] + assert manager._active_base_snapshot is not None + assert manager._active_base_snapshot["auto_ban_threshold"] == 16 + + +@pytest.mark.asyncio +async def test_collection_values_do_not_accumulate_across_versions() -> None: + config = SecurityConfig( + geo_ip_handler=MagicMock(spec=GeoIPHandler), + blocked_user_agents=["base-bot"], + block_cloud_providers=frozenset({"AWS"}), + ) + manager = DynamicRuleManager(config) + + await _apply( + manager, + _rules( + 1, + blocked_user_agents=["bot-a"], + blocked_cloud_providers={"GCP"}, + blocked_countries=["AA"], + endpoint_rate_limits={"/a": (1, 60)}, + ), + ) + await _apply( + manager, + _rules( + 2, + blocked_user_agents=["bot-b"], + blocked_cloud_providers={"Azure"}, + blocked_countries=["BB"], + endpoint_rate_limits={"/b": (2, 60)}, + ), + ) + + assert config.blocked_user_agents == ["bot-b"] + assert config.block_cloud_providers == frozenset({"Azure"}) + assert config.blocked_countries == frozenset({"BB"}) + assert config.endpoint_rate_limits == {"/b": (2, 60)} + + +@pytest.mark.asyncio +async def test_new_version_drops_change_from_previous_version() -> None: + config = SecurityConfig( + rate_limit=100, enable_ip_banning=True, auto_ban_threshold=5 + ) + manager = DynamicRuleManager(config) + + await _apply( + manager, + _rules(1, global_rate_limit=7, enable_ip_banning=False, auto_ban_threshold=2), + ) + assert (config.rate_limit, config.enable_ip_banning) == (7, False) + + await _apply(manager, _rules(2, global_rate_limit=9)) + + assert config.rate_limit == 9 + assert config.enable_ip_banning is True + assert config.auto_ban_threshold == 5 + + +@pytest.mark.asyncio +async def test_different_rule_id_overlays_the_base() -> None: + config = SecurityConfig(rate_limit=100, auto_ban_threshold=16) + manager = DynamicRuleManager(config) + + await _apply(manager, _rules(1, "r1", emergency_mode=True, emergency_whitelist=[])) + await _apply(manager, _rules(1, "r2", global_rate_limit=3)) + + assert config.rate_limit == 3 + assert config.auto_ban_threshold == 16 + assert config.emergency_mode is False + + +@pytest.mark.asyncio +async def test_failed_apply_rolls_back_to_the_active_version() -> None: + config = SecurityConfig(rate_limit=100, auto_ban_threshold=16) + manager = DynamicRuleManager(config) + await _apply(manager, _rules(1, global_rate_limit=7, auto_ban_threshold=4)) + + with patch.object( + manager, "_apply_feature_toggles", AsyncMock(side_effect=RuntimeError("boom")) + ): + with pytest.raises(RuntimeError, match="boom"): + await manager._apply_rules(_rules(2, global_rate_limit=9)) + + assert config.rate_limit == 7 + assert config.auto_ban_threshold == 4 + assert manager._active_base_snapshot is not None + assert manager._active_base_snapshot["rate_limit"] == 100 + + +@pytest.mark.asyncio +async def test_expiry_restores_the_true_base_after_several_versions() -> None: + config = SecurityConfig(auto_ban_threshold=16) + manager = DynamicRuleManager(config) + await _apply(manager, _rules(1, emergency_mode=True, emergency_whitelist=[])) + await _apply( + manager, + _rules( + 2, + emergency_mode=True, + emergency_whitelist=[], + expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), + ), + ) + assert config.auto_ban_threshold == 8 + + await manager._check_rule_expiry() + + assert config.auto_ban_threshold == 16 + assert config.emergency_mode is False + assert manager.current_rules is None + assert manager._active_base_snapshot is None + + +@pytest.mark.asyncio +async def test_new_version_after_hydration_overlays_the_base() -> None: + config = SecurityConfig( + enable_dynamic_rules=True, + enable_agent=True, + agent_api_key="key", + auto_ban_threshold=16, + ) + manager = DynamicRuleManager(config) + hydrated = _rules(1, emergency_mode=True, emergency_whitelist=[]) + agent = AsyncMock() + agent.get_dynamic_rules.return_value = _rules( + 2, emergency_mode=True, emergency_whitelist=[] + ) + with patch.object( + manager, "_load_last_known_rules", AsyncMock(return_value=hydrated) + ): + with patch.object( + DynamicRuleManager, "_rule_update_loop", new_callable=AsyncMock + ): + await manager.initialize_agent(agent) + assert config.auto_ban_threshold == 8 + + await manager.update_rules() + + assert manager.current_rules is not None + assert manager.current_rules.version == 2 + assert config.auto_ban_threshold == 8 + await manager.stop() + + +@pytest.mark.asyncio +async def test_suspicious_patterns_are_not_duplicated_across_versions() -> None: + from guard_core.handlers.suspatterns_handler import sus_patterns_handler + + config = SecurityConfig() + manager = DynamicRuleManager(config) + pattern = "overlay-unique-marker-[0-9]{3}" + sus_patterns_handler.patterns[:] = [ + p for p in sus_patterns_handler.patterns if p != pattern + ] + try: + for version in (1, 2, 3): + await _apply(manager, _rules(version, suspicious_patterns=[pattern])) + + assert sus_patterns_handler.patterns.count(pattern) == 1 + finally: + await sus_patterns_handler.remove_pattern(pattern) diff --git a/tests/test_sync/test_dynamic_rule_atomicity.py b/tests/test_sync/test_dynamic_rule_atomicity.py index 972d8f45..7e6e6e75 100644 --- a/tests/test_sync/test_dynamic_rule_atomicity.py +++ b/tests/test_sync/test_dynamic_rule_atomicity.py @@ -85,7 +85,8 @@ def slow(rules: DynamicRules) -> None: t.join() assert len(observed) == 2 - assert observed[0] != observed[1] + assert observed == [[], []] + assert config.blocked_countries in (frozenset({"AA"}), frozenset({"BB"})) def test_rollback_restores_all_snapshot_fields() -> None: diff --git a/tests/test_sync/test_dynamic_rule_overlay.py b/tests/test_sync/test_dynamic_rule_overlay.py new file mode 100644 index 00000000..233ea773 --- /dev/null +++ b/tests/test_sync/test_dynamic_rule_overlay.py @@ -0,0 +1,197 @@ +from collections.abc import Generator +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from guard_core.models import DynamicRules, SecurityConfig +from guard_core.sync.handlers.dynamic_rule_handler import DynamicRuleManager +from guard_core.sync.protocols.geo_ip_protocol import SyncGeoIPHandler + + +def _rules(version: int = 1, rule_id: str = "r1", **kwargs: object) -> DynamicRules: + base = { + "rule_id": rule_id, + "version": version, + "timestamp": datetime.now(timezone.utc), + } + base.update(kwargs) + return DynamicRules(**base) + + +@pytest.fixture(autouse=True) +def reset_singleton() -> Generator[None, None, None]: + DynamicRuleManager._instance = None + yield + DynamicRuleManager._instance = None + + +def _apply(manager: DynamicRuleManager, rules: DynamicRules) -> None: + manager._apply_rules(rules) + manager.current_rules = rules + + +def test_emergency_halving_does_not_compound_across_versions() -> None: + config = SecurityConfig(auto_ban_threshold=16) + manager = DynamicRuleManager(config) + + seen = [] + for version in (1, 2, 3): + _apply(manager, _rules(version, emergency_mode=True, emergency_whitelist=[])) + seen.append(config.auto_ban_threshold) + + assert seen == [8, 8, 8] + assert manager._active_base_snapshot is not None + assert manager._active_base_snapshot["auto_ban_threshold"] == 16 + + +def test_collection_values_do_not_accumulate_across_versions() -> None: + config = SecurityConfig( + geo_ip_handler=MagicMock(spec=SyncGeoIPHandler), + blocked_user_agents=["base-bot"], + block_cloud_providers=frozenset({"AWS"}), + ) + manager = DynamicRuleManager(config) + + _apply( + manager, + _rules( + 1, + blocked_user_agents=["bot-a"], + blocked_cloud_providers={"GCP"}, + blocked_countries=["AA"], + endpoint_rate_limits={"/a": (1, 60)}, + ), + ) + _apply( + manager, + _rules( + 2, + blocked_user_agents=["bot-b"], + blocked_cloud_providers={"Azure"}, + blocked_countries=["BB"], + endpoint_rate_limits={"/b": (2, 60)}, + ), + ) + + assert config.blocked_user_agents == ["bot-b"] + assert config.block_cloud_providers == frozenset({"Azure"}) + assert config.blocked_countries == frozenset({"BB"}) + assert config.endpoint_rate_limits == {"/b": (2, 60)} + + +def test_new_version_drops_change_from_previous_version() -> None: + config = SecurityConfig( + rate_limit=100, enable_ip_banning=True, auto_ban_threshold=5 + ) + manager = DynamicRuleManager(config) + + _apply( + manager, + _rules(1, global_rate_limit=7, enable_ip_banning=False, auto_ban_threshold=2), + ) + assert (config.rate_limit, config.enable_ip_banning) == (7, False) + + _apply(manager, _rules(2, global_rate_limit=9)) + + assert config.rate_limit == 9 + assert config.enable_ip_banning is True + assert config.auto_ban_threshold == 5 + + +def test_different_rule_id_overlays_the_base() -> None: + config = SecurityConfig(rate_limit=100, auto_ban_threshold=16) + manager = DynamicRuleManager(config) + + _apply(manager, _rules(1, "r1", emergency_mode=True, emergency_whitelist=[])) + _apply(manager, _rules(1, "r2", global_rate_limit=3)) + + assert config.rate_limit == 3 + assert config.auto_ban_threshold == 16 + assert config.emergency_mode is False + + +def test_failed_apply_rolls_back_to_the_active_version() -> None: + config = SecurityConfig(rate_limit=100, auto_ban_threshold=16) + manager = DynamicRuleManager(config) + _apply(manager, _rules(1, global_rate_limit=7, auto_ban_threshold=4)) + + with patch.object( + manager, "_apply_feature_toggles", MagicMock(side_effect=RuntimeError("boom")) + ): + with pytest.raises(RuntimeError, match="boom"): + manager._apply_rules(_rules(2, global_rate_limit=9)) + + assert config.rate_limit == 7 + assert config.auto_ban_threshold == 4 + assert manager._active_base_snapshot is not None + assert manager._active_base_snapshot["rate_limit"] == 100 + + +def test_expiry_restores_the_true_base_after_several_versions() -> None: + config = SecurityConfig(auto_ban_threshold=16) + manager = DynamicRuleManager(config) + _apply(manager, _rules(1, emergency_mode=True, emergency_whitelist=[])) + _apply( + manager, + _rules( + 2, + emergency_mode=True, + emergency_whitelist=[], + expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), + ), + ) + assert config.auto_ban_threshold == 8 + + manager._check_rule_expiry() + + assert config.auto_ban_threshold == 16 + assert config.emergency_mode is False + assert manager.current_rules is None + assert manager._active_base_snapshot is None + + +def test_new_version_after_hydration_overlays_the_base() -> None: + config = SecurityConfig( + enable_dynamic_rules=True, + enable_agent=True, + agent_api_key="key", + auto_ban_threshold=16, + ) + manager = DynamicRuleManager(config) + hydrated = _rules(1, emergency_mode=True, emergency_whitelist=[]) + agent = MagicMock() + agent.get_dynamic_rules.return_value = _rules( + 2, emergency_mode=True, emergency_whitelist=[] + ) + with patch.object( + manager, "_load_last_known_rules", MagicMock(return_value=hydrated) + ): + with patch.object(DynamicRuleManager, "_rule_update_loop"): + manager.initialize_agent(agent) + assert config.auto_ban_threshold == 8 + + manager.update_rules() + + assert manager.current_rules is not None + assert manager.current_rules.version == 2 + assert config.auto_ban_threshold == 8 + manager.stop() + + +def test_suspicious_patterns_are_not_duplicated_across_versions() -> None: + from guard_core.sync.handlers.suspatterns_handler import sus_patterns_handler + + config = SecurityConfig() + manager = DynamicRuleManager(config) + pattern = "overlay-unique-marker-[0-9]{3}" + sus_patterns_handler.patterns[:] = [ + p for p in sus_patterns_handler.patterns if p != pattern + ] + try: + for version in (1, 2, 3): + _apply(manager, _rules(version, suspicious_patterns=[pattern])) + + assert sus_patterns_handler.patterns.count(pattern) == 1 + finally: + sus_patterns_handler.remove_pattern(pattern)