65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from hashlib import sha256
|
|
import json
|
|
|
|
from backend.app.database import get_connection
|
|
|
|
MAX_FAILURES = 5
|
|
FAILURE_WINDOW = timedelta(minutes=15)
|
|
LOCKOUT_DURATION = timedelta(minutes=2)
|
|
|
|
|
|
def _setting_name(ip_address: str, email: str) -> str:
|
|
key = sha256(f'{ip_address}\0{email.casefold()}'.encode('utf-8')).hexdigest()
|
|
return f'login_rate:{key}'
|
|
|
|
|
|
def _read_rate(setting_name: str) -> dict:
|
|
with get_connection() as conn:
|
|
row = conn.execute('SELECT value FROM app_settings WHERE name = ?', (setting_name,)).fetchone()
|
|
if not row:
|
|
return {}
|
|
try:
|
|
return json.loads(row['value'])
|
|
except (TypeError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def check_login_allowed(ip_address: str, email: str) -> int | None:
|
|
rate = _read_rate(_setting_name(ip_address, email))
|
|
now = datetime.now(timezone.utc)
|
|
locked_until = datetime.fromisoformat(rate['locked_until']) if rate.get('locked_until') else None
|
|
if locked_until and locked_until > now:
|
|
return int((locked_until - now).total_seconds()) + 1
|
|
return None
|
|
|
|
|
|
def record_login_failure(ip_address: str, email: str) -> None:
|
|
setting_name = _setting_name(ip_address, email)
|
|
now = datetime.now(timezone.utc)
|
|
rate = _read_rate(setting_name)
|
|
first_failure = datetime.fromisoformat(rate['first_failure']) if rate.get('first_failure') else now
|
|
if now - first_failure >= FAILURE_WINDOW:
|
|
rate = {}
|
|
first_failure = now
|
|
failures = int(rate.get('failures', 0)) + 1
|
|
updated = {'failures': failures, 'first_failure': first_failure.isoformat()}
|
|
if failures >= MAX_FAILURES:
|
|
updated['locked_until'] = (now + LOCKOUT_DURATION).isoformat()
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
'''INSERT INTO app_settings (name, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(name) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP''',
|
|
(setting_name, json.dumps(updated)),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def clear_login_failures(ip_address: str, email: str) -> None:
|
|
with get_connection() as conn:
|
|
conn.execute('DELETE FROM app_settings WHERE name = ?', (_setting_name(ip_address, email),))
|
|
conn.commit()
|