INNER CODE UNIT · Python

_rate_ok

cipi-sh/cipi · lib/cipi-crowdsec-rescue.py:64

def _rate_ok(ip: str) -> bool:
    now = time.monotonic()
    with _fails_lock:
        q = _fails[ip]
        while q and now - q[0] > WINDOW:
            q.popleft()
        return len(q) < MAX_FAILS


def _rate_hit(ip: str) -> None:
    with _fails_lock:
        if len(_fails) >= MAX_TRACKED_PEERS and ip not in _fails:
            # One deque per source address is unbounded memory for a port anyone
            # can reach. Drop windows that have already expired, then give up on
            # tracking rather than grow.
            now = time.monotonic()
            for stale in [k for k, q in _fails.items() if not q or now - q[-1] > WINDOW]:
                del _fails[stale]

View source record →