Harden security: password hashing, session persistence, input validation

- Admin password stored as scrypt hash instead of reversible encryption
- Session secret key persisted to file (survives container restarts)
- All API endpoints protected with @require_auth
- Timestamp/date parameters validated against format regex
- Security headers (X-Content-Type-Options, X-Frame-Options, etc.)
- Docker container runs as non-root user (uid 1000)
- 97 tests passing (15 new security tests)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: itsDNNS

CHANGELOG.md

@@ -4,6 +4,16 @@ All notable changes to this project will be documented in this file.
 
 Versioning: `YYYY-MM-DD.N` (date + sequential build number per day)
 
+## [2026-02-09.14]
+
+### Changed
+- **Password hashing**: Admin password stored as scrypt hash instead of reversible encryption
+- **Session key persisted**: Flask session secret key saved to file, sessions survive container restarts
+- **All API endpoints require auth**: `/api/calendar`, `/api/trends`, `/api/export`, `/api/snapshots`, `/api/snapshot/daily` now protected
+- **Input validation**: Timestamp and date parameters validated against format regex
+- **Security headers**: `X-Content-Type-Options`, `X-Frame-Options`, `X-XSS-Protection`, `Referrer-Policy` on all responses
+- **Non-root Docker user**: Container runs as `appuser` (uid 1000) instead of root
+
 ## [2026-02-09.13]
 
 ### Added

Dockerfile

@@ -3,6 +3,9 @@ WORKDIR /app
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 COPY app/ ./app/
+RUN adduser --disabled-password --gecos "" --uid 1000 appuser && \
+    mkdir -p /data && chown appuser:appuser /data
+USER appuser
 HEALTHCHECK --interval=60s --timeout=5s --retries=3 \
     CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8765/health')" || exit 1
 CMD ["python", "-m", "app.main"]

app/config.py

@@ -6,13 +6,15 @@
 import stat
 
 from cryptography.fernet import Fernet
+from werkzeug.security import generate_password_hash
 
 log = logging.getLogger("docsis.config")
 
 POLL_MIN = 60
 POLL_MAX = 3600
 
