test(security): cover SQL transport authorization

Author: farhan-syah

Cargo.lock

@@ -11,8 +11,8 @@ use tokio::net::TcpStream;
 use nodedb_types::protocol::request_fields::RequestFields;
 use nodedb_types::protocol::text_fields::TextFields;
 use nodedb_types::protocol::{
-    FRAME_HEADER_LEN, HELLO_ACK_MAGIC, HELLO_ERROR_MAGIC_U32, HelloAckFrame, HelloErrorFrame,
-    HelloFrame, NativeRequest, NativeResponse, OpCode,
+    AuthMethod, FRAME_HEADER_LEN, HELLO_ACK_MAGIC, HELLO_ERROR_MAGIC_U32, HelloAckFrame,
+    HelloErrorFrame, HelloFrame, NativeRequest, NativeResponse, OpCode,
 };
 
 /// Perform the handshake with a custom `HelloFrame`.
@@ -127,6 +127,21 @@ pub async fn send_request(
     sonic_rs::from_slice(&response_payload).expect("json decode NativeResponse")
 }
 
+/// Authenticate a fresh native connection with an API key. The JSON Auth
+/// request also selects JSON framing for the rest of the session.
+pub async fn send_api_key_auth(stream: &mut TcpStream, seq: u64, token: String) -> NativeResponse {
+    send_request(
+        stream,
+        seq,
+        OpCode::Auth,
+        TextFields {
+            auth: Some(AuthMethod::ApiKey { token }),
+            ..Default::default()
+        },
+    )
+    .await
+}
+
 /// Send a `SHOW`/SQL statement over an established JSON-encoding session and
 /// decode the `NativeResponse`. Assumes the session's first frame already
 /// selected JSON (see `json_request_gets_json_response`) — callers that open

nodedb-test-support/src/native_harness/frames.rs

@@ -9,5 +9,7 @@
 mod frames;
 mod server;
 
-pub use frames::{do_handshake, read_frame, send_request, send_sql, write_frame};
+pub use frames::{
+    do_handshake, read_frame, send_api_key_auth, send_request, send_sql, write_frame,
+};
 pub use server::NativeTestServer;

nodedb-test-support/src/native_harness/mod.rs

