rewritten after codex recommendations

2026-07-18 20:27:19 +02:00
parent f9dc885937
commit 399f539b4f
22 changed files with 1448 additions and 996 deletions
@@ -1,40 +1,68 @@
-from Config import *
-import pickle, os
-import datetime
+"""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():
-    access_list = []

-    def __init__(self):
-        five_minutes_ago = datetime.datetime.now() - datetime.timedelta(minutes=5)
-        new_list = []
-        if os.path.exists(ACCESSFILE):
-            with open(ACCESSFILE, 'rb') as pf:
-                self.access_list = pickle.load(pf)
-            for entry in self.access_list:
-                if entry['time'] > five_minutes_ago:
-                    new_list.append(entry)
-            self.access_list = new_list
+class Access:
+    limit = 3
+    window = timedelta(minutes=5)

-    def granted(self, ipaddress):
-        result = 0
-        for entry in self.access_list:
-            if entry['ip'] == ipaddress:
-                result += 1
+    def __init__(self, path: str = ACCESSFILE) -> None:
+        self.path = Path(path)
+        self.access_list = self._load()

-        if result<3:
+    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("Access denied for {}".format(ipaddress))
-        self.deny(ipaddress)
+        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 deny(self, ipaddress):
-        now = datetime.datetime.now()
-        self.access_list.append({'ip': ipaddress, 'time': now})
-        with open(ACCESSFILE, 'wb') as pf:
-            pickle.dump(self.access_list, pf, 2)
-
+    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)