69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
"""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)
|