fix: audit expert vault unlock failures

app.py

@@ -8,16 +8,33 @@
 from __future__ import annotations
 
 import datetime
+import hashlib
 import json
 import os
 import uuid
+from logging.config import dictConfig
 from flask import Flask, render_template, request, jsonify, session, Response, abort
 
 import config
 import card_svg
 from challenges import load_all, get, core_labs, advanced_labs
 from challenges import expert_vault
 
+dictConfig({
+    "version": 1,
+    "disable_existing_loggers": False,
+    "handlers": {
+        "console": {
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stderr",
+        },
+    },
+    "root": {
+        "level": "INFO",
+        "handlers": ["console"],
+    },
+})
+
 app = Flask(__name__)
 app.secret_key = config.SECRET_KEY
 CHALLENGES = load_all()
@@ -44,6 +61,14 @@ def save_progress():
     os.replace(tmp, config.DATA_FILE)
 
 
+def log_expert_unlock(result: str) -> None:
+    """Record an unlock attempt without logging the access key or player name."""
+    sid = session.get("sid", "")
+    sid_tag = hashlib.sha256(sid.encode()).hexdigest()[:12] if sid else "none"
+    app.logger.info("expert_unlock result=%s session=%s remote_addr=%s", result, sid_tag,
+                    request.remote_addr or "unknown")
+
+
 _load_progress()
 
 
@@ -236,10 +261,18 @@ def unlock_expert():
     key = (request.get_json(force=True).get("key", "") or "").strip()
     if not key:
         return jsonify(ok=False, error="Enter the access key."), 400
-    if expert_vault.try_unlock(key):
+    try:
+        unlocked = expert_vault.try_unlock(key)
+    except expert_vault.VaultLoadError:
+        log_expert_unlock("error")
+        app.logger.exception("expert vault could not be loaded")
+        return jsonify(ok=False, error="Expert vault is temporarily unavailable."), 500
+    if unlocked:
         p["expert_unlocked"] = True
         save_progress()
+        log_expert_unlock("success")
         return jsonify(ok=True, count=expert_vault.expert_count())
+    log_expert_unlock("invalid_key")
     return jsonify(ok=False, error="Invalid access key."), 403
 
 

challenges/expert_vault.py

@@ -10,6 +10,8 @@
 from __future__ import annotations
 
 import base64
+import hashlib
+import hmac
 import json
 import os
 import re
@@ -22,11 +24,19 @@
 _HERE = os.path.dirname(os.path.abspath(__file__))
 _ENC = os.path.join(_HERE, "expert.enc")
 _META = os.path.join(_HERE, "expert_meta.json")
+# Updated alongside the vault bundle by the release process. Keeping these values
+# in code makes an unexpected vault-file change distinguishable from a wrong key.
+EXPECTED_ENC_SHA256 = "5e17bbd9e6525490b47c566b678ca60f42cd0330d3b1b89e0855e2ae71deb551"
+EXPECTED_META_SHA256 = "251c75cfcf8838cab57955336a7c785dd1a16b115d6d751437829a6dc18c1f80"
 
 _SPECS: list[dict] | None = None          # populated only after a valid unlock
 _CHALLENGES: dict[str, "DeclarativeChallenge"] = {}
 
 
+class VaultLoadError(RuntimeError):
+    """The encrypted vault could not be loaded because of a server-side fault."""
+
+
 class DeclarativeChallenge(Challenge):
     """A challenge whose vulnerable behaviour is a list of match-rules (from the vault)."""
     def __init__(self, spec: dict):
@@ -60,16 +70,30 @@ def try_unlock(access_key: str) -> bool:
         # already decrypted this process; verify the supplied key still matches
         return _verify(access_key)
     if not (os.path.exists(_ENC) and os.path.exists(_META)):
-        return False
-    meta = json.load(open(_META))
-    salt = base64.b64decode(meta["salt"])
-    fkey = _derive(access_key.strip(), salt, meta["iterations"])
+        raise VaultLoadError("expert vault files are missing")
     try:
-        plain = Fernet(fkey).decrypt(open(_ENC, "rb").read())
-    except (InvalidToken, Exception):
+        with open(_META, "rb") as meta_file:
+            meta_bytes = meta_file.read()
+        with open(_ENC, "rb") as encrypted_file:
+            encrypted = encrypted_file.read()
+        meta_digest = hashlib.sha256(meta_bytes).hexdigest()
+        if not hmac.compare_digest(meta_digest, EXPECTED_META_SHA256):
+            raise VaultLoadError("expert vault metadata integrity check failed")
+        encrypted_digest = hashlib.sha256(encrypted).hexdigest()
+        if not hmac.compare_digest(encrypted_digest, EXPECTED_ENC_SHA256):
+            raise VaultLoadError("expert vault ciphertext integrity check failed")
+        meta = json.loads(meta_bytes)
+        salt = base64.b64decode(meta["salt"], validate=True)
+        fkey = _derive(access_key.strip(), salt, meta["iterations"])
+        plain = Fernet(fkey).decrypt(encrypted)
+        specs = json.loads(plain)
+        challenges = {s["id"]: DeclarativeChallenge(s) for s in specs}
+    except InvalidToken:
         return False
