feat(grpc): add support for kicking players and showing current online players

Cargo.lock

@@ -2666,6 +2666,16 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "rnex-server-api"
+version = "0.1.0"
+dependencies = [
+ "prost",
+ "tonic",
+ "tonic-prost",
+ "tonic-prost-build",
+]
+
 [[package]]
 name = "rnex-server-backend-auth"
 version = "0.1.0"
@@ -2686,7 +2696,9 @@ dependencies = [
  "rnex-msg",
  "rnex-rk",
  "rnex-server",
+ "rnex-server-api",
  "tokio",
+ "tonic",
 ]
 
 [[package]]

rnex-server-api-grpc/meta.proto

@@ -21,7 +21,28 @@ message ApiFeatures {
   repeated ApiFeature api_features = 1;
 }
 
+message Empty {}
+
 service ServerMetaService {
   rpc GetBuildInfo(google.protobuf.Empty) returns (BuildInfo);
   rpc GetApiFeatures(google.protobuf.Empty) returns (ApiFeatures);
 }
+
+message KickPidsRequest {
+  repeated uint64 pids = 1;
+  string reason = 2;
+}
+
+message KickPidsResponse {
+  repeated uint64 kicked = 1;
+  repeated uint64 missing = 2;
+}
+
+message OnlinePidsResponse {
+  repeated uint64 pids = 1;
+}
+
+service AdminService {
+  rpc KickPids(KickPidsRequest) returns (KickPidsResponse);
+  rpc ListOnlinePids(Empty) returns (OnlinePidsResponse);
+}

rnex-server/backend-secure/Cargo.toml

@@ -12,6 +12,8 @@ rnex-fpd = {path = "../../rnex-server-nex-modules/rnex-fpd", optional = true}
 rnex-msg = {path = "../../rnex-server-nex-modules/rnex-msg", optional = true}
 rnex-rk = {path = "../../rnex-server-nex-modules/rnex-rk", optional = true}
 tokio = { version = "1.52.3", features = ["time", "sync", "rt-multi-thread"] }
+rnex-server-api = { path = "../../rnex-server-api" }
+tonic = { version = "0.14.6", features = ["transport"] }
 
 [features]
 match-making = ["dep:rnex-mm"]

rnex-server/backend-secure/src/admin.rs

