diff --git a/libsql-server/src/rpc/replication/auth.rs b/libsql-server/src/rpc/replication/auth.rs index 7c726e6d6d..14755d1dca 100644 --- a/libsql-server/src/rpc/replication/auth.rs +++ b/libsql-server/src/rpc/replication/auth.rs @@ -39,7 +39,12 @@ pub async fn authenticate( 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(()) diff --git a/libsql-server/tests/embedded_replica/mod.rs b/libsql-server/tests/embedded_replica/mod.rs index 9c40ea4a42..53ab9d426d 100644 --- a/libsql-server/tests/embedded_replica/mod.rs +++ b/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(); +}