diff --git a/app.py b/app.py index c3e06f8..c1de942 100644 --- a/app.py +++ b/app.py @@ -8,9 +8,11 @@ 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 @@ -18,6 +20,21 @@ 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 diff --git a/challenges/expert_vault.py b/challenges/expert_vault.py index 6e61ca9..3e02ffd 100644 --- a/challenges/expert_vault.py +++ b/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 diff --git a/tests/test_expert_unlock.py b/tests/test_expert_unlock.py new file mode 100644 index 0000000..1d0ded1 --- /dev/null +++ b/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