From 24250e06dfdb30ef1cc58d81ebad5d747ae435c3 Mon Sep 17 00:00:00 2001 From: Mathieu Moran Date: Sat, 15 Aug 2026 18:28:55 -0400 Subject: [PATCH] security: validate sha256 digests, add login rate limiting, harden session cookies --- server/.env.example | 5 +++ server/app.py | 83 ++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 83 insertions(+), 5 deletions(-) diff --git a/server/.env.example b/server/.env.example index 6887a2b..3ff303f 100644 --- a/server/.env.example +++ b/server/.env.example @@ -24,6 +24,11 @@ MAX_DASHCAM_UPLOAD_MB=2048 # python -c "import secrets; print(secrets.token_hex(32))" SECRET_KEY=changeme_generate_a_real_secret_key +# Mark session cookies Secure (HTTPS-only). The portal is served over HTTPS +# via Cloudflare Tunnel, so leave this as 1 unless you also log in to the +# portal over plain HTTP on the LAN. +COOKIE_SECURE=1 + # The device name registered in the devices table - used by the portal to # scope all queries to the right Pi. DEVICE_NAME=yaris-2013 diff --git a/server/app.py b/server/app.py index e4b73d4..76b97ff 100644 --- a/server/app.py +++ b/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"))