Skip to content

Behavior Manager

guard_core.handlers.behavior_handler.BehaviorTracker(config)

Bases: BehaviorResponsePatternMixin, BehaviorActionDispatchMixin

Source code in guard_core/handlers/behavior_handler.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(self, config: SecurityConfig):
    self.config = config
    self.logger = logging.getLogger("guard_core.handlers.behavior")
    self.usage_counts: dict[str, dict[str, list[float]]] = defaultdict(
        lambda: defaultdict(list)
    )
    self.return_patterns: dict[str, dict[str, list[float]]] = defaultdict(
        lambda: defaultdict(list)
    )
    self.redis_handler: Any | None = None
    self.agent_handler: Any | None = None
    self._body_unavailable_log_cache: TTLCache[str, bool] = TTLCache(
        maxsize=1000, ttl=300
    )
    self._lock = threading.Lock()

agent_handler = None instance-attribute

config = config instance-attribute

logger = logging.getLogger('guard_core.handlers.behavior') instance-attribute

redis_handler = None instance-attribute

return_patterns = defaultdict(lambda: defaultdict(list)) instance-attribute

usage_counts = defaultdict(lambda: defaultdict(list)) instance-attribute

get_recent_event_count(ip, window_seconds)

Source code in guard_core/handlers/behavior_handler.py 
81
82
83
84
85
86
87
88
89
90
91
def get_recent_event_count(self, ip: str, window_seconds: int) -> int:
    if not ip:
        return 0
    cutoff = time.time() - window_seconds
    count = 0
    with self._lock:
        for endpoint_bucket in self.usage_counts.values():
            for ts in endpoint_bucket.get(ip, []):
                if ts >= cutoff:
                    count += 1
    return count

initialize_agent(agent_handler) async

Source code in guard_core/handlers/behavior_handler.py 
78
79
async def initialize_agent(self, agent_handler: Any) -> None:
    self.agent_handler = agent_handler

initialize_redis(redis_handler) async

Source code in guard_core/handlers/behavior_handler.py 
75
76
async def initialize_redis(self, redis_handler: Any) -> None:
    self.redis_handler = redis_handler

track_endpoint_usage(endpoint_id, client_ip, rule) async

Source code in guard_core/handlers/behavior_handler.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def track_endpoint_usage(
    self, endpoint_id: str, client_ip: str, rule: BehaviorRule
) -> bool:
    current_time = time.time()
    window_start = current_time - rule.window

    if self.redis_handler:
        key = (
            f"behavior:usage:{_hash_identity_segment(endpoint_id)}:"
            f"{_hash_identity_segment(client_ip)}"
        )

        valid_count: int = await self.redis_handler.record_sliding_window_hit(
            "behavior_usage", key, current_time, window_start, rule.window
        )

        return valid_count > rule.threshold

    with self._lock:
        bucket = _lru_pop_or_create(
            self.usage_counts,
            endpoint_id,
            _MAX_TRACKED_ENDPOINTS,
            lambda: defaultdict(list),
        )
        self.usage_counts[endpoint_id] = bucket
        timestamps = _lru_pop_or_create(
            bucket, client_ip, _MAX_TRACKED_CLIENTS_PER_ENDPOINT, list
        )

        timestamps[:] = [ts for ts in timestamps if ts >= window_start]

        timestamps.append(current_time)
        bucket[client_ip] = timestamps

        return len(timestamps) > rule.threshold

track_return_pattern(endpoint_id, client_ip, response, rule, effective_threshold=None) async

Source code in guard_core/handlers/behavior_handler.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def track_return_pattern(
    self,
    endpoint_id: str,
    client_ip: str,
    response: GuardResponse,
    rule: BehaviorRule,
    effective_threshold: int | None = None,
) -> bool:
    if not rule.pattern:
        return False

    threshold = (
        effective_threshold if effective_threshold is not None else rule.threshold
    )
    current_time = time.time()
    window_start = current_time - rule.window

    pattern_matched = await self._check_response_pattern(response, rule.pattern)

    if not pattern_matched:
        return False

    if self.redis_handler:
        key = (
            f"behavior:return:{_hash_identity_segment(endpoint_id)}:"
            f"{_hash_identity_segment(client_ip)}:"
            f"{_hash_identity_segment(rule.pattern)}"
        )

        valid_count: int = await self.redis_handler.record_sliding_window_hit(
            "behavior_returns", key, current_time, window_start, rule.window
        )

        return valid_count > threshold

    pattern_key = f"{endpoint_id}:{rule.pattern}"
    with self._lock:
        bucket = _lru_pop_or_create(
            self.return_patterns,
            pattern_key,
            _MAX_TRACKED_ENDPOINTS,
            lambda: defaultdict(list),
        )
        self.return_patterns[pattern_key] = bucket
        timestamps = _lru_pop_or_create(
            bucket, client_ip, _MAX_TRACKED_CLIENTS_PER_ENDPOINT, list
        )

        timestamps[:] = [ts for ts in timestamps if ts >= window_start]

        timestamps.append(current_time)
        bucket[client_ip] = timestamps

        return len(timestamps) > threshold

The Behavior Manager handles behavioral analysis for detecting suspicious usage patterns.

BehaviorRule

guard_core.handlers.behavior_handler.BehaviorRule(rule_type, threshold, window=3600, pattern=None, action='log', custom_action=None, ban_duration=None, correlate_with_detection=False)

Source code in guard_core/handlers/behavior_handler.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    rule_type: Literal["usage", "return_pattern", "frequency"],
    threshold: int,
    window: int = 3600,
    pattern: str | None = None,
    action: Literal["ban", "log", "throttle", "alert"] = "log",
    custom_action: Callable | None = None,
    ban_duration: int | None = None,
    correlate_with_detection: bool = False,
):
    self.rule_type = rule_type
    self.threshold = threshold
    self.window = window
    self.pattern = pattern
    self.action = action
    self.custom_action = custom_action
    self.ban_duration = ban_duration
    self.correlate_with_detection = correlate_with_detection

action = action instance-attribute

ban_duration = ban_duration instance-attribute

correlate_with_detection = correlate_with_detection instance-attribute

custom_action = custom_action instance-attribute

pattern = pattern instance-attribute

rule_type = rule_type instance-attribute

threshold = threshold instance-attribute

window = window instance-attribute

Rule types: usage, return_pattern, frequency

Pattern formats: Simple string, JSON path (json:result.status==win), Regex (regex:win|victory), Status code (status:200)

Actions: ban, log, alert, throttle

Integration with Decorators

guard_deco = SecurityDecorator(config)

@guard_deco.usage_monitor(max_calls=10, window=3600, action="ban")
@guard_deco.return_monitor("rare_item", max_occurrences=3, window=86400, action="alert")
def rewards_endpoint(request):
    return JsonResponse({"reward": "rare_item"})

See Also