-SECRET_KEYS = {"fritz_password", "mqtt_password", "admin_password"}
+SECRET_KEYS = {"fritz_password", "mqtt_password"}
+HASH_KEYS = {"admin_password"}
 PASSWORD_MASK = "\u2022\u2022\u2022\u2022\u2022\u2022\u2022\u2022"
 
 DEFAULTS = {
@@ -127,6 +129,11 @@ def get(self, key, default=None):
             val = self._file_config[key]
             if key in INT_KEYS and not isinstance(val, int):
                 return int(val)
+            if key in HASH_KEYS:
+                # Return werkzeug hash as-is; legacy Fernet-encrypted values get decrypted
+                if val and (val.startswith("scrypt:") or val.startswith("pbkdf2:")):
+                    return val
+                return self._decrypt(val)
             if key in SECRET_KEYS:
                 return self._decrypt(val)
             return val
@@ -140,10 +147,16 @@ def save(self, data):
         os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
 
         # Don't overwrite passwords with the mask placeholder
-        for key in SECRET_KEYS:
+        for key in SECRET_KEYS | HASH_KEYS:
             if key in data and data[key] == PASSWORD_MASK:
                 del data[key]
 
+        # Hash password keys (admin_password) before storing
+        for key in HASH_KEYS:
+            if key in data and data[key]:
+                if not (data[key].startswith("scrypt:") or data[key].startswith("pbkdf2:")):
+                    data[key] = generate_password_hash(data[key])
+
         # Encrypt secret values before storing
         for key in SECRET_KEYS:
             if key in data and data[key]:
@@ -187,7 +200,7 @@ def get_all(self, mask_secrets=False):
         result = {}
         for key in DEFAULTS:
             val = self.get(key)
-            if mask_secrets and key in SECRET_KEYS and val:
+            if mask_secrets and key in (SECRET_KEYS | HASH_KEYS) and val:
                 result[key] = PASSWORD_MASK
             else:
                 result[key] = val

app/web.py

@@ -3,18 +3,24 @@
 import functools
 import logging
 import os
+import re
+import stat
 import time
 from datetime import datetime, timedelta
 
 from flask import Flask, render_template, request, jsonify, redirect, session, url_for
+from werkzeug.security import check_password_hash
 
 from .config import POLL_MIN, POLL_MAX, PASSWORD_MASK, SECRET_KEYS
 from .i18n import get_translations, LANGUAGES
 
 log = logging.getLogger("docsis.web")
 
 app = Flask(__name__, template_folder="templates")
-app.secret_key = os.environ.get("FLASK_SECRET_KEY", os.urandom(32))
+app.secret_key = os.urandom(32)  # overwritten by _init_session_key
+
+_TS_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$")
+_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 
 
 @app.template_filter("fmt_k")
@@ -63,11 +69,30 @@ def init_storage(storage):
     _storage = storage
 
 
+def _init_session_key(data_dir):
+    """Load or generate a persistent session secret key."""
+    key_path = os.path.join(data_dir, ".session_key")
+    if os.path.exists(key_path):
+        with open(key_path, "rb") as f:
+            app.secret_key = f.read()
+    else:
+        key = os.urandom(32)
+        os.makedirs(data_dir, exist_ok=True)
+        with open(key_path, "wb") as f:
+            f.write(key)
+        try:
+            os.chmod(key_path, stat.S_IRUSR | stat.S_IWUSR)
+        except OSError:
+            pass
+        app.secret_key = key
+
+
 def init_config(config_manager, on_config_changed=None):
     """Set the config manager and optional change callback."""
     global _config_manager, _on_config_changed
     _config_manager = config_manager
     _on_config_changed = on_config_changed
+    _init_session_key(config_manager.data_dir)
 
 
 def _auth_required():
@@ -100,7 +125,12 @@ def login():
     error = None
     if request.method == "POST":
         pw = request.form.get("password", "")
-        if pw == _config_manager.get("admin_password", ""):
+        stored = _config_manager.get("admin_password", "")
+        if stored.startswith(("scrypt:", "pbkdf2:")):
+            success = check_password_hash(stored, pw)
+        else:
+            success = (pw == stored)  # legacy plaintext / env var
+        if success:
             session["authenticated"] = True
             return redirect("/")
         error = t.get("login_failed", "Invalid password")
@@ -148,6 +178,8 @@ def index():
     conn_info = _state.get("connection_info") or {}
 
     ts = request.args.get("t")
+    if ts and not _TS_RE.match(ts):
+        return redirect("/")
     if ts and _storage:
         snapshot = _storage.get_snapshot(ts)
         if snapshot:
@@ -269,6 +301,7 @@ def api_test_mqtt():
 
 
 @app.route("/api/calendar")
+@require_auth
 def api_calendar():
     """Return dates that have snapshot data."""
     if _storage:
@@ -277,17 +310,21 @@ def api_calendar():
 
 
 @app.route("/api/snapshot/daily")
+@require_auth
 def api_snapshot_daily():
     """Return the daily snapshot closest to the configured snapshot_time."""
     date = request.args.get("date")
     if not date or not _storage:
         return jsonify(None)
+    if not _DATE_RE.match(date):
+        return jsonify({"error": "Invalid date format"}), 400
     target_time = _config_manager.get("snapshot_time", "06:00") if _config_manager else "06:00"
     snap = _storage.get_daily_snapshot(date, target_time)
     return jsonify(snap)
 
 
 @app.route("/api/trends")
+@require_auth
 def api_trends():
     """Return trend data for a date range.
     ?range=day|week|month&date=YYYY-MM-DD (date defaults to today)."""
@@ -321,6 +358,7 @@ def api_trends():
 
 
 @app.route("/api/export")
+@require_auth
 def api_export():
     """Generate a structured markdown report for LLM analysis."""
     analysis = _state.get("analysis")
@@ -407,13 +445,23 @@ def api_export():
 
 
 @app.route("/api/snapshots")
+@require_auth
 def api_snapshots():
     """Return list of available snapshot timestamps."""
     if _storage:
         return jsonify(_storage.get_snapshot_list())
     return jsonify([])
 
 
+@app.after_request
+def add_security_headers(response):
+    response.headers["X-Content-Type-Options"] = "nosniff"
+    response.headers["X-Frame-Options"] = "DENY"
+    response.headers["X-XSS-Protection"] = "1; mode=block"
+    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+    return response
+
+
 @app.route("/health")
 def health():
     """Simple health check endpoint."""

tests/test_auth.py

@@ -105,3 +105,24 @@ def test_api_config_requires_auth(self, auth_client):
             content_type="application/json",
         )
         assert resp.status_code == 302
