diff --git a/CMakeLists.txt b/CMakeLists.txt index 410d30e..e4ecdc9 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -225,6 +225,7 @@ include_directories( # Create flapi-lib add_library(flapi-lib STATIC src/api_server.cpp + src/audit_logger.cpp src/auth_middleware.cpp src/cache_manager.cpp src/database_manager_cache_adapter.cpp diff --git a/src/audit_logger.cpp b/src/audit_logger.cpp new file mode 100644 index 0000000..9b09ec8 --- /dev/null +++ b/src/audit_logger.cpp @@ -0,0 +1,120 @@ +#include "audit_logger.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace flapi { + +namespace { + +std::string randHex(int len) { + static thread_local std::mt19937_64 rng(std::random_device{}()); + std::uniform_int_distribution dist(0, 15); + static const char* kHex = "0123456789abcdef"; + std::string out; + out.reserve(len); + for (int i = 0; i < len; ++i) { + out.push_back(kHex[dist(rng)]); + } + return out; +} + +} // namespace + +AuditLogger::AuditLogger(AuditConfig config) : config_(std::move(config)) { + if (!config_.enabled) { + return; + } + if (config_.sink == "stdout") { + sink_stream_ = &std::cout; + } else if (config_.sink == "file") { + auto file = std::make_unique(config_.path, std::ios::app); + if (!file->is_open()) { + throw std::runtime_error("audit log: cannot open " + config_.path); + } + file_stream_ = std::move(file); + sink_stream_ = file_stream_.get(); + } else if (config_.sink == "null") { + sink_stream_ = nullptr; + } else { + throw std::runtime_error("audit log: unknown sink '" + config_.sink + "'"); + } +} + +AuditLogger::~AuditLogger() = default; + +void AuditLogger::log(AuditEvent event) { + if (!config_.enabled) { + return; + } + if (event.timestamp.empty()) { + event.timestamp = nowIso8601(); + } + if (event.request_id.empty()) { + event.request_id = generateRequestId(); + } + + const std::string line = serialiseEvent(event); + + if (sink_stream_ == nullptr) { + return; // null sink — no I/O + } + + std::lock_guard guard(write_mutex_); + (*sink_stream_) << line << '\n'; + sink_stream_->flush(); +} + +std::string AuditLogger::nowIso8601() { + const auto now = std::chrono::system_clock::now(); + const auto now_t = std::chrono::system_clock::to_time_t(now); + const auto micros = std::chrono::duration_cast( + now.time_since_epoch()).count() % 1'000'000; + std::tm tm_buf{}; +#ifdef _WIN32 + gmtime_s(&tm_buf, &now_t); +#else + gmtime_r(&now_t, &tm_buf); +#endif + std::ostringstream oss; + oss << std::put_time(&tm_buf, "%Y-%m-%dT%H:%M:%S") + << '.' << std::setfill('0') << std::setw(6) << micros << 'Z'; + return oss.str(); +} + +std::string AuditLogger::generateRequestId() { + // 16 hex chars — short enough for logs, wide enough to avoid collisions + // within any realistic flapi deployment. + return "req-" + randHex(16); +} + +std::string AuditLogger::serialiseEvent(const AuditEvent& event) const { + crow::json::wvalue line; + line["timestamp"] = event.timestamp; + line["request_id"] = event.request_id; + line["principal"] = event.principal; + line["method"] = event.method; + line["target"] = event.target; + line["status"] = event.status; + line["row_count"] = event.row_count; + line["latency_ms"] = event.latency_ms; + + crow::json::wvalue params = crow::json::wvalue::object(); + for (const auto& [key, value] : event.params) { + if (config_.redact_keys.count(key) > 0) { + params[key] = ""; + } else { + params[key] = value; + } + } + line["params"] = std::move(params); + return line.dump(); +} + +} // namespace flapi diff --git a/src/config_manager.cpp b/src/config_manager.cpp index 5f7bb0c..15423d8 100644 --- a/src/config_manager.cpp +++ b/src/config_manager.cpp @@ -121,6 +121,7 @@ void ConfigManager::parseMainConfig() { parseDuckDBConfig(); parseDuckLakeConfig(); parseMCPConfig(); + parseAuditConfig(); parseStorageConfig(); parseTemplateConfig(); parseGlobalHeartbeatConfig(); @@ -253,6 +254,42 @@ void ConfigManager::parseDuckLakeConfig() { } // Storage configuration methods +std::shared_ptr ConfigManager::getAuditLogger() { + // Eagerly built once at the end of parseAuditConfig(); the AuditLogger + // itself owns the write mutex so this method is a simple accessor. + if (!audit_logger_) { + audit_logger_ = std::make_shared(audit_config); + } + return audit_logger_; +} + +void ConfigManager::parseAuditConfig() { + CROW_LOG_INFO << "Parsing audit configuration"; + audit_config = AuditConfig{}; // Reset to defaults (enabled=false) + + if (!config["audit"]) { + CROW_LOG_DEBUG << "Audit configuration not found, using defaults (enabled=false)"; + return; + } + + auto audit_node = config["audit"]; + audit_config.enabled = safeGet(audit_node, "enabled", "audit.enabled", false); + audit_config.sink = safeGet(audit_node, "sink", "audit.sink", "stdout"); + audit_config.path = safeGet(audit_node, "path", "audit.path", ""); + + if (audit_node["redact"]) { + for (const auto& key_node : audit_node["redact"]) { + audit_config.redact_keys.insert(key_node.as()); + } + } + + CROW_LOG_DEBUG << "Audit enabled: " << (audit_config.enabled ? "true" : "false"); + CROW_LOG_DEBUG << "Audit sink: " << audit_config.sink; + if (!audit_config.path.empty()) { + CROW_LOG_DEBUG << "Audit path: " << audit_config.path; + } +} + void ConfigManager::parseStorageConfig() { CROW_LOG_INFO << "Parsing storage configuration"; storage_config = StorageConfig{}; // Reset to defaults diff --git a/src/include/audit_logger.hpp b/src/include/audit_logger.hpp new file mode 100644 index 0000000..280f2f1 --- /dev/null +++ b/src/include/audit_logger.hpp @@ -0,0 +1,61 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace flapi { + +struct AuditConfig { + bool enabled = false; + std::string sink = "stdout"; // "stdout" | "file" | "null" + std::string path; // used when sink == "file" + std::unordered_set redact_keys; // params with these keys are masked +}; + +struct AuditEvent { + std::string timestamp; // auto-filled if empty + std::string request_id; // auto-filled if empty + std::string principal = "anonymous"; // username, or "anonymous" when unauthenticated + std::string method; // "GET", "POST", "tools/call", etc. + std::string target; // url path or tool name + std::string status; // "success", "denied", "error:" — free-form + std::int64_t row_count = -1; // -1 when not applicable (e.g. denial) + std::int64_t latency_ms = -1; // wall-clock elapsed + std::map params; // already-redacted-by-caller is fine +}; + +// Append-only JSONL audit logger. Construct one per server; share by +// std::shared_ptr. The logger is thread-safe — call sites can race +// without coordination. +// +// Lifecycle is owned by whatever constructs it (typically the server +// bootstrap in main.cpp / APIServer). Writers SHOULD NOT bypass log() +// for any reason; redaction happens inside. +class AuditLogger { +public: + explicit AuditLogger(AuditConfig config); + ~AuditLogger(); + + AuditLogger(const AuditLogger&) = delete; + AuditLogger& operator=(const AuditLogger&) = delete; + + void log(AuditEvent event); + bool isEnabled() const { return config_.enabled; } + const AuditConfig& config() const { return config_; } + +private: + AuditConfig config_; + std::mutex write_mutex_; + std::unique_ptr file_stream_; + std::ostream* sink_stream_ = nullptr; // non-owning view onto the active sink + + static std::string nowIso8601(); + static std::string generateRequestId(); + std::string serialiseEvent(const AuditEvent& event) const; +}; + +} // namespace flapi diff --git a/src/include/config_manager.hpp b/src/include/config_manager.hpp index cdb76df..9742934 100644 --- a/src/include/config_manager.hpp +++ b/src/include/config_manager.hpp @@ -10,9 +10,11 @@ #include #include #include +#include #include #include +#include "audit_logger.hpp" #include "route_translator.hpp" #include "extended_yaml_parser.hpp" #include "path_utils.hpp" @@ -514,6 +516,12 @@ class ConfigManager { const DuckLakeConfig& getDuckLakeConfig() const { return ducklake_config; } const MCPConfig& getMCPConfig() const { return mcp_config; } const StorageConfig& getStorageConfig() const { return storage_config; } + const AuditConfig& getAuditConfig() const { return audit_config; } + + // Process-wide audit sink. Initialised lazily on first access from the + // current AuditConfig; shared across REST and MCP handlers so every + // request lands in the same JSONL stream. + std::shared_ptr getAuditLogger(); bool isTelemetryEnabled() const { return telemetry_enabled; } // Load MCP server instructions (inline or from file) @@ -575,6 +583,8 @@ class ConfigManager { DuckLakeConfig ducklake_config; MCPConfig mcp_config; StorageConfig storage_config; + AuditConfig audit_config; + std::shared_ptr audit_logger_; bool telemetry_enabled = true; ExtendedYamlParser yaml_parser; @@ -597,6 +607,7 @@ class ConfigManager { void parseTemplateConfig(); void parseDuckLakeConfig(); void parseMCPConfig(); + void parseAuditConfig(); void parseStorageConfig(); void parseEndpointConfig(const std::filesystem::path& config_file); void parseEndpointRequestFields(const YAML::Node& endpoint_config, EndpointConfig& endpoint); diff --git a/src/include/mcp_tool_handler.hpp b/src/include/mcp_tool_handler.hpp index dccea17..df63799 100644 --- a/src/include/mcp_tool_handler.hpp +++ b/src/include/mcp_tool_handler.hpp @@ -6,6 +6,7 @@ #include #include +#include "audit_logger.hpp" #include "config_manager.hpp" #include "database_manager.hpp" #include "sql_template_processor.hpp" @@ -67,6 +68,7 @@ QueryResult executeQueryWithEndpoint(const EndpointConfig& endpoint_config, std::shared_ptr config_manager; std::shared_ptr validator; std::unique_ptr sql_processor; + std::shared_ptr audit_logger; }; } // namespace flapi diff --git a/src/mcp_tool_handler.cpp b/src/mcp_tool_handler.cpp index bfaaf2d..62520b2 100644 --- a/src/mcp_tool_handler.cpp +++ b/src/mcp_tool_handler.cpp @@ -1,4 +1,5 @@ #include "mcp_tool_handler.hpp" +#include #include #include @@ -8,20 +9,60 @@ MCPToolHandler::MCPToolHandler(std::shared_ptr db_manager, std::shared_ptr config_manager) : db_manager(db_manager), config_manager(config_manager), validator(std::make_shared()), - sql_processor(std::make_unique(config_manager)) + sql_processor(std::make_unique(config_manager)), + audit_logger(config_manager->getAuditLogger()) { } MCPToolExecutionResult MCPToolHandler::executeTool(const MCPToolCallRequest& request) { + const auto audit_started_at = std::chrono::steady_clock::now(); + const auto emit_audit = [&](const std::string& status, std::int64_t row_count) { + if (!audit_logger || !audit_logger->isEnabled()) { + return; + } + AuditEvent ev; + ev.method = "tools/call"; + ev.target = request.tool_name; + ev.status = status; + ev.row_count = row_count; + ev.latency_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - audit_started_at).count(); + auto principal_it = request.context.find("auth.username"); + if (principal_it != request.context.end() && !principal_it->second.empty()) { + ev.principal = principal_it->second; + } + // Mirror the JSON arguments into the audit params map as strings; the + // logger applies the configured redaction. We re-wrap each rvalue + // child as a wvalue so .dump() handles all JSON types uniformly. + if (request.arguments.t() == crow::json::type::Object) { + auto parsed = crow::json::load(request.arguments.dump()); + if (parsed) { + for (const auto& key : parsed.keys()) { + crow::json::wvalue tmp(parsed[key]); + std::string serialised = tmp.dump(); + // Strip surrounding quotes for plain string values so the + // audit log doesn't contain visually doubled quoting. + if (serialised.size() >= 2 && serialised.front() == '"' && serialised.back() == '"') { + serialised = serialised.substr(1, serialised.size() - 2); + } + ev.params[key] = std::move(serialised); + } + } + } + audit_logger->log(std::move(ev)); + }; + try { // Get the endpoint configuration by tool name const EndpointConfig* endpoint_config = getEndpointConfigByToolName(request.tool_name); if (!endpoint_config) { + emit_audit("error:tool_not_found", -1); return createErrorResult("Tool not found: " + request.tool_name); } // Validate arguments if (!validateToolArguments(request.tool_name, request.arguments)) { + emit_audit("error:invalid_arguments", -1); return createErrorResult("Invalid arguments for tool: " + request.tool_name); } @@ -65,6 +106,7 @@ MCPToolExecutionResult MCPToolHandler::executeTool(const MCPToolCallRequest& req metadata["rows_affected"] = std::to_string(write_result.rows_affected); metadata["execution_time_ms"] = "0"; // Simplified + emit_audit("success", static_cast(write_result.rows_affected)); return createSuccessResult(write_response.dump(), metadata); } else { // Execute read query @@ -81,9 +123,11 @@ MCPToolExecutionResult MCPToolHandler::executeTool(const MCPToolCallRequest& req metadata["query_rows"] = std::to_string(query_result.data.size()); metadata["execution_time_ms"] = "0"; // Simplified + emit_audit("success", static_cast(query_result.data.size())); return createSuccessResult(formatted_result, metadata); } } catch (const std::exception& e) { + emit_audit("error:exception", -1); return createErrorResult("Tool execution error: " + std::string(e.what())); } } diff --git a/test/cpp/CMakeLists.txt b/test/cpp/CMakeLists.txt index 5c196c6..03f0a2e 100644 --- a/test/cpp/CMakeLists.txt +++ b/test/cpp/CMakeLists.txt @@ -2,6 +2,7 @@ find_package(Catch2 3 REQUIRED) add_executable(flapi_tests main.cpp + audit_logger_test.cpp auth_middleware_test.cpp config_manager_test.cpp config_manager_yaml_validation_test.cpp diff --git a/test/cpp/audit_logger_test.cpp b/test/cpp/audit_logger_test.cpp new file mode 100644 index 0000000..5644e74 --- /dev/null +++ b/test/cpp/audit_logger_test.cpp @@ -0,0 +1,219 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "audit_logger.hpp" + +namespace flapi { +namespace test { + +namespace { + +namespace fs = std::filesystem; + +class TempAuditFile { +public: + TempAuditFile() + : path_(fs::temp_directory_path() / ("flapi_audit_test_" + + std::to_string(::rand()) + ".jsonl")) {} + ~TempAuditFile() { + if (fs::exists(path_)) { + fs::remove(path_); + } + } + fs::path path() const { return path_; } + std::vector readLines() const { + std::vector lines; + std::ifstream f(path_); + std::string line; + while (std::getline(f, line)) { + if (!line.empty()) { + lines.push_back(line); + } + } + return lines; + } +private: + fs::path path_; +}; + +AuditEvent sampleEvent() { + AuditEvent ev; + ev.principal = "alice"; + ev.target = "tools/call:customer_lookup"; + ev.method = "tools/call"; + ev.status = "success"; + ev.row_count = 7; + ev.latency_ms = 12; + ev.params = {{"id", "42"}, {"token", "secret123"}}; + return ev; +} + +} // namespace + +TEST_CASE("AuditLogger: disabled config is a no-op even when path is set", + "[security][audit]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = false; + cfg.sink = "file"; + cfg.path = sink.path().string(); + + AuditLogger logger(cfg); + logger.log(sampleEvent()); + + // The file must not even be created when the logger is disabled. + REQUIRE_FALSE(fs::exists(sink.path())); +} + +TEST_CASE("AuditLogger: file sink emits one JSONL line per event", + "[security][audit]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "file"; + cfg.path = sink.path().string(); + + AuditLogger logger(cfg); + logger.log(sampleEvent()); + logger.log(sampleEvent()); + + auto lines = sink.readLines(); + REQUIRE(lines.size() == 2); + + auto parsed = crow::json::load(lines[0]); + REQUIRE(parsed); + REQUIRE(parsed["principal"].s() == std::string("alice")); + REQUIRE(parsed["target"].s() == std::string("tools/call:customer_lookup")); + REQUIRE(parsed["method"].s() == std::string("tools/call")); + REQUIRE(parsed["status"].s() == std::string("success")); + REQUIRE(parsed["row_count"].i() == 7); + REQUIRE(parsed["latency_ms"].i() == 12); +} + +TEST_CASE("AuditLogger: redact list masks listed param keys", + "[security][audit]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "file"; + cfg.path = sink.path().string(); + cfg.redact_keys = {"token", "password"}; + + AuditLogger logger(cfg); + logger.log(sampleEvent()); + + auto lines = sink.readLines(); + REQUIRE(lines.size() == 1); + auto parsed = crow::json::load(lines[0]); + REQUIRE(parsed); + REQUIRE(parsed["params"]["id"].s() == std::string("42")); + // The literal redaction marker must replace the secret value. + REQUIRE(parsed["params"]["token"].s() == std::string("")); + // Original secret must not appear anywhere in the line. + REQUIRE(lines[0].find("secret123") == std::string::npos); +} + +TEST_CASE("AuditLogger: every event carries timestamp and request_id", + "[security][audit]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "file"; + cfg.path = sink.path().string(); + + AuditLogger logger(cfg); + logger.log(sampleEvent()); + + auto lines = sink.readLines(); + REQUIRE(lines.size() == 1); + auto parsed = crow::json::load(lines[0]); + REQUIRE(parsed); + REQUIRE_FALSE(std::string(parsed["timestamp"].s()).empty()); + REQUIRE_FALSE(std::string(parsed["request_id"].s()).empty()); + // Timestamp must look ISO 8601-ish — at minimum start with four digits and a dash. + const std::string ts = parsed["timestamp"].s(); + REQUIRE(ts.size() >= 5); + REQUIRE(ts[4] == '-'); +} + +TEST_CASE("AuditLogger: explicit request_id is preserved", + "[security][audit]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "file"; + cfg.path = sink.path().string(); + + AuditLogger logger(cfg); + auto ev = sampleEvent(); + ev.request_id = "req-deadbeef"; + logger.log(ev); + + auto lines = sink.readLines(); + REQUIRE(lines.size() == 1); + auto parsed = crow::json::load(lines[0]); + REQUIRE(parsed); + REQUIRE(parsed["request_id"].s() == std::string("req-deadbeef")); +} + +TEST_CASE("AuditLogger: concurrent writes produce well-formed JSONL", + "[security][audit][threading]") { + TempAuditFile sink; + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "file"; + cfg.path = sink.path().string(); + + AuditLogger logger(cfg); + + constexpr int kThreads = 8; + constexpr int kEventsPerThread = 25; + std::vector workers; + workers.reserve(kThreads); + for (int t = 0; t < kThreads; ++t) { + workers.emplace_back([&logger, t]() { + for (int i = 0; i < kEventsPerThread; ++i) { + AuditEvent ev; + ev.principal = "thread-" + std::to_string(t); + ev.target = "tool-" + std::to_string(i); + ev.method = "tools/call"; + ev.status = "success"; + ev.row_count = i; + ev.latency_ms = 1; + logger.log(ev); + } + }); + } + for (auto& w : workers) { + w.join(); + } + + auto lines = sink.readLines(); + REQUIRE(lines.size() == static_cast(kThreads * kEventsPerThread)); + // Every line must be valid JSON in isolation; concurrent writes + // must not interleave inside a single line. + for (const auto& l : lines) { + auto parsed = crow::json::load(l); + REQUIRE(parsed); + REQUIRE(parsed["method"].s() == std::string("tools/call")); + } +} + +TEST_CASE("AuditLogger: null sink is honoured (no I/O)", + "[security][audit]") { + // "null" sink is the no-op writer — useful for tests that exercise the + // production code path without an audit file. + AuditConfig cfg; + cfg.enabled = true; + cfg.sink = "null"; + AuditLogger logger(cfg); + REQUIRE_NOTHROW(logger.log(sampleEvent())); +} + +} // namespace test +} // namespace flapi diff --git a/test/integration/test_audit_log.py b/test/integration/test_audit_log.py new file mode 100644 index 0000000..de3f01e --- /dev/null +++ b/test/integration/test_audit_log.py @@ -0,0 +1,290 @@ +"""End-to-end tests for the JSONL audit log (issue #23, W1.3). + +Boots a real flapi server with the audit log enabled and pointed at a +file in a temp directory. After issuing one or more MCP tool calls, +the audit file must contain one well-formed JSON line per call with +the expected fields and redaction applied to configured keys. + +Marked `standalone_server` so the conftest autouse fixture does not +spin up the shared api_configuration server. Skips cleanly on local +environments where flapi cannot boot due to the v1.5.1/v1.5.2 DuckDB +extension-cache mismatch; CI runs against fresh extensions. +""" + +import base64 +import hashlib +import hmac +import json +import os +import socket +import subprocess +import tempfile +import time +from typing import Dict, Iterator, List, Optional + +import pytest +import requests + + +JWT_SECRET = "audit-test-secret" +JWT_ISSUER = "audit-test-issuer" + + +def _repo_root() -> str: + return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + + +def _flapi_binary() -> str: + candidates: List[str] = [] + for build_type in ("release", "debug"): + path = os.path.join(_repo_root(), "build", build_type, "flapi") + if os.path.exists(path): + candidates.append(path) + if not candidates: + pytest.skip("flapi binary not found in build/release or build/debug") + candidates.sort(key=os.path.getmtime, reverse=True) + return candidates[0] + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("utf-8") + + +def _make_jwt(sub: str = "audit-user", roles: Optional[List[str]] = None) -> str: + if roles is None: + roles = ["analyst"] + header = {"alg": "HS256", "typ": "JWT"} + now = int(time.time()) + payload = { + "iss": JWT_ISSUER, + "sub": sub, + "roles": roles, + "iat": now, + "exp": now + 3600, + } + h = _b64url(json.dumps(header, separators=(",", ":")).encode("utf-8")) + p = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8")) + signature = hmac.new(JWT_SECRET.encode("utf-8"), f"{h}.{p}".encode("utf-8"), hashlib.sha256).digest() + return f"{h}.{p}.{_b64url(signature)}" + + +def _write_config(dirpath: str, port: int, audit_path: str) -> str: + sqls = os.path.join(dirpath, "sqls") + os.makedirs(sqls) + + with open(os.path.join(dirpath, "flapi.yaml"), "w") as f: + f.write( + f"project-name: audit-log-test\n" + f"project-description: Audit log E2E\n" + f"http-port: {port}\n" + f"template:\n" + f" path: ./sqls\n" + f"connections:\n" + f" inmem:\n" + f" properties:\n" + f" database: ':memory:'\n" + f"mcp:\n" + f" enabled: true\n" + f" auth:\n" + f" enabled: true\n" + f" type: bearer\n" + f" jwt-secret: {JWT_SECRET}\n" + f" jwt-issuer: {JWT_ISSUER}\n" + f"audit:\n" + f" enabled: true\n" + f" sink: file\n" + f" path: {audit_path}\n" + f" redact:\n" + f" - token\n" + ) + + with open(os.path.join(sqls, "lookup.yaml"), "w") as f: + f.write(""" +template-source: lookup.sql +connection: [inmem] +request: + - field-name: id + field-in: query + field-type: int + required: true + validators: + - type: int + min: 1 + - field-name: token + field-in: query + field-type: string + required: false +mcp-tool: + name: customer_lookup + description: Look up a customer by id +""") + with open(os.path.join(sqls, "lookup.sql"), "w") as f: + f.write("SELECT {{ params.id }} AS id\n") + + return os.path.join(dirpath, "flapi.yaml") + + +@pytest.fixture +def audit_server() -> Iterator[Dict[str, str]]: + """Start a flapi server with audit enabled; yield {base_url, audit_path}.""" + binary = _flapi_binary() + port = _free_port() + with tempfile.TemporaryDirectory(prefix="flapi_audit_") as tmpdir: + audit_path = os.path.join(tmpdir, "audit.jsonl") + config_path = _write_config(tmpdir, port, audit_path) + log_path = os.path.join(tmpdir, "server.log") + log_file = open(log_path, "w") + proc = subprocess.Popen( + [binary, "-c", config_path, "--no-telemetry"], + cwd=tmpdir, + stdout=log_file, + stderr=subprocess.STDOUT, + ) + try: + base_url = f"http://127.0.0.1:{port}" + deadline = time.time() + 30 + up = False + while time.time() < deadline: + if proc.poll() is not None: + break + try: + r = requests.get(f"{base_url}/mcp/health", timeout=1) + if r.status_code < 500: + up = True + break + except requests.exceptions.RequestException: + time.sleep(0.5) + if not up: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + log_file.close() + with open(log_path) as f: + log_text = f.read() + if "core_functions_duckdb_cpp_init" in log_text and "unique_ptr that is NULL" in log_text: + pytest.skip( + "flapi could not boot: local DuckDB extension cache is " + "incompatible with the in-tree DuckDB submodule. CI exercises this path." + ) + raise RuntimeError(f"flapi failed to start. Log:\n{log_text}") + yield {"base_url": base_url, "audit_path": audit_path} + finally: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + log_file.close() + + +def _open_session(base_url: str, token: str) -> str: + r = requests.post( + f"{base_url}/mcp/jsonrpc", + headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, + json={ + "jsonrpc": "2.0", + "id": "init-1", + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "clientInfo": {"name": "audit-test", "version": "0.1"}, + "capabilities": {}, + }, + }, + timeout=10, + ) + assert r.status_code == 200, r.text + sid = r.headers.get("Mcp-Session-Id") + assert sid, f"no session id: {dict(r.headers)}" + return sid + + +def _tools_call(base_url: str, token: str, session_id: str, args: dict) -> requests.Response: + return requests.post( + f"{base_url}/mcp/jsonrpc", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Mcp-Session-Id": session_id, + }, + json={ + "jsonrpc": "2.0", + "id": "call-1", + "method": "tools/call", + "params": {"name": "customer_lookup", "arguments": args}, + }, + timeout=10, + ) + + +def _read_audit_lines(audit_path: str) -> List[dict]: + """Wait briefly for the audit file to be flushed, then read parsed lines.""" + deadline = time.time() + 5 + lines: List[str] = [] + while time.time() < deadline: + if os.path.exists(audit_path): + with open(audit_path) as f: + lines = [l for l in f.read().splitlines() if l.strip()] + if lines: + break + time.sleep(0.1) + return [json.loads(l) for l in lines] + + +@pytest.mark.standalone_server +class TestAuditLog: + """End-to-end coverage for `audit.enabled` + `audit.sink=file`.""" + + def test_successful_tool_call_writes_audit_line(self, audit_server): + token = _make_jwt(sub="alice") + sid = _open_session(audit_server["base_url"], token) + + r = _tools_call(audit_server["base_url"], token, sid, {"id": 42, "token": "secret123"}) + assert r.status_code == 200, r.text + + events = _read_audit_lines(audit_server["audit_path"]) + # There should be exactly one event for the tool call. + assert len(events) == 1, events + ev = events[0] + assert ev["method"] == "tools/call" + assert ev["target"] == "customer_lookup" + assert ev["principal"] == "alice" + # Status may be success or error depending on whether the DB actually + # executed the query (e.g., on environments where in-mem works); the + # contract here is "an event was emitted", not "the query succeeded". + assert ev["status"], ev + assert "request_id" in ev and ev["request_id"] + assert "timestamp" in ev and ev["timestamp"] + # Redaction: token must be masked, non-redacted id must survive. + assert ev["params"]["token"] == "" + assert ev["params"]["id"] == "42" + # The literal secret must not appear anywhere in the line. + with open(audit_server["audit_path"]) as f: + content = f.read() + assert "secret123" not in content + + def test_invalid_arguments_still_audited(self, audit_server): + # When validation rejects the call, an audit event is still emitted + # so operators can investigate "why is this client failing?" without + # turning on debug logging. + token = _make_jwt(sub="bob") + sid = _open_session(audit_server["base_url"], token) + + # `id` is required and must be int>=1; send a string instead. + r = _tools_call(audit_server["base_url"], token, sid, {"id": "not-a-number"}) + assert r.status_code == 200, r.text # JSON-RPC error in body + + events = _read_audit_lines(audit_server["audit_path"]) + assert len(events) == 1, events + ev = events[0] + assert ev["status"].startswith("error:"), ev + assert ev["principal"] == "bob" + assert ev["target"] == "customer_lookup"