diff --git a/Cargo.lock b/Cargo.lock index 3112347..36709c2 100644 --- a/Cargo.lock +++ b/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]] diff --git a/rnex-server-api-grpc/meta.proto b/rnex-server-api-grpc/meta.proto index e256e62..c289cc6 100644 --- a/rnex-server-api-grpc/meta.proto +++ b/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); +} \ No newline at end of file diff --git a/rnex-server/backend-secure/Cargo.toml b/rnex-server/backend-secure/Cargo.toml index abb5bec..4d53aba 100644 --- a/rnex-server/backend-secure/Cargo.toml +++ b/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"] diff --git a/rnex-server/backend-secure/src/admin.rs b/rnex-server/backend-secure/src/admin.rs new file mode 100644 index 0000000..f0e0409 --- /dev/null +++ b/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, + token: Option, +} + +impl AdminGrpc { + pub fn new(registry: Arc, token: Option) -> Self { + Self { registry, token } + } + + fn authorize(&self, request: &Request) -> 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, + ) -> Result, 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, + ) -> Result, 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::, _>>()?; + + Ok(Response::new(OnlinePidsResponse { pids })) + } +} + +pub async fn serve( + addr: SocketAddr, + registry: Arc, + token: Option, +) -> Result<(), tonic::transport::Error> { + Server::builder() + .add_service(AdminServiceServer::new(AdminGrpc::new(registry, token))) + .serve(addr) + .await +} \ No newline at end of file diff --git a/rnex-server/backend-secure/src/main.rs b/rnex-server/backend-secure/src/main.rs index 3e6e022..4f37af0 100644 --- a/rnex-server/backend-secure/src/main.rs +++ b/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; +} \ No newline at end of file diff --git a/rnex-server/src/lib.rs b/rnex-server/src/lib.rs index 56a2bdc..bb08abb 100644 --- a/rnex-server/src/lib.rs +++ b/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>, +} + +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 { + 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, + ) -> (Vec, Vec) { + 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> = OnceLock::new(); + +pub fn connection_registry() -> Arc { + CONNECTION_REGISTRY + .get_or_init(|| Arc::new(ConnectionRegistry::new())) + .clone() +} +#[derive(Debug, Default)] pub struct ModuleHolder { modules: HashMap>, } @@ -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();