INNER CODE UNIT · Python

_rate_hit

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

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]
            if len(_fails) >= MAX_TRACKED_PEERS:
                return
        _fails[ip].append(time.monotonic())


def _scrape_token(out: str) -> str:
    for line in (out or "").splitlines():
        if line.startswith("RESCUE_OK "):
            return line[len("RESCUE_OK ") :].strip()

View source record →