128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""Auth blueprint — login/logout via Dovecot IMAP credentials.
|
|
|
|
Phase 7 minimal: validate by attempting an IMAP login against the configured
|
|
host/port. On success store user_email + password (signed cookie via Flask
|
|
session) so subsequent requests can talk to IMAP/SMTP as that user.
|
|
|
|
App-passwords + persistent server-side sessions land later.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from collections import defaultdict, deque
|
|
|
|
from flask import Blueprint, current_app, flash, redirect, render_template, request, session, url_for
|
|
from imapclient import IMAPClient
|
|
|
|
bp = Blueprint("auth", __name__, template_folder="../../templates")
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# ── In-memory rate limiter ─────────────────────────────────────────────
|
|
# Per-IP: max 5 failed attempts in 60s → 60s cooldown before next try.
|
|
_RATE_MAX = 5
|
|
_RATE_WINDOW = 60.0
|
|
_RATE_COOLDOWN = 60.0
|
|
|
|
_rate_lock = threading.Lock()
|
|
_attempts: dict[str, deque] = defaultdict(deque)
|
|
|
|
|
|
def _client_ip() -> str:
|
|
fwd = request.headers.get("X-Forwarded-For", "")
|
|
return (fwd.split(",")[0].strip() if fwd else "") or request.remote_addr or "?"
|
|
|
|
|
|
def _rate_check() -> tuple[bool, float]:
|
|
"""Return (allowed, seconds_to_wait)."""
|
|
ip = _client_ip()
|
|
now = time.monotonic()
|
|
with _rate_lock:
|
|
q = _attempts[ip]
|
|
while q and q[0] < now - _RATE_WINDOW:
|
|
q.popleft()
|
|
if len(q) >= _RATE_MAX:
|
|
wait = _RATE_COOLDOWN - (now - q[-1])
|
|
return False, max(0.0, wait)
|
|
return True, 0.0
|
|
|
|
|
|
def _rate_record_failure() -> int:
|
|
ip = _client_ip()
|
|
now = time.monotonic()
|
|
with _rate_lock:
|
|
q = _attempts[ip]
|
|
while q and q[0] < now - _RATE_WINDOW:
|
|
q.popleft()
|
|
q.append(now)
|
|
return len(q)
|
|
|
|
|
|
def _rate_clear():
|
|
with _rate_lock:
|
|
_attempts.pop(_client_ip(), None)
|
|
|
|
|
|
def _try_imap_login(email: str, password: str) -> bool:
|
|
host = current_app.config["IMAP_HOST"]
|
|
port = int(current_app.config.get("IMAP_PORT", 143))
|
|
try:
|
|
c = IMAPClient(host, port=port, use_uid=True, ssl=False)
|
|
try:
|
|
c.login(email, password)
|
|
return True
|
|
finally:
|
|
try: c.logout()
|
|
except Exception: pass
|
|
except Exception as exc:
|
|
log.info("login failed for %s: %s", email, exc)
|
|
return False
|
|
|
|
|
|
@bp.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
from ...services import sessions as sess
|
|
ip = _client_ip()
|
|
ua = (request.headers.get("User-Agent") or "")[:400]
|
|
if request.method == "POST":
|
|
ok, wait = _rate_check()
|
|
if not ok:
|
|
flash(f"Слишком много попыток. Подождите {int(wait) + 1} сек.", "err")
|
|
return render_template("auth/login.html")
|
|
email = (request.form.get("email") or "").strip().lower()
|
|
password = request.form.get("password") or ""
|
|
if not email or not password:
|
|
flash("Заполните email и пароль", "err")
|
|
elif not _try_imap_login(email, password):
|
|
n = _rate_record_failure()
|
|
remaining = max(0, _RATE_MAX - n)
|
|
sess.audit("login_fail", email, ip, ua, extra=f"attempts={n}")
|
|
if remaining > 0:
|
|
flash(f"Неверный email или пароль (осталось попыток: {remaining})", "err")
|
|
else:
|
|
flash(f"Слишком много неудачных попыток. Подождите {int(_RATE_COOLDOWN)} сек.", "err")
|
|
else:
|
|
_rate_clear()
|
|
token = sess.create(email, password, ip=ip, user_agent=ua)
|
|
sess.audit("login_ok", email, ip, ua)
|
|
session.permanent = True
|
|
session["sid"] = token
|
|
next_url = request.args.get("next") or url_for("mail.folder", folder_key="inbox")
|
|
return redirect(next_url)
|
|
if session.get("sid"):
|
|
return redirect(url_for("mail.folder", folder_key="inbox"))
|
|
return render_template("auth/login.html")
|
|
|
|
|
|
@bp.route("/logout", methods=["GET", "POST"])
|
|
def logout():
|
|
from ...services import sessions as sess
|
|
token = session.get("sid")
|
|
if token:
|
|
sess.audit("logout", request.cookies.get("user_email", ""), _client_ip(), request.headers.get("User-Agent", ""))
|
|
sess.revoke(token)
|
|
session.clear()
|
|
return redirect(url_for("auth.login"))
|