flowers/AccessControl.py

"""Small file-backed login throttle.

The legacy pickle file is deliberately not read: pickle is unsafe for mutable
runtime files. A JSON document is used instead and malformed files fail closed
to an empty recent-attempt list.
"""

from __future__ import annotations

import json
import os
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

from Config import ACCESSFILE
from Log import Log


class Access:
    limit = 3
    window = timedelta(minutes=5)

    def __init__(self, path: str = ACCESSFILE) -> None:
        self.path = Path(path)
        self.access_list = self._load()

    def granted(self, ipaddress: str) -> bool:
        failures = sum(entry["ip"] == ipaddress for entry in self.access_list)
        if failures < self.limit:
            return True
        Log.info(f"Access denied for {ipaddress}")
        return False

    def deny(self, ipaddress: str) -> None:
        self.access_list.append(
            {"ip": str(ipaddress), "time": datetime.now(timezone.utc).isoformat()}
        )
        self._save()

    def _load(self) -> list[dict[str, str]]:
        cutoff = datetime.now(timezone.utc) - self.window
        try:
            raw = json.loads(self.path.read_text(encoding="utf-8"))
            recent = []
            for entry in raw:
                recorded = datetime.fromisoformat(entry["time"])
                if recorded.tzinfo is None:
                    recorded = recorded.replace(tzinfo=timezone.utc)
                if recorded > cutoff:
                    recent.append({"ip": str(entry["ip"]), "time": recorded.isoformat()})
            return recent
        except (OSError, ValueError, TypeError, KeyError):
            return []

    def _save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        descriptor, temporary = tempfile.mkstemp(
            prefix=f".{self.path.name}.", dir=str(self.path.parent), text=True
        )
        try:
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                json.dump(self.access_list, handle, separators=(",", ":"))
            os.chmod(temporary, 0o600)
            os.replace(temporary, self.path)
        finally:
            if os.path.exists(temporary):
                os.unlink(temporary)