feat(security): W1.3 — JSONL request audit log + MCP integration

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

src/audit_logger.cpp

@@ -0,0 +1,120 @@
+#include "audit_logger.hpp"
+
+#include <chrono>
+#include <crow/json.h>
+#include <ctime>
+#include <fstream>
+#include <iomanip>
+#include <iostream>
+#include <random>
+#include <sstream>
+
+namespace flapi {
+
+namespace {
+
+std::string randHex(int len) {
+    static thread_local std::mt19937_64 rng(std::random_device{}());
+    std::uniform_int_distribution<int> 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<std::ofstream>(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<std::mutex> 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<std::chrono::microseconds>(
+                            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] = "<redacted>";
+        } else {
+            params[key] = value;
+        }
+    }
+    line["params"] = std::move(params);
+    return line.dump();
+}
+
+} // namespace flapi

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<AuditLogger> 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<AuditLogger>(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<bool>(audit_node, "enabled", "audit.enabled", false);
+    audit_config.sink = safeGet<std::string>(audit_node, "sink", "audit.sink", "stdout");
+    audit_config.path = safeGet<std::string>(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<std::string>());
+        }
+    }
+
+    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

src/include/audit_logger.hpp

@@ -0,0 +1,61 @@
+#pragma once
+
+#include <map>
+#include <memory>
+#include <mutex>
+#include <ostream>
+#include <string>
+#include <unordered_set>
+
+namespace flapi {
+
+struct AuditConfig {
+    bool enabled = false;
+    std::string sink = "stdout";          // "stdout" | "file" | "null"
+    std::string path;                     // used when sink == "file"
+    std::unordered_set<std::string> 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:<code>" — 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<std::string, std::string> 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<std::ostream> 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

src/include/config_manager.hpp

@@ -10,9 +10,11 @@
 #include <stdexcept>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 #include <yaml-cpp/yaml.h>
 
+#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<AuditLogger> 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<AuditLogger> 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);

src/include/mcp_tool_handler.hpp

@@ -6,6 +6,7 @@
 #include <unordered_map>
 #include <vector>
 
+#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<ConfigManager> config_manager;
     std::shared_ptr<RequestValidator> validator;
     std::unique_ptr<SQLTemplateProcessor> sql_processor;
+    std::shared_ptr<AuditLogger> audit_logger;
 };
 
 } // namespace flapi

src/mcp_tool_handler.cpp

@@ -1,4 +1,5 @@
 #include "mcp_tool_handler.hpp"
+#include <chrono>
 #include <sstream>
 #include <algorithm>
 
@@ -8,20 +9,60 @@ MCPToolHandler::MCPToolHandler(std::shared_ptr<DatabaseManager> db_manager,
                               std::shared_ptr<ConfigManager> config_manager)
     : db_manager(db_manager), config_manager(config_manager),
       validator(std::make_shared<RequestValidator>()),
-      sql_processor(std::make_unique<SQLTemplateProcessor>(config_manager))
+      sql_processor(std::make_unique<SQLTemplateProcessor>(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::milliseconds>(
+            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<std::int64_t>(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<std::int64_t>(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()));
     }
 }

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