check auth for proxied replication requests

Author: MarinPostma

libsql-server/src/lib.rs

@@ -129,11 +129,12 @@ pub(crate) static BLOCKING_RT: Lazy<Runtime> = Lazy::new(|| {
 type Result<T, E = Error> = std::result::Result<T, E>;
 type StatsSender = mpsc::Sender<(NamespaceName, MetaStoreHandle, Weak<Stats>)>;
 type MakeReplicationSvc = Box<
-    dyn FnOnce(
+    dyn Fn(
             NamespaceStore,
             Option<Auth>,
             Option<IdleShutdownKicker>,
             bool,
+            bool,
         ) -> BoxReplicationService
         + Send
         + 'static,
@@ -620,17 +621,18 @@ where
 
             let replication_service = make_replication_svc(
                 namespace_store.clone(),
-                None,
+                Some(user_auth_strategy.clone()),
                 idle_shutdown_kicker.clone(),
                 false,
+                true,
             );
 
             task_manager.spawn_until_shutdown(run_rpc_server(
                 proxy_service,
                 config.acceptor,
                 config.tls_config,
                 idle_shutdown_kicker.clone(),
-                replication_service,
+                replication_service, // internal replicaton service
             ));
         }
 
@@ -658,12 +660,12 @@ where
                         .await?;
                 }
 
-                let replication_svc = ReplicationLogService::new(
+                let replication_svc = make_replication_svc(
                     namespace_store.clone(),
-                    idle_shutdown_kicker.clone(),
                     Some(user_auth_strategy.clone()),
-                    self.disable_namespaces,
+                    idle_shutdown_kicker.clone(),
                     true,
+                    false, // external replication service
                 );
 
                 let proxy_svc = ProxyService::new(
@@ -936,9 +938,9 @@ where
         let make_replication_svc = Box::new({
             let registry = registry.clone();
             let disable_namespaces = self.disable_namespaces;
-            move |store, user_auth, _, _| -> BoxReplicationService {
+            move |store, user_auth, _, _, _| -> BoxReplicationService {
                 Box::new(LibsqlReplicationService::new(
-                    registry,
+                    registry.clone(),
                     store,
                     user_auth,
                     disable_namespaces,
@@ -1023,13 +1025,14 @@ where
 
         let make_replication_svc = Box::new({
             let disable_namespaces = self.disable_namespaces;
-            move |store, client_auth, idle_shutdown, collect_stats| -> BoxReplicationService {
+            move |store, client_auth, idle_shutdown, collect_stats, is_internal| -> BoxReplicationService {
                 Box::new(ReplicationLogService::new(
                     store,
                     idle_shutdown,
                     client_auth,
                     disable_namespaces,
                     collect_stats,
+                    is_internal,
                 ))
             }
         });
@@ -1055,13 +1058,14 @@ where
 
         let make_replication_svc = Box::new({
             let disable_namespaces = self.disable_namespaces;
-            move |store, client_auth, idle_shutdown, collect_stats| -> BoxReplicationService {
+            move |store, client_auth, idle_shutdown, collect_stats, is_internal| -> BoxReplicationService {
                 Box::new(ReplicationLogService::new(
                     store,
                     idle_shutdown,
                     client_auth,
                     disable_namespaces,
                     collect_stats,
+                    is_internal,
                 ))
             }
         });

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

@@ -37,6 +37,9 @@ pub struct ReplicationLogService {
     disable_namespaces: bool,
     session_token: Bytes,
     collect_stats: bool,
+    // whether this is an internal service. If it is an internal service, auth is checked for
+    // proxied requests
+    service_internal: bool,
 
     //deprecated:
     generation_id: Uuid,
@@ -52,6 +55,7 @@ impl ReplicationLogService {
         user_auth_strategy: Option<Auth>,
         disable_namespaces: bool,
         collect_stats: bool,
+        service_internal: bool,
     ) -> Self {
         let session_token = Uuid::new_v4().to_string().into();
         Self {
@@ -63,6 +67,7 @@ impl ReplicationLogService {
             collect_stats,
             generation_id: Uuid::new_v4(),
             replicas_with_hello: Default::default(),
+            service_internal,
         }
     }
 
@@ -71,14 +76,18 @@ impl ReplicationLogService {
         req: &tonic::Request<T>,
         namespace: NamespaceName,
     ) -> Result<(), Status> {
-        super::auth::authenticate(
-            &self.namespaces,
-            req,
-            namespace,
-            &self.user_auth_strategy,
-            true,
-        )
-        .await
+        if self.service_internal && req.metadata().get("libsql-proxied").is_some() || !self.service_internal {
+            super::auth::authenticate(
+                &self.namespaces,
+                req,
+                namespace,
+                &self.user_auth_strategy,
+                true,
+            )
+                .await
+        } else {
+            Ok(())
+        }
     }
 
     fn verify_session_token<R>(