+
+    def test_api_calendar_requires_auth(self, auth_client):
+        resp = auth_client.get("/api/calendar")
+        assert resp.status_code == 302
+
+    def test_api_snapshots_requires_auth(self, auth_client):
+        resp = auth_client.get("/api/snapshots")
+        assert resp.status_code == 302
+
+    def test_api_export_requires_auth(self, auth_client):
+        resp = auth_client.get("/api/export")
+        assert resp.status_code == 302
+
+    def test_api_trends_requires_auth(self, auth_client):
+        resp = auth_client.get("/api/trends")
+        assert resp.status_code == 302
+
+    def test_password_hashed_not_plaintext(self, auth_config):
+        stored = auth_config.get("admin_password")
+        assert stored != "secret123"
+        assert stored.startswith(("scrypt:", "pbkdf2:"))

tests/test_config.py

@@ -3,7 +3,7 @@
 import json
 import os
 import pytest
-from app.config import ConfigManager, DEFAULTS, SECRET_KEYS, PASSWORD_MASK
+from app.config import ConfigManager, DEFAULTS, SECRET_KEYS, HASH_KEYS, PASSWORD_MASK
 
 
 @pytest.fixture
@@ -93,6 +93,30 @@ def test_get_all_shows_secrets(self, config):
         all_config = config.get_all(mask_secrets=False)
         assert all_config["fritz_password"] == "secret"
 
+    def test_admin_password_hashed_at_rest(self, config, tmp_data_dir):
+        config.save({"admin_password": "admin123"})
+        with open(os.path.join(tmp_data_dir, "config.json")) as f:
+            raw = json.load(f)
+        assert raw["admin_password"] != "admin123"
+        assert raw["admin_password"].startswith(("scrypt:", "pbkdf2:"))
+
+    def test_admin_password_hash_returned(self, config):
+        config.save({"admin_password": "admin123"})
+        stored = config.get("admin_password")
+        assert stored.startswith(("scrypt:", "pbkdf2:"))
+
+    def test_admin_password_mask_not_saved(self, tmp_data_dir):
+        config = ConfigManager(tmp_data_dir)
+        config.save({"admin_password": "original"})
+        hash1 = config.get("admin_password")
+        config.save({"admin_password": PASSWORD_MASK, "fritz_user": "updated"})
+        assert config.get("admin_password") == hash1
+
+    def test_admin_password_masked_in_get_all(self, config):
+        config.save({"admin_password": "secret"})
+        all_config = config.get_all(mask_secrets=True)
+        assert all_config["admin_password"] == PASSWORD_MASK
+
 
 class TestConfigEnvOverride:
     def test_env_overrides_file(self, tmp_data_dir, monkeypatch):

tests/test_web.py

@@ -188,6 +188,52 @@ def test_save_no_data(self, client):
         assert resp.status_code in (400, 500)
 
 
+class TestSecurityHeaders:
+    def test_headers_present(self, client, sample_analysis):
+        update_state(analysis=sample_analysis)
+        resp = client.get("/")
+        assert resp.headers["X-Content-Type-Options"] == "nosniff"
+        assert resp.headers["X-Frame-Options"] == "DENY"
+        assert resp.headers["X-XSS-Protection"] == "1; mode=block"
+        assert resp.headers["Referrer-Policy"] == "strict-origin-when-cross-origin"
+
+    def test_headers_on_health(self, client):
+        resp = client.get("/health")
+        assert resp.headers["X-Content-Type-Options"] == "nosniff"
+
+
+class TestTimestampValidation:
+    def test_invalid_timestamp_rejected(self, client, sample_analysis):
+        update_state(analysis=sample_analysis)
+        resp = client.get("/?t=../../etc/passwd")
+        assert resp.status_code == 302
+        assert resp.headers["Location"] == "/"
+
+    def test_valid_timestamp_accepted(self, client, sample_analysis):
+        update_state(analysis=sample_analysis)
+        # No storage, so snapshot lookup returns None and falls through to live view
+        resp = client.get("/?t=2026-01-01T06:00:00")
+        assert resp.status_code == 200
+
+
+class TestSessionKeyPersistence:
+    def test_session_key_file_created(self, tmp_path):
+        data_dir = str(tmp_path / "data_sk")
+        mgr = ConfigManager(data_dir)
+        init_config(mgr)
+        import os
+        assert os.path.exists(os.path.join(data_dir, ".session_key"))
+
+    def test_session_key_persisted(self, tmp_path):
+        data_dir = str(tmp_path / "data_sk2")
+        mgr = ConfigManager(data_dir)
+        init_config(mgr)
+        key1 = app.secret_key
+        # Re-init should load same key
+        init_config(mgr)
+        assert app.secret_key == key1
+
+
 class TestFormatK:
     def test_large_number(self):
         from app.web import format_k