-    _SPECS = json.loads(plain)
-    _CHALLENGES = {s["id"]: DeclarativeChallenge(s) for s in _SPECS}
+    except (OSError, ValueError, KeyError, TypeError) as exc:
+        raise VaultLoadError("expert vault could not be loaded") from exc
+    _SPECS = specs
+    _CHALLENGES = challenges
     globals()["_VALID_KEY"] = access_key.strip()
     return True
 

tests/test_expert_unlock.py

@@ -0,0 +1,109 @@
+"""Tests for Expert vault failure handling and unlock audit events."""
+import json
+import logging
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, ROOT)
+
+import config
+config.DATA_FILE = os.path.join(tempfile.gettempdir(), "llmvault_expert_unlock_test_progress.json")
+
+import app as server
+from challenges import expert_vault
+
+
+@pytest.fixture(autouse=True)
+def reset_vault_state(monkeypatch):
+    """Keep process-global vault and player state isolated between tests."""
+    monkeypatch.setattr(expert_vault, "_SPECS", None)
+    monkeypatch.setattr(expert_vault, "_CHALLENGES", {})
+    monkeypatch.delitem(expert_vault.__dict__, "_VALID_KEY", raising=False)
+    server.PROGRESS.clear()
+
+
+def test_wrong_key_returns_false():
+    assert expert_vault.try_unlock("definitely-not-the-key") is False
+
+
+@pytest.mark.parametrize(
+    ("meta_contents", "ciphertext"),
+    [
+        (None, b"ciphertext"),
+        ("{", b"ciphertext"),
+        (json.dumps({"salt": "not base64!", "iterations": 1}), b"ciphertext"),
+        (json.dumps({"salt": "YWJjZA==", "iterations": 1}), b"truncated"),
+    ],
+)
+def test_vault_load_failures_return_controlled_error(tmp_path, monkeypatch, meta_contents, ciphertext):
+    enc = tmp_path / "expert.enc"
+    meta = tmp_path / "expert_meta.json"
+    enc.write_bytes(ciphertext)
+    if meta_contents is not None:
+        meta.write_text(meta_contents)
+    monkeypatch.setattr(expert_vault, "_ENC", str(enc))
+    monkeypatch.setattr(expert_vault, "_META", str(meta))
+    monkeypatch.setattr(server, "prereq_done", lambda _player: True)
+
+    response = server.app.test_client().post("/api/unlock-expert", json={"key": "any-key"})
+    assert response.status_code == 500
+    assert response.get_json() == {"ok": False, "error": "Expert vault is temporarily unavailable."}
+
+
+@pytest.mark.parametrize("changed_field", [("salt", "YWJjZA=="), ("iterations", 999_999_999)])
+def test_modified_valid_metadata_is_rejected_before_key_derivation(tmp_path, monkeypatch, changed_field):
+    field, value = changed_field
+    meta = json.loads(Path(expert_vault._META).read_text())
+    meta[field] = value
+    enc = tmp_path / "expert.enc"
+    enc.write_bytes(Path(expert_vault._ENC).read_bytes())
+    meta_path = tmp_path / "expert_meta.json"
+    meta_path.write_text(json.dumps(meta))
+    monkeypatch.setattr(expert_vault, "_ENC", str(enc))
+    monkeypatch.setattr(expert_vault, "_META", str(meta_path))
+    monkeypatch.setattr(server, "prereq_done", lambda _player: True)
+    derived = False
+
+    def fail_if_derived(*_args):
+        nonlocal derived
+        derived = True
+        raise AssertionError("metadata integrity must be verified before key derivation")
+
+    monkeypatch.setattr(expert_vault, "_derive", fail_if_derived)
+    response = server.app.test_client().post("/api/unlock-expert", json={"key": "any-key"})
+
+    assert response.status_code == 500
+    assert derived is False
+
+
+def test_unlock_events_are_audited_without_secrets(monkeypatch, caplog):
+    monkeypatch.setattr(server, "prereq_done", lambda _player: True)
+    client = server.app.test_client()
+    player_name = "Sensitive Player Name"
+    access_key = "do-not-log-this-access-key"
+    client.post("/api/setname", json={"name": player_name})
+    caplog.set_level(logging.INFO, logger=server.app.logger.name)
+
+    monkeypatch.setattr(expert_vault, "try_unlock", lambda _key: True)
+    assert client.post("/api/unlock-expert", json={"key": access_key}).status_code == 200
+
+    monkeypatch.setattr(expert_vault, "try_unlock", lambda _key: False)
+    assert client.post("/api/unlock-expert", json={"key": access_key}).status_code == 403
+
+    def unavailable(_key):
+        raise expert_vault.VaultLoadError("test vault failure")
+
+    monkeypatch.setattr(expert_vault, "try_unlock", unavailable)
+    assert client.post("/api/unlock-expert", json={"key": access_key}).status_code == 500
+
+    messages = "\n".join(record.getMessage() for record in caplog.records)
+    assert "expert_unlock result=success" in messages
+    assert "expert_unlock result=invalid_key" in messages
+    assert "expert_unlock result=error" in messages
+    assert access_key not in messages
+    assert player_name[:10] not in messages