server_support.py

import datetime
import socketserver
import threading
from typing import Any


class ThreadingTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True
    daemon_threads = True


class AdminGuard:
    WINDOW_SECONDS = 300
    MAX_FAILED_IN_WINDOW = 5

    def __init__(self, database: Any):
        self.database = database
        self._lock = threading.Lock()

    def register_failure(self, ip: str) -> None:
        now_dt = datetime.datetime.now(datetime.timezone.utc)
        with self._lock:
            fail_count, first_fail = self.database.get_admin_fail_state(ip)
            if first_fail is None or (now_dt - first_fail).total_seconds() > self.WINDOW_SECONDS:
                self.database.set_admin_fail_state(ip, 1, now_dt)
                return
            self.database.set_admin_fail_state(ip, fail_count + 1, first_fail)

    def register_success(self, ip: str) -> None:
        with self._lock:
            self.database.clear_admin_fail_state(ip)

    def can_attempt(self, ip: str) -> tuple[bool, int]:
        now_dt = datetime.datetime.now(datetime.timezone.utc)
        with self._lock:
            fail_count, first_fail = self.database.get_admin_fail_state(ip)
            if fail_count < self.MAX_FAILED_IN_WINDOW or first_fail is None:
                return True, 0

            elapsed = (now_dt - first_fail).total_seconds()
            if elapsed > self.WINDOW_SECONDS:
                self.database.clear_admin_fail_state(ip)
                return True, 0

            retry_after = int(self.WINDOW_SECONDS - elapsed)
            return False, max(retry_after, 1)