security: validate sha256 digests, add login rate limiting, harden session cookies

server/app.py

@@ -5,8 +5,12 @@ Also serves the remote web portal for browsing and managing the configured car.
 import functools
 import hashlib
 import os
+import re
 import secrets
 import shutil
+import threading
+import time
+from collections import defaultdict
 
 from flask import (
     Flask, request, jsonify, g, render_template, redirect, url_for,
@@ -50,6 +54,15 @@ app = Flask(__name__)
 # Secret key for browser sessions — read from env in production.
 app.secret_key = os.environ.get("SECRET_KEY") or secrets.token_hex(32)
 
+# Session-cookie hardening. The portal is served over HTTPS via Cloudflare
+# Tunnel, so the cookie is marked Secure by default; set COOKIE_SECURE=0 in
+# .env only if you also log in to the portal over plain HTTP on the LAN.
+app.config.update(
+    SESSION_COOKIE_HTTPONLY=True,
+    SESSION_COOKIE_SAMESITE="Lax",
+    SESSION_COOKIE_SECURE=os.environ.get("COOKIE_SECURE", "1") == "1",
+)
+
 # Where uploaded dashcam files get written on disk (mounted to the unraid
 # array via docker-compose - see DASHCAM_STORAGE_DIR in .env).
 STORAGE_DIR = os.environ.get("DASHCAM_STORAGE_DIR", "/data/dashcam")
@@ -60,6 +73,15 @@ app.config["MAX_CONTENT_LENGTH"] = MAX_DASHCAM_UPLOAD_MB * 1024 * 1024
 # groups of 500, this just guards against a runaway/misbehaving client.
 MAX_TRACKING_BATCH_POINTS = 5000
 
+# Dashcam uploads are content-addressed by sha256, and that digest also names
+# the staging directory for chunked uploads. It MUST be a plain 64-char hex
+# string so a crafted digest can't traverse out of the staging directory.
+_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+
+
+def _is_valid_sha256(value):
+    return bool(_SHA256_RE.fullmatch(value or ""))
+
 # Initialise Flask-Login
 login_manager.init_app(app)
 
@@ -176,8 +198,8 @@ def upload_dashcam_file():
 
     relative_path = request.form.get("relative_path", "").strip()
     expected_sha256 = request.form.get("sha256", "").strip().lower()
-    if not relative_path or not expected_sha256:
-        return jsonify(error="missing 'relative_path' or 'sha256' form field"), 400
+    if not relative_path or not _is_valid_sha256(expected_sha256):
+        return jsonify(error="missing or invalid 'relative_path' or 'sha256' form field"), 400
 
     device_id = g.device_id
     if db.dashcam_file_exists(device_id, relative_path, expected_sha256):
@@ -251,8 +273,8 @@ def dashcam_upload_status():
     already has (e.g. after a network drop or app restart mid-transfer)."""
     relative_path = request.args.get("relative_path", "").strip()
     sha256 = request.args.get("sha256", "").strip().lower()
-    if not relative_path or not sha256:
-        return jsonify(error="missing 'relative_path' or 'sha256' query param"), 400
+    if not relative_path or not _is_valid_sha256(sha256):
+        return jsonify(error="missing or invalid 'relative_path' or 'sha256' query param"), 400
 
     if db.dashcam_file_exists(g.device_id, relative_path, sha256):
         return jsonify(status="already_uploaded"), 200
@@ -289,7 +311,8 @@ def dashcam_upload_chunk():
     except ValueError:
         return jsonify(error="'chunk_index'/'total_chunks' must be integers"), 400
 
-    if not relative_path or not expected_sha256 or total_chunks < 1 or not (0 <= chunk_index < total_chunks):
+    if (not relative_path or not _is_valid_sha256(expected_sha256)
+            or total_chunks < 1 or not (0 <= chunk_index < total_chunks)):
         return jsonify(error="missing/invalid form fields"), 400
 
     device_id = g.device_id
@@ -355,6 +378,44 @@ def dashcam_upload_chunk():
     return jsonify(status="complete", size_bytes=size_bytes), 201
 
 
+# ── Login rate limiting ────────────────────────────────────────────────────
+# In-memory failed-attempt limiter. The app runs as a single gunicorn worker
+# (threads=4), so process-local state is sufficient; a multi-worker deployment
+# would need a shared store (Redis/DB) instead. Keys are the client IP (from
+# Cloudflare's CF-Connecting-IP) plus the submitted username, so a distributed
+# brute-force can't bypass the per-IP cap by rotating source addresses.
+_LOGIN_FAIL_WINDOW_S = 900      # 15 minutes
+_LOGIN_FAIL_MAX = 8             # failed attempts per key before blocking
+_login_failures = defaultdict(list)
+_login_failures_lock = threading.Lock()
+
+
+def _login_failure_key_ip():
+    return (
+        request.headers.get("CF-Connecting-IP")
+        or (request.headers.get("X-Forwarded-For") or "").split(",")[0].strip()
+        or request.remote_addr
+        or "unknown"
+    )
+
+
+def _login_blocked(key):
+    if not key:
+        return False
+    now = time.time()
+    cutoff = now - _LOGIN_FAIL_WINDOW_S
+    with _login_failures_lock:
+        _login_failures[key] = [t for t in _login_failures[key] if t > cutoff]
+        return len(_login_failures[key]) >= _LOGIN_FAIL_MAX
+
+
+def _record_login_failure(key):
+    if not key:
+        return
+    with _login_failures_lock:
+        _login_failures[key].append(time.time())
+
+
 # ── Browser auth ───────────────────────────────────────────────────────────
 
 @app.get("/login")
@@ -376,10 +437,22 @@ def login_submit():
 
     username = request.form.get("username", "").strip()
     password = request.form.get("password", "")
+    client_ip = _login_failure_key_ip()
+
+    if _login_blocked(client_ip) or _login_blocked(f"user:{username}"):
+        csrf_token = secrets.token_hex(16)
+        session["csrf_token"] = csrf_token
+        flash("Too many failed attempts. Please try again later.", "error")
+        return render_template("login.html", csrf_token=csrf_token), 429
+
     user_row = db.get_user_by_username(username)
     if user_row and check_password(password, user_row["password_hash"]):
         login_user(User(user_row["id"], user_row["username"]), remember=False)
         return redirect(url_for("portal_index"))
+
+    _record_login_failure(client_ip)
+    if username:
+        _record_login_failure(f"user:{username}")
     flash("Invalid username or password.", "error")
     return redirect(url_for("login_page"))