@@ -0,0 +1,104 @@
+use std::{convert::TryInto, net::SocketAddr, sync::Arc};
+use rnex_server::{ConnectionRegistry, util::PID};
+use rnex_server_api::meta::{admin_service_server::{AdminService, AdminServiceServer}, KickPidsRequest, KickPidsResponse, OnlinePidsResponse};
+use tonic::{Request, Response, Status, transport::Server};
+use rnex_server_api::meta::Empty;
+
+pub struct AdminGrpc {
+    registry: Arc<ConnectionRegistry>,
+    token: Option<String>,
+}
+
+impl AdminGrpc {
+    pub fn new(registry: Arc<ConnectionRegistry>, token: Option<String>) -> Self {
+        Self { registry, token }
+    }
+
+    fn authorize<T>(&self, request: &Request<T>) -> Result<(), Status> {
+        let Some(expected) = &self.token else {
+            return Ok(());
+        };
+
+        let Some(value) = request.metadata().get("authorization") else {
+            return Err(Status::unauthenticated("missing authorization header"));
+        };
+
+        let header = value
+            .to_str()
+            .map_err(|_| Status::unauthenticated("authorization header must be ASCII"))?;
+
+        let supplied = header
+            .strip_prefix("Bearer ")
+            .or_else(|| header.strip_prefix("bearer "))
+            .ok_or_else(|| Status::unauthenticated("authorization header must use Bearer scheme"))?;
+
+        if supplied == expected {
+            Ok(())
+        } else {
+            Err(Status::unauthenticated("invalid admin token"))
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl AdminService for AdminGrpc {
+    async fn kick_pids(
+        &self,
+        request: Request<KickPidsRequest>,
+    ) -> Result<Response<KickPidsResponse>, Status> {
+        self.authorize(&request)?;
+
+        let KickPidsRequest { pids, reason } = request.into_inner();
+
+        rnex_server::tracing::info!(%reason, pid_count = pids.len(), "received kick request");
+
+        let mut kicked = Vec::new();
+        let mut missing = Vec::new();
+
+        for raw_pid in pids {
+            let pid: PID = raw_pid
+                .try_into()
+                .map_err(|_| Status::invalid_argument(format!("pid {raw_pid} is not a valid pid for this server")))?;
+
+            if self.registry.kick(pid).await {
+                kicked.push(raw_pid);
+            } else {
+                missing.push(raw_pid);
+            }
+        }
+
+        Ok(Response::new(KickPidsResponse { kicked, missing }))
+    }
+
+    async fn list_online_pids(
+        &self,
+        request: Request<Empty>,
+    ) -> Result<Response<OnlinePidsResponse>, Status> {
+        self.authorize(&request)?;
+
+        let pids = self
+            .registry
+            .online_pids()
+            .await
+            .into_iter()
+            .map(|pid| {
+                pid.try_into().map_err(|_| {
+                    Status::internal(format!("pid {pid} is not a valid pid for this server"))
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(Response::new(OnlinePidsResponse { pids }))
+    }
+}
+
+pub async fn serve(
+    addr: SocketAddr,
+    registry: Arc<ConnectionRegistry>,
+    token: Option<String>,
+) -> Result<(), tonic::transport::Error> {
+    Server::builder()
+        .add_service(AdminServiceServer::new(AdminGrpc::new(registry, token)))
+        .serve(addr)
+        .await
+}

rnex-server/backend-secure/src/main.rs

@@ -1,3 +1,5 @@
+mod admin;
+use std::net::SocketAddr;
 use rnex_base::BaseModule;
 #[cfg(feature = "datastore")]
 use rnex_ds::DatastoreModule;
@@ -9,11 +11,29 @@ use rnex_mm::MatchMakeModule;
 use rnex_msg::MessagingModule;
 #[cfg(feature = "ranking")]
 use rnex_rk::RankingModule;
+use rnex_server::{ConnectionInitData, connection_registry, launch_rnex_module_server};
 
-use rnex_server::{ConnectionInitData, launch_rnex_module_server};
+fn admin_addr() -> SocketAddr {
+    std::env::var("RNEX_ADMIN_GRPC_ADDR")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or_else(|| SocketAddr::from(([127, 0, 0, 1], 50051)))
+}
 
 #[tokio::main]
 async fn main() {
+    let registry = connection_registry();
+    let token = std::env::var("RNEX_ADMIN_GRPC_TOKEN").ok();
+
+    let admin_task = tokio::spawn(async move {
+        let result: Result<(), tonic::transport::Error> =
+            admin::serve(admin_addr(), registry, token).await;
+
+        if let Err(err) = result {
+            rnex_server::tracing::error!(%err, "admin gRPC server stopped");
+        }
+    });
+
     launch_rnex_module_server! {
         ConnectionInitData;
         BaseModule,
@@ -27,6 +47,8 @@ async fn main() {
         RankingModule,
         #[cfg(feature="messaging")]
         MessagingModule
-
     }
-}
+
+    admin_task.abort();
+    let _ = admin_task.await;
+}

rnex-server/src/lib.rs

@@ -18,6 +18,7 @@ pub use anyhow;
 pub use paste;
 pub use rnex_rmc as rmc;
 use rnex_rmc::{RmcCallable, RmcConnection, RmcSerialize, serialization::RmcSerialize, util::PID};
+use rnex_util::{SendingBufferConnection, SplittableBufferConnection, UnitPacketRead};
 pub use rnex_util as util;
 use thiserror::Error;
 pub use tokio;
@@ -33,6 +34,62 @@ pub struct ConnectionInitData {
     pub pid: PID,
 }
 #[derive(Debug, Default)]
+pub struct ConnectionRegistry {
+    connections: tokio::sync::RwLock<HashMap<PID, SendingBufferConnection>>,
+}
+
+impl ConnectionRegistry {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub async fn register(&self, pid: PID, conn: SendingBufferConnection) {
+        let old = self.connections.write().await.insert(pid, conn);
+        if let Some(old) = old {
+            old.disconnect().await;
+        }
+    }
+
+    pub async fn online_pids(&self) -> Vec<PID> {
+        self.connections.read().await.keys().copied().collect()
+    }
+
+    pub async fn kick(&self, pid: PID) -> bool {
+        let Some(conn) = self.connections.write().await.remove(&pid) else {
+            return false;
+        };
+
+        conn.disconnect().await;
+        true
+    }
+
+    pub async fn kick_many(
+        &self,
+        pids: impl IntoIterator<Item = PID>,
+    ) -> (Vec<PID>, Vec<PID>) {
+        let mut kicked = Vec::new();
+        let mut missing = Vec::new();
+
+        for pid in pids {
+            if self.kick(pid).await {
+                kicked.push(pid);
+            } else {
+                missing.push(pid);
+            }
+        }
+
+        (kicked, missing)
+    }
+}
+
+static CONNECTION_REGISTRY: OnceLock<Arc<ConnectionRegistry>> = OnceLock::new();
+
+pub fn connection_registry() -> Arc<ConnectionRegistry> {
+    CONNECTION_REGISTRY
+        .get_or_init(|| Arc::new(ConnectionRegistry::new()))
+        .clone()
+}
+#[derive(Debug, Default)]
 pub struct ModuleHolder {
     modules: HashMap<TypeId, Arc<dyn Any + Send + Sync>>,
 }
@@ -273,7 +330,13 @@ macro_rules! launch_rnex_module_server {
                             return;
                         };
 
-                        $crate::rmc::new_rmc_gateway_connection(stream.into(),
+                        let stream = <$crate::util::SplittableBufferConnection as From<_>>::from(stream);
+                        $crate::connection_registry()
+                            .register(conn_data.pid, stream.duplicate_sender())
+                            .await;
+
+                        $crate::rmc::new_rmc_gateway_connection(
+                            stream,
                             async |r| {
                                $crate::tracing::info!("creating module holder for module users");
                                let mut holder = $crate::ModuleHolder::default();