diff --git a/database/schema.py b/database/schema.py index 717b5aa..d8fa57b 100644 --- a/database/schema.py +++ b/database/schema.py @@ -3,6 +3,16 @@ from database.connection import get_db def _migrate(conn): """Add columns that may not exist in older databases.""" + # users.role migration + cursor = conn.execute("PRAGMA table_info(users)") + user_cols = {row[1] for row in cursor.fetchall()} + if "role" not in user_cols: + conn.execute("ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'user'") + # Promote the first-created user to admin so the app remains usable. + first = conn.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone() + if first: + conn.execute("UPDATE users SET role='admin' WHERE id=?", (first[0],)) + cursor = conn.execute("PRAGMA table_info(documents)") existing = {row[1] for row in cursor.fetchall()} migrations = [ @@ -44,9 +54,20 @@ def create_tables(): id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); + CREATE TABLE IF NOT EXISTS login_attempts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT, + ip TEXT, + success INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS idx_login_attempts_username ON login_attempts(username, created_at); + CREATE INDEX IF NOT EXISTS idx_login_attempts_ip ON login_attempts(ip, created_at); + CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, diff --git a/main.py b/main.py index 3b29145..1ec74b8 100644 --- a/main.py +++ b/main.py @@ -51,10 +51,13 @@ async def check_authentication(request: Request, call_next): allowed_paths = ["/auth/login", "/static"] is_allowed = any(path.startswith(p) for p in allowed_paths) + request.state.user = None if not is_allowed: session_id = request.cookies.get("session_id") - if not session_id or not get_session(session_id): + user = get_session(session_id) if session_id else None + if not user: return RedirectResponse(url="/auth/login", status_code=303) + request.state.user = user response = await call_next(request) return response diff --git a/routers/auth.py b/routers/auth.py index 86d5ca4..3990299 100644 --- a/routers/auth.py +++ b/routers/auth.py @@ -4,6 +4,8 @@ from fastapi.templating import Jinja2Templates from services.auth_service import ( verify_password, get_user_by_username, create_user, delete_user, get_all_users, create_session, get_session, delete_session, SESSION_TTL_SECONDS, + needs_rehash, update_user_password, set_user_role, count_admins, + record_login_attempt, is_login_blocked, ) from services.backup_service import create_backup import os @@ -12,10 +14,19 @@ router = APIRouter() templates = Jinja2Templates(directory="templates") +def _client_ip(request: Request) -> str: + # Trust X-Forwarded-For when behind a reverse proxy + xff = request.headers.get("x-forwarded-for") + if xff: + return xff.split(",")[0].strip() + return request.client.host if request.client else "unknown" + + def get_current_user_from_request(request: Request): session_id = request.cookies.get("session_id") return get_session(session_id) if session_id else None + def get_current_user(request: Request): user = get_current_user_from_request(request) if not user: @@ -25,20 +36,53 @@ def get_current_user(request: Request): ) return user + +def require_admin(request: Request): + user = get_current_user(request) + if user.get("role") != "admin": + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + @router.get("/login", response_class=HTMLResponse) async def login_get(request: Request): return templates.TemplateResponse(request=request, name="login.html", context={"error": None}) @router.post("/login", response_class=HTMLResponse) async def login_post(request: Request, username: str = Form(...), password: str = Form(...)): + username = (username or "").strip() + ip = _client_ip(request) + + if is_login_blocked(username, ip): + return templates.TemplateResponse( + request=request, + name="login.html", + context={"error": "تم حجب محاولات تسجيل الدخول مؤقتاً. حاول بعد 15 دقيقة."}, + status_code=429, + ) + user = get_user_by_username(username) if not user or not verify_password(user["password_hash"], password): - return templates.TemplateResponse(request=request, name="login.html", context={"error": "Invalid username or password"}) + record_login_attempt(username, ip, success=False) + return templates.TemplateResponse( + request=request, + name="login.html", + context={"error": "Invalid username or password"}, + status_code=401, + ) - session_id = create_session(user["id"], user["username"]) + record_login_attempt(username, ip, success=True) + + # Opportunistically upgrade the password hash if it's using weaker params. + try: + if needs_rehash(user["password_hash"]): + update_user_password(user["id"], password) + except Exception: + pass + + role = user["role"] if "role" in user.keys() else "user" + session_id = create_session(user["id"], user["username"], role=role) from config import ENVIRONMENT - # secure=True only when accessed via HTTPS (check X-Forwarded-Proto from nginx) is_https = ENVIRONMENT == "production" and request.headers.get("x-forwarded-proto") == "https" response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER) response.set_cookie( @@ -48,6 +92,7 @@ async def login_post(request: Request, username: str = Form(...), password: str secure=is_https, samesite="lax", max_age=SESSION_TTL_SECONDS, + path="/", ) return response @@ -61,34 +106,101 @@ async def logout(request: Request): return response @router.get("/users", response_class=HTMLResponse) -async def users_list(request: Request, _=Depends(get_current_user)): +async def users_list(request: Request, current=Depends(require_admin)): users = get_all_users() - return templates.TemplateResponse(request=request, name="users.html", context={"users": users}) + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current}, + ) @router.post("/users/create") -async def add_user(request: Request, username: str = Form(...), password: str = Form(...), _=Depends(get_current_user)): +async def add_user( + request: Request, + username: str = Form(...), + password: str = Form(...), + role: str = Form("user"), + current=Depends(require_admin), +): + username = (username or "").strip() + if role not in {"admin", "user"}: + role = "user" if len(password) < 8: users = get_all_users() - return templates.TemplateResponse(request=request, name="users.html", context={"users": users, "error": "Password must be at least 8 characters."}) + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current, "error": "Password must be at least 8 characters."}, + ) if get_user_by_username(username): users = get_all_users() - return templates.TemplateResponse(request=request, name="users.html", context={"users": users, "error": f"User '{username}' already exists."}) - create_user(username, password) + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current, "error": f"User '{username}' already exists."}, + ) + create_user(username, password, role=role) return RedirectResponse(url="/auth/users", status_code=status.HTTP_303_SEE_OTHER) @router.post("/users/delete/{user_id}") -async def remove_user(user_id: int, _=Depends(get_current_user)): +async def remove_user(request: Request, user_id: int, current=Depends(require_admin)): + if user_id == current["user_id"]: + users = get_all_users() + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current, "error": "لا يمكنك حذف حسابك الخاص."}, + status_code=400, + ) + # Prevent removing the last admin + target = next((u for u in get_all_users() if u["id"] == user_id), None) + if target and target["role"] == "admin" and count_admins() <= 1: + users = get_all_users() + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current, "error": "لا يمكن حذف آخر مسؤول في النظام."}, + status_code=400, + ) delete_user(user_id) return RedirectResponse(url="/auth/users", status_code=status.HTTP_303_SEE_OTHER) + +@router.post("/users/{user_id}/role") +async def change_role( + request: Request, + user_id: int, + role: str = Form(...), + current=Depends(require_admin), +): + if role not in {"admin", "user"}: + raise HTTPException(status_code=400, detail="invalid role") + # Prevent demoting the last admin + target = next((u for u in get_all_users() if u["id"] == user_id), None) + if target and target["role"] == "admin" and role != "admin" and count_admins() <= 1: + users = get_all_users() + return templates.TemplateResponse( + request=request, + name="users.html", + context={"users": users, "current_user": current, "error": "لا يمكن تخفيض رتبة آخر مسؤول."}, + status_code=400, + ) + set_user_role(user_id, role) + return RedirectResponse(url="/auth/users", status_code=status.HTTP_303_SEE_OTHER) + @router.get("/backup") -async def backup_db(request: Request, _=Depends(get_current_user)): +async def backup_db(request: Request, current=Depends(require_admin)): try: backup_path = create_backup() return FileResponse(backup_path, media_type="application/octet-stream", filename=os.path.basename(backup_path)) except Exception as e: msg = f"Backup failed: {str(e)}" - return templates.TemplateResponse(request=request, name="users.html", context={ - "users": get_all_users(), - "backup_msg": msg - }) + return templates.TemplateResponse( + request=request, + name="users.html", + context={ + "users": get_all_users(), + "current_user": current, + "backup_msg": msg, + }, + ) diff --git a/routers/upload.py b/routers/upload.py index 7e58217..b53e46d 100644 --- a/routers/upload.py +++ b/routers/upload.py @@ -5,7 +5,7 @@ import uuid from datetime import date from pathlib import Path -from fastapi import APIRouter, File, Form, Request, UploadFile +from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile from fastapi.responses import JSONResponse, RedirectResponse from fastapi.templating import Jinja2Templates @@ -21,6 +21,35 @@ ALLOWED_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp"} ALLOWED_PDF_EXTS = {".pdf"} ALLOWED_EXTENSIONS = ALLOWED_IMAGE_EXTS | ALLOWED_PDF_EXTS +MAX_FILE_BYTES = 25 * 1024 * 1024 # 25 MB per file +MAX_TOTAL_BYTES = 100 * 1024 * 1024 # 100 MB per request + + +def _check_request_size(request: Request) -> None: + """Reject oversize uploads early via Content-Length header.""" + cl = request.headers.get("content-length") + if cl and cl.isdigit() and int(cl) > MAX_TOTAL_BYTES: + raise HTTPException( + status_code=413, + detail=f"حجم الطلب يتجاوز الحد الأقصى ({MAX_TOTAL_BYTES // (1024*1024)} ميغابايت).", + ) + + +async def _read_capped(upload: UploadFile, running_total: int) -> bytes: + """Read upload bytes, enforcing per-file and cumulative caps.""" + data = await upload.read() + if len(data) > MAX_FILE_BYTES: + raise HTTPException( + status_code=413, + detail=f"الملف '{upload.filename}' يتجاوز {MAX_FILE_BYTES // (1024*1024)} ميغابايت.", + ) + if running_total + len(data) > MAX_TOTAL_BYTES: + raise HTTPException( + status_code=413, + detail=f"إجمالي حجم الرفع يتجاوز {MAX_TOTAL_BYTES // (1024*1024)} ميغابايت.", + ) + return data + def _hash_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() @@ -179,18 +208,21 @@ async def upload_files( files: list[UploadFile] = File(...), provider: str = Form(""), ): + _check_request_size(request) if not provider: provider = get_default_provider() doc_ids = [] duplicates = [] + total_bytes = 0 for upload in files: suffix = Path(upload.filename).suffix.lower() if suffix not in ALLOWED_EXTENSIONS: continue - file_bytes = await upload.read() + file_bytes = await _read_capped(upload, total_bytes) + total_bytes += len(file_bytes) if suffix in ALLOWED_PDF_EXTS: # PDF: split into per-page images; hash each rendered page @@ -268,14 +300,17 @@ async def stage_file( files: list[UploadFile] = File(...), ): """Save uploaded images without triggering AI extraction.""" + _check_request_size(request) staged = [] duplicates = [] + total_bytes = 0 for upload in files: suffix = Path(upload.filename).suffix.lower() if suffix not in ALLOWED_EXTENSIONS: continue - file_bytes = await upload.read() + file_bytes = await _read_capped(upload, total_bytes) + total_bytes += len(file_bytes) if suffix in ALLOWED_PDF_EXTS: pages = pdf_to_images(file_bytes, upload.filename) diff --git a/services/auth_service.py b/services/auth_service.py index 11c0c82..99f636f 100644 --- a/services/auth_service.py +++ b/services/auth_service.py @@ -4,47 +4,148 @@ import secrets from datetime import datetime, timedelta SESSION_TTL_SECONDS = 86400 # 24h +PBKDF2_ITERATIONS = 600_000 # OWASP 2023+ guidance for PBKDF2-HMAC-SHA256 -def hash_password(password: str) -> str: +# Login rate-limit tuning +LOGIN_WINDOW_SECONDS = 900 # 15 min sliding window +LOGIN_MAX_FAILS_USER = 5 # per username +LOGIN_MAX_FAILS_IP = 20 # per IP (higher because NATs) + +def hash_password(password: str, iterations: int = PBKDF2_ITERATIONS) -> str: salt = secrets.token_hex(16) - hashed = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt.encode('utf-8'), 100000).hex() - return f"{salt}${hashed}" + hashed = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt.encode('utf-8'), iterations).hex() + return f"pbkdf2${iterations}${salt}${hashed}" def verify_password(stored_password: str, provided_password: str) -> bool: + """Supports both the legacy 'salt$hash' (100k iter) format and the new + 'pbkdf2$$salt$hash' format.""" try: - salt, stored_hash = stored_password.split('$') - hashed = hashlib.pbkdf2_hmac('sha256', provided_password.encode('utf-8'), salt.encode('utf-8'), 100000).hex() + parts = stored_password.split('$') + if len(parts) == 4 and parts[0] == 'pbkdf2': + iterations = int(parts[1]) + salt = parts[2] + stored_hash = parts[3] + elif len(parts) == 2: + iterations = 100_000 # legacy + salt, stored_hash = parts + else: + return False + hashed = hashlib.pbkdf2_hmac('sha256', provided_password.encode('utf-8'), salt.encode('utf-8'), iterations).hex() return hmac.compare_digest(hashed, stored_hash) except Exception: return False + +def needs_rehash(stored_password: str) -> bool: + """Return True if the stored hash uses weaker params than current defaults.""" + try: + parts = stored_password.split('$') + if len(parts) == 4 and parts[0] == 'pbkdf2': + return int(parts[1]) < PBKDF2_ITERATIONS + return True # legacy format → rehash + except Exception: + return True + + def get_user_by_username(username: str): from database.connection import get_db with get_db() as conn: cursor = conn.execute("SELECT * FROM users WHERE username = ?", (username,)) return cursor.fetchone() -def create_user(username: str, password: str): +def create_user(username: str, password: str, role: str = "user"): from database.connection import get_db + if role not in {"admin", "user"}: + role = "user" with get_db() as conn: hashed = hash_password(password) - conn.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)", (username, hashed)) + conn.execute( + "INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)", + (username, hashed, role), + ) def delete_user(user_id: int): from database.connection import get_db with get_db() as conn: conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) + conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) + + +def update_user_password(user_id: int, new_password: str) -> None: + from database.connection import get_db + with get_db() as conn: + conn.execute( + "UPDATE users SET password_hash=? WHERE id=?", + (hash_password(new_password), user_id), + ) + # Rotate any active sessions for this user. + conn.execute("DELETE FROM sessions WHERE user_id=?", (user_id,)) + + +def set_user_role(user_id: int, role: str) -> None: + from database.connection import get_db + if role not in {"admin", "user"}: + raise ValueError("invalid role") + with get_db() as conn: + conn.execute("UPDATE users SET role=? WHERE id=?", (role, user_id)) + + +def count_admins() -> int: + from database.connection import get_db + with get_db() as conn: + row = conn.execute("SELECT COUNT(*) AS n FROM users WHERE role='admin'").fetchone() + return int(row["n"]) if row else 0 + def get_all_users(): from database.connection import get_db with get_db() as conn: - cursor = conn.execute("SELECT id, username, created_at FROM users ORDER BY created_at DESC") + cursor = conn.execute( + "SELECT id, username, role, created_at FROM users ORDER BY created_at DESC" + ) return cursor.fetchall() +# ─── Login rate-limiting ────────────────────────────────────────── + +def record_login_attempt(username: str, ip: str, success: bool) -> None: + from database.connection import get_db + with get_db() as conn: + conn.execute( + "INSERT INTO login_attempts (username, ip, success) VALUES (?, ?, ?)", + (username, ip, 1 if success else 0), + ) + + +def is_login_blocked(username: str, ip: str) -> bool: + """Return True if too many recent failures for this username OR this IP.""" + from database.connection import get_db + cutoff = (datetime.utcnow() - timedelta(seconds=LOGIN_WINDOW_SECONDS)).strftime("%Y-%m-%d %H:%M:%S") + with get_db() as conn: + by_user = conn.execute( + "SELECT COUNT(*) AS n FROM login_attempts WHERE username=? AND success=0 AND created_at > ?", + (username, cutoff), + ).fetchone()["n"] + if by_user >= LOGIN_MAX_FAILS_USER: + return True + by_ip = conn.execute( + "SELECT COUNT(*) AS n FROM login_attempts WHERE ip=? AND success=0 AND created_at > ?", + (ip, cutoff), + ).fetchone()["n"] + return by_ip >= LOGIN_MAX_FAILS_IP + + +def cleanup_old_login_attempts() -> int: + from database.connection import get_db + cutoff = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d %H:%M:%S") + with get_db() as conn: + cur = conn.execute("DELETE FROM login_attempts WHERE created_at < ?", (cutoff,)) + return cur.rowcount or 0 + + # ─── Sessions (DB-backed) ───────────────────────────────────────── -def create_session(user_id: int, username: str, ttl_seconds: int = SESSION_TTL_SECONDS) -> str: +def create_session(user_id: int, username: str, role: str = "user", ttl_seconds: int = SESSION_TTL_SECONDS) -> str: from database.connection import get_db session_id = secrets.token_urlsafe(32) expires_at = (datetime.utcnow() + timedelta(seconds=ttl_seconds)).strftime("%Y-%m-%d %H:%M:%S") @@ -57,17 +158,24 @@ def create_session(user_id: int, username: str, ttl_seconds: int = SESSION_TTL_S def get_session(session_id: str): - """Return dict with user_id/username if session is valid and unexpired, else None.""" + """Return dict with user_id/username/role if valid and unexpired, else None.""" if not session_id: return None from database.connection import get_db now = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") with get_db() as conn: row = conn.execute( - "SELECT id, user_id, username, expires_at FROM sessions WHERE id = ? AND expires_at > ?", + """SELECT s.id, s.user_id, s.username, s.expires_at, u.role + FROM sessions s + LEFT JOIN users u ON u.id = s.user_id + WHERE s.id = ? AND s.expires_at > ?""", (session_id, now), ).fetchone() - return dict(row) if row else None + if not row: + return None + d = dict(row) + d["role"] = d.get("role") or "user" + return d def delete_session(session_id: str) -> None: diff --git a/templates/base.html b/templates/base.html index 9ad5f1b..5f3df69 100644 --- a/templates/base.html +++ b/templates/base.html @@ -50,10 +50,12 @@ قائمة الوثائق + {% if request.state.user and request.state.user.role == 'admin' %} الإدارة + {% endif %} خروج diff --git a/templates/users.html b/templates/users.html index 72e7903..dc27d4b 100644 --- a/templates/users.html +++ b/templates/users.html @@ -15,6 +15,9 @@ {% if backup_msg %}
{{ backup_msg }}
{% endif %} + {% if error %} +
{{ error }}
+ {% endif %}

@@ -37,7 +40,14 @@

- + +
+
+ +
@@ -54,6 +64,7 @@ الرقم اسم المستخدم + الصلاحية تاريخ الإنشاء @@ -62,12 +73,30 @@ {% for user in users %} {{ user.id }} - {{ user.username }} + + {{ user.username }} + {% if current_user and user.id == current_user.user_id %}(أنت){% endif %} + + +
+ + {% if not (current_user and user.id == current_user.user_id) %} + + {% endif %} +
+ {{ user.created_at }} + {% if current_user and user.id == current_user.user_id %} + — + {% else %}
+ {% endif %} {% endfor %}