feat: api gateway and session token for securing cluster (#292)

* feat: secure with jwt in API gateway

* refactor: make secure context generic

* feat: validate token with join room

* WIP: api gateway

* feat: gateway agent and gateway store service

* WIP: mock for cross node rpc

* WIP: handle gateway request with WHIP

* WIP: allow whep connect over gateway

* WIP: allow webrtc sdk connect over gateway

* WIP: allow forward from gateway to gateway

* feat: geo-location

* temp allow rsa@0.9.6

Cargo.toml

@@ -6,11 +6,13 @@ members = [
     "packages/media_core",
     "packages/media_runner",
     "packages/transport_webrtc",
+    "packages/media_secure",
+    "packages/media_gateway",
 ]
 
 [workspace.dependencies]
-sans-io-runtime = { git = "https://github.com/giangndm/sans-io-runtime.git", rev = "e7ef60e0eef35c532c8544c472514ae831a8908f" }
-atm0s-sdn = { git = "https://github.com/giangndm/8xFF-decentralized-sdn.git", rev = "9200a1615def0ddffce8338b34afd24421f24269" }
+sans-io-runtime = { git = "https://github.com/giangndm/sans-io-runtime.git", rev = "c781cef12b2a435b5e31a6ede69d301a23719452" }
+atm0s-sdn = { git = "https://github.com/giangndm/8xFF-decentralized-sdn.git", rev = "e3456db45912bdd461755088a5dde5e004b0f17a" }
 tracing-subscriber = { version = "0.3", features = ["env-filter", "std"] }
 convert-enum = "0.1"
 num_enum = "0.7"

bin/Cargo.toml

@@ -16,6 +16,24 @@ poem-openapi = { version = "5.0", features = ["swagger-ui"] }
 tokio = { version = "1.37", features = ["full"] }
 sans-io-runtime = { workspace = true }
 atm0s-sdn = { workspace = true }
-media-server-protocol = { path = "../packages/protocol" }
-media-server-runner = { path = "../packages/media_runner" }
+media-server-protocol = { path = "../packages/protocol", features = ["quinn-rpc"] }
+media-server-secure = { path = "../packages/media_secure" }
+media-server-runner = { path = "../packages/media_runner", optional = true }
+media-server-gateway = { path = "../packages/media_gateway", optional = true }
 local-ip-address = "0.6"
+serde = { version = "1.0", features = ["derive"] }
+quinn = { version = "0.11", optional = true }
+rustls = { version = "0.23", optional = true }
+convert-enum = { workspace = true }
+num_enum = { workspace = true }
+derive_more = { workspace = true }
+rcgen = { version = "0.13", optional = true }
+maxminddb = "0.24.0"
+
+[features]
+default = ["gateway", "media", "connector", "cert_utils"]
+gateway = ["media-server-gateway", "quinn_vnet"]
+media = ["media-server-runner", "quinn_vnet"]
+connector = ["quinn_vnet"]
+cert_utils = ["rcgen", "rustls"]
+quinn_vnet = ["rustls", "quinn"]

bin/gate_z0_n1.sh

@@ -0,0 +1,11 @@
+RUST_LOG=atm0s_sdn_network::features::socket=debug,info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 3000 \
+    --node-id 0 \
+    --sdn-port 10000 \
+    --sdn-zone 0 \
+    gateway \
+        --lat 10 \
+        --lon 20 \
+        --geo-db "../maxminddb-data/GeoLite2-City.mmdb"

bin/gate_z256_n1.sh

@@ -0,0 +1,12 @@
+RUST_LOG=atm0s_sdn_network::features::socket=debug,info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 4000 \
+    --node-id 256 \
+    --sdn-zone 256 \
+    --sdn-port 11000 \
+    --seeds 0@/ip4/127.0.0.1/udp/10000 \
+    gateway \
+        --lat 20 \
+        --lon 30 \
+        --geo-db "../maxminddb-data/GeoLite2-City.mmdb"

bin/media_z0_n1.sh

@@ -0,0 +1,11 @@
+RUST_LOG=atm0s_sdn_network::features::socket=debug,info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 3001 \
+    --node-id 1 \
+    --sdn-port 10001 \
+    --sdn-zone 0 \
+    --seeds 0@/ip4/127.0.0.1/udp/10000 \
+    media \
+        --allow-private-ip \
+        --enable-token-api

bin/media_z0_n2.sh

@@ -0,0 +1,11 @@
+RUST_LOG=info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 3002 \
+    --node-id 2 \
+    --sdn-port 10002 \
+    --sdn-zone 0 \
+    --seeds 0@/ip4/127.0.0.1/udp/10000 \
+    media \
+        --allow-private-ip \
+        --enable-token-api

bin/media_z256_n1.sh

@@ -0,0 +1,11 @@
+RUST_LOG=atm0s_sdn_network::features::socket=debug,info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 4001 \
+    --node-id 257 \
+    --sdn-port 11001 \
+    --sdn-zone 256 \
+    --seeds 256@/ip4/127.0.0.1/udp/11000 \
+    media \
+        --allow-private-ip \
+        --enable-token-api

bin/media_z256_n2.sh

@@ -0,0 +1,11 @@
+RUST_LOG=info \
+RUST_BACKTRACE=1 \
+cargo run -- \
+    --http-port 4002 \
+    --node-id 258 \
+    --sdn-port 11002 \
+    --sdn-zone 256 \
+    --seeds 256@/ip4/127.0.0.1/udp/11000 \
+    media \
+        --allow-private-ip \
+        --enable-token-api

bin/src/errors.rs

@@ -0,0 +1,6 @@
+#[derive(num_enum::TryFromPrimitive, num_enum::IntoPrimitive, derive_more::Display)]
+#[repr(u32)]
+pub enum MediaServerError {
+    GatewayRpcError = 0x00020001,
+    InvalidConnId = 0x00020002,
+}

bin/src/http.rs

@@ -1,14 +1,21 @@
 use std::net::SocketAddr;
+use std::sync::Arc;
 
 use media_server_protocol::endpoint::ClusterConnId;
 use media_server_protocol::transport::{RpcReq, RpcRes};
+use media_server_secure::{MediaEdgeSecure, MediaGatewaySecure};
 use poem::endpoint::StaticFilesEndpoint;
 use poem::{listener::TcpListener, middleware::Cors, EndpointExt, Route, Server};
 use poem_openapi::types::{ToJSON, Type};
 use poem_openapi::OpenApiService;
 use poem_openapi::{types::ParseFromJSON, Object};
 use tokio::sync::mpsc::Sender;
 
+mod api_connector;
+mod api_media;
+mod api_token;
+mod utils;
+
 #[derive(Debug, Default, Object)]
 pub struct Response<T: ParseFromJSON + ToJSON + Type + Send + Sync> {
     pub status: bool,
@@ -34,36 +41,58 @@ impl<Req, Res> Rpc<Req, Res> {
     }
 }
 
-mod api_connector;
-mod api_media;
-mod utils;
-
-pub async fn run_gateway_http_server(sender: Sender<Rpc<RpcReq<ClusterConnId>, RpcRes<ClusterConnId>>>) -> Result<(), Box<dyn std::error::Error>> {
-    let api_service: OpenApiService<_, ()> = OpenApiService::new(api_media::MediaApis, "Media Gateway APIs", env!("CARGO_PKG_VERSION")).server("/");
-    let ui = api_service.swagger_ui();
-    let spec = api_service.spec();
+pub async fn run_gateway_http_server<ES: 'static + MediaEdgeSecure + Send + Sync, GS: 'static + MediaGatewaySecure + Send + Sync>(
+    port: u16,
+    sender: Sender<Rpc<RpcReq<ClusterConnId>, RpcRes<ClusterConnId>>>,
+    edge_secure: Arc<ES>,
+    gateway_secure: Arc<GS>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let token_service: OpenApiService<_, ()> = OpenApiService::new(api_token::TokenApis::<GS>::new(), "App APIs", env!("CARGO_PKG_VERSION")).server("/token/");
+    let token_ui = token_service.swagger_ui();
+    let token_spec = token_service.spec();
+    let media_service: OpenApiService<_, ()> = OpenApiService::new(api_media::MediaApis::<ES>::new(), "Media Gateway APIs", env!("CARGO_PKG_VERSION")).server("/media/");
+    let media_ui = media_service.swagger_ui();
+    let media_spec = media_service.spec();
     let route = Route::new()
-        .nest("/", api_service)
-        .nest("/ui", ui)
-        .at("/spec", poem::endpoint::make_sync(move |_| spec.clone()))
-        .with(Cors::new())
-        .data(api_media::MediaServerCtx { sender });
+        .nest("/samples", StaticFilesEndpoint::new("./public").index_file("index.html"))
+        .nest("/token/", token_service.data(api_token::TokenServerCtx { secure: gateway_secure }))
+        .nest("/token/ui", token_ui)
+        .at("/token/spec", poem::endpoint::make_sync(move |_| token_spec.clone()))
+        .nest("/", media_service.data(api_media::MediaServerCtx { sender, secure: edge_secure }))
+        .nest("/ui", media_ui)
+        .at("/spec", poem::endpoint::make_sync(move |_| media_spec.clone()))
+        .with(Cors::new());
 
-    Server::new(TcpListener::bind("0.0.0.0:3000")).run(route).await?;
+    Server::new(TcpListener::bind(SocketAddr::new([0, 0, 0, 0].into(), port))).run(route).await?;
     Ok(())
 }
 
-pub async fn run_media_http_server(port: u16, sender: Sender<Rpc<RpcReq<ClusterConnId>, RpcRes<ClusterConnId>>>) -> Result<(), Box<dyn std::error::Error>> {
-    let api_service: OpenApiService<_, ()> = OpenApiService::new(api_media::MediaApis, "Media Server APIs", env!("CARGO_PKG_VERSION")).server("/");
-    let ui = api_service.swagger_ui();
-    let spec = api_service.spec();
-    let route = Route::new()
-        .nest("/", api_service)
+pub async fn run_media_http_server<ES: 'static + MediaEdgeSecure + Send + Sync, GS: 'static + MediaGatewaySecure + Send + Sync>(
+    port: u16,
+    sender: Sender<Rpc<RpcReq<ClusterConnId>, RpcRes<ClusterConnId>>>,
+    edge_secure: Arc<ES>,
+    gateway_secure: Option<Arc<GS>>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut route = Route::new();
+
+    if let Some(gateway_secure) = gateway_secure {
+        let token_service: OpenApiService<_, ()> = OpenApiService::new(api_token::TokenApis::<GS>::new(), "App APIs", env!("CARGO_PKG_VERSION")).server("/token/");
+        let token_ui = token_service.swagger_ui();
+        let token_spec = token_service.spec();
+        route = route
+            .nest("/token/", token_service.data(api_token::TokenServerCtx { secure: gateway_secure }))
+            .nest("/token/ui", token_ui)
+            .at("/token/spec", poem::endpoint::make_sync(move |_| token_spec.clone()));
+    }
+    let media_service: OpenApiService<_, ()> = OpenApiService::new(api_media::MediaApis::<ES>::new(), "Media Gateway APIs", env!("CARGO_PKG_VERSION")).server("/media/");
+    let media_ui = media_service.swagger_ui();
+    let media_spec = media_service.spec();
+    let route = route
         .nest("/samples", StaticFilesEndpoint::new("./public").index_file("index.html"))
-        .nest("/ui", ui)
-        .at("/spec", poem::endpoint::make_sync(move |_| spec.clone()))
-        .with(Cors::new())
-        .data(api_media::MediaServerCtx { sender });
+        .nest("/", media_service.data(api_media::MediaServerCtx { sender, secure: edge_secure }))
+        .nest("/ui", media_ui)
+        .at("/spec", poem::endpoint::make_sync(move |_| media_spec.clone()))
+        .with(Cors::new());
 
     Server::new(TcpListener::bind(SocketAddr::new([0, 0, 0, 0].into(), port))).run(route).await?;
     Ok(())