@@ -17,6 +17,7 @@ use nodedb::wal::WalManager;
 /// A running native-protocol test server.
 pub struct NativeTestServer {
     pub addr: std::net::SocketAddr,
+    pub shared: Arc<SharedState>,
     pub(super) shutdown_bus: nodedb::control::shutdown::ShutdownBus,
     pub(super) poller_shutdown_tx: tokio::sync::watch::Sender<bool>,
     pub(super) core_stop_tx: std::sync::mpsc::Sender<()>,
@@ -31,6 +32,15 @@ impl NativeTestServer {
     /// Spawn a single-core NodeDB server with the native listener bound to
     /// an ephemeral `127.0.0.1` port (trust-mode auth).
     pub async fn start() -> Self {
+        Self::start_with_auth_mode(AuthMode::Trust).await
+    }
+
+    /// Spawn a single-core server that requires an explicit native Auth frame.
+    pub async fn start_authenticated() -> Self {
+        Self::start_with_auth_mode(AuthMode::Password).await
+    }
+
+    async fn start_with_auth_mode(auth_mode: AuthMode) -> Self {
         let dir = tempfile::tempdir().expect("tempdir");
         let wal_path = dir.path().join("test.wal");
         let wal = Arc::new(WalManager::open_for_testing(&wal_path).expect("open wal"));
@@ -128,7 +138,7 @@ impl NativeTestServer {
             listener
                 .run(nodedb::control::server::listener::ListenerRunParams {
                     state: shared_listener,
-                    auth_mode: AuthMode::Trust,
+                    auth_mode,
                     tls_acceptor: None,
                     conn_semaphore: Arc::new(tokio::sync::Semaphore::new(128)),
                     startup_gate: test_startup_gate,
@@ -145,6 +155,7 @@ impl NativeTestServer {
 
         Self {
             addr,
+            shared,
             shutdown_bus,
             poller_shutdown_tx,
             core_stop_tx,

nodedb-test-support/src/native_harness/server.rs

@@ -0,0 +1,201 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Authorization parity for materialized HTTP SQL queries.
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use common::pgwire_harness::TestServer;
+use nodedb::config::auth::AuthMode;
+use nodedb::control::security::apikey::CreateKeyParams;
+use nodedb::control::security::identity::{Permission, Role};
+use nodedb::control::security::permission::collection_target;
+use nodedb::control::state::SharedState;
+use nodedb::types::{DatabaseId, TenantId};
+
+struct AuthenticatedHttpEndpoint {
+    local_addr: std::net::SocketAddr,
+    _server: tokio::task::JoinHandle<()>,
+}
+
+async fn start_authenticated_http(shared: Arc<SharedState>) -> AuthenticatedHttpEndpoint {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind authenticated HTTP listener");
+    let local_addr = listener.local_addr().expect("authenticated HTTP address");
+    let (bus, _) = nodedb::control::shutdown::ShutdownBus::new(Arc::clone(&shared.shutdown));
+    let handle = tokio::spawn(async move {
+        nodedb::control::server::http::server::run_with_listener(
+            listener,
+            shared,
+            AuthMode::Password,
+            None,
+            bus,
+        )
+        .await
+        .expect("authenticated HTTP server");
+    });
+    tokio::time::sleep(Duration::from_millis(40)).await;
+
+    AuthenticatedHttpEndpoint {
+        local_addr,
+        _server: handle,
+    }
+}
+
+fn create_api_key(shared: &SharedState, username: &str, roles: Vec<Role>) -> String {
+    let user_id = shared
+        .credentials
+        .create_service_account(username, TenantId::new(1), roles, vec![DatabaseId::DEFAULT])
+        .expect("create database-scoped service account");
+    shared
+        .api_keys
+        .create_key(
+            CreateKeyParams {
+                username,
+                user_id,
+                tenant_id: TenantId::new(1),
+                expires_secs: 0,
+                scope: vec![],
+                accessible_databases: vec![DatabaseId::DEFAULT],
+            },
+            Some(shared.credentials.catalog()),
+        )
+        .expect("create API key")
+}
+
+async fn post_query(http: &AuthenticatedHttpEndpoint, token: &str, sql: &str) -> reqwest::Response {
+    reqwest::Client::new()
+        .post(format!("http://{}/v1/query", http.local_addr))
+        .header("Authorization", format!("Bearer {token}"))
+        .json(&serde_json::json!({"sql": sql}))
+        .send()
+        .await
+        .expect("POST authenticated query")
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn query_rejects_database_outside_api_key_scope() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE DATABASE private_http_db")
+        .await
+        .expect("create inaccessible database");
+    let token = create_api_key(&srv.shared, "http_db_reader", vec![Role::ReadOnly]);
+    let http = start_authenticated_http(Arc::clone(&srv.shared)).await;
+
+    let response = reqwest::Client::new()
+        .post(format!("http://{}/v1/query", http.local_addr))
+        .header("Authorization", format!("Bearer {token}"))
+        .header("X-NodeDB-Database", "private_http_db")
+        .json(&serde_json::json!({"sql": "SELECT 1"}))
+        .send()
+        .await
+        .expect("POST cross-database query");
+
+    assert_eq!(
+        response.status(),
+        reqwest::StatusCode::FORBIDDEN,
+        "HTTP queries must enforce the API key's database scope before planning or execution"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn query_rejects_cross_database_write_without_mutating_target() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE DATABASE private_write_db")
+        .await
+        .expect("create inaccessible write database");
+    srv.exec("USE DATABASE private_write_db")
+        .await
+        .expect("switch to write database as superuser");
+    srv.exec("CREATE COLLECTION private_write_rows")
+        .await
+        .expect("create private write collection");
+    srv.exec("USE DATABASE default")
+        .await
+        .expect("return to default database");
+
+    let token = create_api_key(&srv.shared, "http_db_writer", vec![Role::ReadWrite]);
+    let http = start_authenticated_http(Arc::clone(&srv.shared)).await;
+    let response = reqwest::Client::new()
+        .post(format!("http://{}/v1/query", http.local_addr))
+        .header("Authorization", format!("Bearer {token}"))
+        .header("X-NodeDB-Database", "private_write_db")
+        .json(&serde_json::json!({
+            "sql": "INSERT INTO private_write_rows { id: 'forbidden', value: 5 }"
+        }))
+        .send()
+        .await
+        .expect("POST cross-database write");
+
+    srv.exec("USE DATABASE private_write_db")
+        .await
+        .expect("inspect write database");
+    let rows = srv
+        .query_text("SELECT id FROM private_write_rows")
+        .await
+        .expect("query private write rows");
+    assert!(
+        rows.is_empty(),
+        "a cross-database write must not mutate the target: {rows:?}"
+    );
+    assert_eq!(
+        response.status(),
+        reqwest::StatusCode::FORBIDDEN,
+        "cross-database HTTP writes must be rejected"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn query_rejects_system_catalog_for_non_superuser() {
+    let srv = TestServer::start().await;
+    let token = create_api_key(&srv.shared, "http_catalog_reader", vec![Role::ReadOnly]);
+    let http = start_authenticated_http(Arc::clone(&srv.shared)).await;
+
+    let response = post_query(&http, &token, "SELECT * FROM _system.audit_log").await;
+
+    assert_eq!(
+        response.status(),
+        reqwest::StatusCode::FORBIDDEN,
+        "system catalog access must require a superuser on every SQL transport"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn query_honors_explicit_collection_grant_for_custom_role() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE COLLECTION granted_rows")
+        .await
+        .expect("create granted collection");
+    srv.exec("INSERT INTO granted_rows { id: 'visible', value: 9 }")
+        .await
+        .expect("seed granted collection");
+
+    let username = "http_explicit_reader";
+    let token = create_api_key(
+        &srv.shared,
+        username,
+        vec![Role::Custom("http_explicit_role".into())],
+    );
+    srv.shared
+        .permissions
+        .grant(
+            &collection_target(TenantId::new(1), "granted_rows"),
+            &format!("user:{username}"),
+            Permission::Read,
+            "nodedb",
+            Some(srv.shared.credentials.catalog()),
+        )
+        .expect("grant collection read");
+    let http = start_authenticated_http(Arc::clone(&srv.shared)).await;
+
+    let response = post_query(&http, &token, "SELECT * FROM granted_rows").await;
+
+    assert_eq!(
+        response.status(),
+        reqwest::StatusCode::OK,
+        "HTTP authorization must honor PermissionStore grants before built-in role fallback"
+    );
+}