libsql-server: enforce namespace authorization on gRPC replication RPCs

The replication path validated the JWT signature but discarded the
returned Authenticated, so any valid token could replicate any
namespace regardless of its ns claim. Check is_namespace_authorized
against the served namespace and add a negative cross-namespace test.

Author: bartzon

libsql-server/src/rpc/replication/auth.rs

@@ -39,7 +39,12 @@ pub async fn authenticate<T>(
     if let Some(auth) = auth {
         let context = parse_grpc_auth_header(req.metadata(), &auth.user_strategy.required_fields())
             .map_err(|e| tonic::Status::internal(format!("Error parsing auth header: {}", e)))?;
-        auth.authenticate(context)?;
+        let authenticated = auth.authenticate(context)?;
+        if !authenticated.is_namespace_authorized(&namespace) {
+            return Err(tonic::Status::permission_denied(
+                "not authorized to replicate this namespace",
+            ));
+        }
     }
 
     Ok(())

libsql-server/tests/embedded_replica/mod.rs

@@ -1796,3 +1796,118 @@ fn remote_replica_namespace_header_support() {
 
     sim.run().unwrap();
 }
+
+#[test]
+fn replicate_cross_namespace_denied() {
+    init_tracing();
+    let mut sim = Builder::new()
+        .simulation_duration(Duration::from_secs(1000))
+        .build();
+
+    let (encoding, decoding) = common::auth::key_pair();
+    sim.host("primary", {
+        let decoding = decoding.clone();
+        move || {
+            let decoding = decoding.clone();
+            async move {
+                let tmp = tempdir()?;
+                let jwt_keys =
+                    vec![jsonwebtoken::DecodingKey::from_ed_components(&decoding).unwrap()];
+                let auth = Auth::new(user_auth_strategies::Jwt::new(jwt_keys));
+                let server = TestServer {
+                    path: tmp.path().to_owned().into(),
+                    user_api_config: UserApiConfig {
+                        hrana_ws_acceptor: None,
+                        auth_strategy: auth,
+                        ..Default::default()
+                    },
+                    admin_api_config: Some(AdminApiConfig {
+                        acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 9090)).await?,
+                        connector: TurmoilConnector,
+                        disable_metrics: true,
+                        auth_key: None,
+                    }),
+                    rpc_server_config: Some(RpcServerConfig {
+                        acceptor: TurmoilAcceptor::bind(([0, 0, 0, 0], 4567)).await?,
+                        tls_config: None,
+                    }),
+                    disable_namespaces: false,
+                    disable_default_namespace: true,
+                    ..Default::default()
+                };
+
+                server.start_sim(8080).await?;
+
+                Ok(())
+            }
+        }
+    });
+
+    sim.client("client", async move {
+        // Create two distinct namespaces.
+        let client = Client::new();
+        client
+            .post("http://primary:9090/v1/namespaces/ns1/create", json!({}))
+            .await?;
+        client
+            .post("http://primary:9090/v1/namespaces/ns2/create", json!({}))
+            .await?;
+
+        // Token scoped to ns1 only.
+        let ns1_token = encode(
+            &serde_json::json!({
+                "p": { "ro": { "ns": ["ns1"] } },
+            }),
+            &encoding,
+        );
+
+        // Sanity: ns1 token works against ns1.
+        let tmp = tempdir().unwrap();
+        let db = Database::open_with_remote_sync_connector(
+            tmp.path().join("embedded-ns1").to_str().unwrap(),
+            "http://ns1.primary:8080",
+            ns1_token.clone(),
+            TurmoilConnector,
+            false,
+            None,
+        )
+        .await?;
+        assert!(db.sync().await.is_ok());
+
+        // Negative: ns1 token must NOT replicate ns2.
+        let tmp = tempdir().unwrap();
+        let db = Database::open_with_remote_sync_connector(
+            tmp.path().join("embedded-ns2").to_str().unwrap(),
+            "http://ns2.primary:8080",
+            ns1_token.clone(),
+            TurmoilConnector,
+            false,
+            None,
+        )
+        .await?;
+        assert!(db.sync().await.is_err());
+
+        // Token scoped to both namespaces can replicate both (control).
+        let both_token = encode(
+            &serde_json::json!({
+                "p": { "ro": { "ns": ["ns1", "ns2"] } },
+            }),
+            &encoding,
+        );
+        let tmp = tempdir().unwrap();
+        let db = Database::open_with_remote_sync_connector(
+            tmp.path().join("embedded-ns2-both").to_str().unwrap(),
+            "http://ns2.primary:8080",
+            both_token.clone(),
+            TurmoilConnector,
+            false,
+            None,
+        )
+        .await?;
+        assert!(db.sync().await.is_ok());
+
+        Ok(())
+    });
+
+    sim.run().unwrap();
+}