Coverage for src/lexigram/auth/services/activity_tracker.py: 95%

1"""In-process tracker for auth activity (failed logins, token refreshes). 

2 

3Bounded ring buffers with wall-clock windows. Counters are real runtime 

4observations written by the login and JWT-lifecycle paths. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections import deque 

10from collections.abc import Callable 

11import threading 

12import time 

13 

14_MAX_EVENTS = 10_000 

15 

16 

17class AuthActivityTracker: 

18 """Thread-safe, bounded tracker of auth events.""" 

19 

20 def __init__( 

21 self, 

22 max_events: int = _MAX_EVENTS, 

23 window_size_seconds: int = 3600, 

24 now: Callable[[], float] | None = None, 

25 ) -> None: 

26 """Initialize the tracker. 

27 

28 Args: 

29 max_events: Maximum buffered events per stream before pruning. 

30 window_size_seconds: Retention window for buffered events. 

31 now: Clock function returning seconds; defaults to ``time.monotonic``. 

32 """ 

33 self._max_events = max_events 

34 self._window_size = window_size_seconds 

35 self._now = now or time.monotonic 

36 self._failures: deque[tuple[float, str]] = deque() 

37 self._refreshes: deque[float] = deque() 

38 self._lock = threading.Lock() 

39 

40 def record_failed_login(self, ip: str = "unknown") -> None: 

41 """Record one failed login attempt from ``ip``. 

42 

43 Args: 

44 ip: Client IP when known; defaults to ``"unknown"`` for auth 

45 paths that carry no request context. 

46 """ 

47 with self._lock: 

48 self._failures.append((self._now(), ip)) 

49 self._prune(self._failures, self._now() - self._window_size) 

50 

51 def record_refresh(self) -> None: 

52 """Record one token refresh.""" 

53 with self._lock: 

54 self._refreshes.append(self._now()) 

55 self._prune(self._refreshes, self._now() - self._window_size) 

56 

57 def failed_login_summary(self, window_minutes: int) -> tuple[int, int]: 

58 """Return ``(count, unique_ips)`` for the last ``window_minutes``. 

59 

60 Args: 

61 window_minutes: Look-back window in minutes. 

62 

63 Returns: 

64 Tuple of failure count and distinct source IPs. 

65 """ 

66 cutoff = self._now() - window_minutes * 60 

67 with self._lock: 

68 recent = [ip for ts, ip in self._failures if ts >= cutoff] 

69 return len(recent), len(set(recent)) 

70 

71 def refresh_summary(self, window_minutes: int) -> tuple[float, int]: 

72 """Return ``(rate_per_minute, total)`` for the last ``window_minutes``. 

73 

74 Args: 

75 window_minutes: Look-back window in minutes. 

76 

77 Returns: 

78 Tuple of refreshes-per-minute and total refresh count. 

79 """ 

80 cutoff = self._now() - window_minutes * 60 

81 with self._lock: 

82 recent = [ts for ts in self._refreshes if ts >= cutoff] 

83 minutes = max(window_minutes, 1) 

84 return round(len(recent) / minutes, 1), len(recent) 

85 

86 def _prune(self, buffer: deque, cutoff: float, key: int = 0) -> None: 

87 """Drop entries older than ``cutoff`` and over the size bound.""" 

88 while buffer: 

89 entry = buffer[0] 

90 value = entry[key] if isinstance(entry, tuple) else entry 

91 if value >= cutoff: 

92 break 

93 buffer.popleft() 

94 while len(buffer) > self._max_events: 

95 buffer.popleft() 

96 

97 

98__all__ = ["AuthActivityTracker"]