fix(dynamic-rules): apply each rule version over the base config

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(

guard_core/handlers/dynamic_rule_handler.py

@@ -225,14 +225,17 @@ 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:
             self._active_base_snapshot = snapshot
 
     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
 

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(

guard_core/sync/handlers/dynamic_rule_handler.py

@@ -224,14 +224,17 @@ 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:
             self._active_base_snapshot = snapshot
 
     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
 

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

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)

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: