117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
import hashlib
|
|
import os
|
|
import secrets
|
|
import time
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional
|
|
from app import config
|
|
from app.database import get_db
|
|
|
|
# In-memory IP-based rate limiting for login attempts
|
|
_login_failures: dict[str, list[float]] = {}
|
|
|
|
def hash_password(password: str, salt: Optional[str] = None) -> tuple[str, str]:
|
|
if not salt:
|
|
salt = secrets.token_hex(16)
|
|
salt_bytes = bytes.fromhex(salt)
|
|
key = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt_bytes, 100_000)
|
|
return key.hex(), salt
|
|
|
|
def verify_password(password: str, stored_hash: str, salt: str) -> bool:
|
|
new_hash, _ = hash_password(password, salt)
|
|
return secrets.compare_digest(new_hash, stored_hash)
|
|
|
|
def check_login_rate_limit(ip: str) -> bool:
|
|
now = time.time()
|
|
cutoff = now - config.RATE_LIMIT_LOGIN_WINDOW_SEC
|
|
attempts = _login_failures.get(ip, [])
|
|
# Filter attempts within the time window
|
|
recent_attempts = [t for t in attempts if t > cutoff]
|
|
_login_failures[ip] = recent_attempts
|
|
return len(recent_attempts) < config.RATE_LIMIT_LOGIN_MAX
|
|
|
|
def record_login_failure(ip: str):
|
|
now = time.time()
|
|
if ip not in _login_failures:
|
|
_login_failures[ip] = []
|
|
_login_failures[ip].append(now)
|
|
|
|
def record_login_success(ip: str):
|
|
if ip in _login_failures:
|
|
_login_failures.pop(ip, None)
|
|
|
|
def create_user(username: str, password: str) -> int:
|
|
pwd_hash, salt = hash_password(password)
|
|
with get_db() as conn:
|
|
cursor = conn.execute(
|
|
"INSERT INTO users (username, password_hash, salt) VALUES (?, ?, ?)",
|
|
(username.strip(), pwd_hash, salt)
|
|
)
|
|
return cursor.lastrowid
|
|
|
|
def authenticate_user(username: str, password: str) -> Optional[dict]:
|
|
with get_db() as conn:
|
|
row = conn.execute(
|
|
"SELECT id, username, password_hash, salt FROM users WHERE username = ?",
|
|
(username.strip(),)
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
if verify_password(password, row["password_hash"], row["salt"]):
|
|
return {"id": row["id"], "username": row["username"]}
|
|
return None
|
|
|
|
def create_session(user_id: int, user_agent: str = "", ip: str = "") -> str:
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = (datetime.now(timezone.utc) + timedelta(days=config.SESSION_EXPIRE_DAYS)).isoformat()
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"INSERT INTO sessions (token, user_id, expires_at, user_agent, ip) VALUES (?, ?, ?, ?, ?)",
|
|
(token, user_id, expires_at, user_agent, ip)
|
|
)
|
|
return token
|
|
|
|
def get_user_from_session(token: str) -> Optional[dict]:
|
|
if not token:
|
|
return None
|
|
with get_db() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT u.id, u.username, s.expires_at
|
|
FROM sessions s
|
|
JOIN users u ON s.user_id = u.id
|
|
WHERE s.token = ?
|
|
""",
|
|
(token,)
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
|
|
expires_at_str = row["expires_at"]
|
|
try:
|
|
expires_at = datetime.fromisoformat(expires_at_str)
|
|
if expires_at.tzinfo is None:
|
|
expires_at = expires_at.replace(tzinfo=timezone.utc)
|
|
except Exception:
|
|
return None
|
|
|
|
if expires_at < datetime.now(timezone.utc):
|
|
# Session expired
|
|
delete_session(token)
|
|
return None
|
|
|
|
return {"id": row["id"], "username": row["username"]}
|
|
|
|
def delete_session(token: str):
|
|
if not token:
|
|
return
|
|
with get_db() as conn:
|
|
conn.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
|
|
def ensure_admin_user():
|
|
"""Ensures at least one user exists; creates default admin from config if table empty."""
|
|
with get_db() as conn:
|
|
count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
|
|
if count == 0 and config.ADMIN_PASSWORD:
|
|
create_user(config.ADMIN_USERNAME, config.ADMIN_PASSWORD)
|