From 0a4f3fe1f159adce860ead0fb5bd5b6b6782ddf6 Mon Sep 17 00:00:00 2001 From: Leonid Kaganov Date: Mon, 18 Aug 2025 18:35:57 +0300 Subject: [PATCH] Fix workspace security for WebSocket --- scripts/TEST.html | 27 ++++++---- scripts/TEST_HTTP_API.sh | 4 ++ src/handlers_ws.rs | 83 +++++++++++++++++++++++++++-- src/main.rs | 109 ++++++++++++++++++++++----------------- src/redis_events.rs | 28 ++++------ src/redis_lib.rs | 1 + src/ws_hub.rs | 24 ++++++--- 7 files changed, 190 insertions(+), 86 deletions(-) diff --git a/scripts/TEST.html b/scripts/TEST.html index bfdb3d2842..2070094d09 100644 --- a/scripts/TEST.html +++ b/scripts/TEST.html @@ -67,20 +67,20 @@

- - + + - - + + - - + + - - + + - - + + @@ -92,7 +92,12 @@ const output = document.getElementById("output"); const textarea = document.getElementById("jsonInput"); - let ws = new WebSocket("ws://localhost:8095/ws"); +// const workspace="00000000-0000-0000-0000-000000000001"; + + let token="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50IjoiYWFhYWFhYWEtYmJiYi1jY2NjLWRkZGQtZWVlZWVlZWVlZWVlIiwiZXh0cmEiOnsic2VydmljZSI6ImFjY291bnQifSwid29ya3NwYWNlIjoiMDAwMDAwMDAtMDAwMC0wMDAwLTAwMDAtMDAwMDAwMDAwMDAxIn0.ZrwMvv_0CjuKeF2CkmHyMK2vHd9Ro4M3kHZcZBrBxZQ"; + let ws = new WebSocket(`ws://localhost:8095/ws?token=${token}`); + +// let ws = new WebSocket(`ws://localhost:8095/ws`); ws.onopen = () => { output.textContent = "✅ WebSocket connected."; diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index 050093fcf7..563d21de2f 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -4,6 +4,10 @@ clear source ./pulse_lib.sh TOKEN=$(./token.sh claims.json) +#echo ${TOKEN} +#exit + + ZP="00000000-0000-0000-0000-000000000001/TESTS" diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index c6a093f356..ae36e457ac 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -1,3 +1,8 @@ + +use uuid::Uuid; + +// ------------------- + use actix::{prelude::*}; use crate::ws_hub::{ @@ -105,12 +110,15 @@ pub enum WsCommand { }, } +use hulyrs::services::jwt::Claims; + /// Session condition #[allow(dead_code)] pub struct WsSession { pub redis: Arc>, pub id: SessionId, pub hub: Addr, + pub claims: Option, } @@ -186,6 +194,21 @@ impl StreamHandler> for WsSession { /// All logic in one impl impl WsSession { + fn ws_error(&self, ctx: &mut ws::WebsocketContext, msg: &str) { + ctx.text(format!(r#"{{"error":"{}"}}"#, msg)); + } + + fn workspace_check_ws(&self, key: &str) -> Result<(), &'static str> { + let claims = self.claims.as_ref().ok_or("Missing auth claims")?; + if claims.is_system() { return Ok(()); } + let jwt_workspace = claims.workspace.as_ref().ok_or("Missing workspace in token")?; + let path_ws = key.split('/').next().ok_or("Invalid key: missing workspace")?; + if path_ws.is_empty() { return Err("Invalid key: missing workspace"); } + let path_ws_uuid = Uuid::parse_str(path_ws).map_err(|_| "Invalid workspace UUID in key")?; + if jwt_workspace != &path_ws_uuid { return Err("Workspace mismatch"); } + Ok(()) + } + fn wait_and_send( &mut self, ctx: &mut ws::WebsocketContext, @@ -214,6 +237,9 @@ impl WsSession { println!("PUT {} = {} (expires_at: {:?}) (ttl: {:?})", key, data, expires_at, ttl); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let redis = self.redis.clone(); let mut base = JsonMap::new(); @@ -272,6 +298,9 @@ impl WsSession { WsCommand::Delete { key, correlation, if_match } => { println!("DELETE {}", key); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let redis = self.redis.clone(); let mut base = JsonMap::new(); @@ -302,6 +331,9 @@ impl WsSession { WsCommand::Get { key, correlation } => { println!("GET {}{:?}", key, correlation); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let redis = self.redis.clone(); let mut base = JsonMap::new(); @@ -334,6 +366,9 @@ impl WsSession { WsCommand::List { key, correlation } => { println!("LIST {:?}{:?}", key, correlation); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let redis = self.redis.clone(); let mut base = JsonMap::new(); @@ -357,13 +392,13 @@ impl WsSession { } - - - WsCommand::Sub { key, correlation } => { // LEVENT 3 println!("SUB {}{:?}", key, correlation); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let mut obj = JsonMap::new(); obj.insert("action".into(), json!("sub")); obj.insert("key".into(), json!(key)); @@ -381,6 +416,10 @@ impl WsSession { WsCommand::Unsub { key, correlation } => { // LEVENT 4 println!("UNSUB {}{:?}", key, correlation); + + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let mut obj = JsonMap::new(); obj.insert("action".into(), json!("unsub")); obj.insert("key".into(), json!(key)); @@ -402,6 +441,8 @@ impl WsSession { WsCommand::Sublist { correlation } => { println!("SUBLIST {:?}", correlation); + // w/o Check workspace! + let mut base = JsonMap::new(); base.insert("action".into(), json!("sublist")); if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } @@ -425,18 +466,50 @@ impl WsSession { } +// ---- auth + +use actix_web::{HttpMessage,error}; +use url::form_urlencoded; +use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm}; +use crate::CONFIG; pub async fn handler( req: HttpRequest, stream: web::Payload, redis: web::Data>>, - hub: web::Data>, + hub: web::Data>, ) -> Result { + + let token_opt = req.uri().query().and_then(|q| { + form_urlencoded::parse(q.as_bytes()) + .find(|(k, _)| k == "token") + .map(|(_, v)| v.into_owned()) + }); + + let claims = match token_opt { + Some(t) if !t.is_empty() => { + + let mut validation = Validation::new(Algorithm::HS256); + validation.required_spec_claims = HashSet::new(); // no: exp/iat/nbf + + let c = decode::(&t, &DecodingKey::from_secret(CONFIG.token_secret.as_bytes()), &validation ) + .map(|td| td.claims) + .map_err(|_e| error::ErrorUnauthorized("Invalid token"))?; + + + Some(c) + } + _ => None, + }; + + // println!("claims={:?}",&claims); + let session = WsSession { redis: redis.get_ref().clone(), hub: hub.get_ref().clone(), id: 0, + claims, }; + ws::start(session, &req, stream) } - diff --git a/src/main.rs b/src/main.rs index ffd4171748..122289a715 100644 --- a/src/main.rs +++ b/src/main.rs @@ -86,17 +86,6 @@ async fn interceptor( - - - - - - - - - - -// NEW // ===================================================================================== // ===================================================================================== // ===================================================================================== @@ -105,7 +94,7 @@ async fn interceptor( // ===================================================================================== // ===================================================================================== // ===================================================================================== -use crate::redis_events::RedisEventKind::*; // Set, Del, Unlink, Expired, Other +use crate::redis_events::RedisEventAction::*; // Set, Del, Unlink, Expired, Other pub async fn start_redis_logger(redis_url: String, hub: Addr) { let client = match redis::Client::open(redis_url) { @@ -116,54 +105,29 @@ pub async fn start_redis_logger(redis_url: String, hub: Addr) { match crate::redis_events::make_pubsub_with_kea(&client).await { Ok(pubsub) => { let (mut rx, _handle) = crate::redis_events::start_keyevent_listener(pubsub); - - // Просто читаем события и шлём их в хаб. while let Some(ev) = rx.recv().await { - match ev.kind { + + match ev.action { Set => println!("[redis] db{} SET {}", ev.db, ev.key), Del | Unlink => println!("[redis] db{} DEL {}", ev.db, ev.key), Expired => println!("[redis] db{} EXPIRED {}", ev.db, ev.key), Other(ref k) => println!("[redis] db{} {} {}", ev.db, k, ev.key), } - hub.do_send(ev.clone()); // RedisEvent помечен #[derive(Message)] + + hub.do_send(ev.clone()); } } Err(e) => eprintln!("[redis] pubsub init error: {e}"), } } -/* -async fn start_redis_logger(redis_url: &str) { - let client = match redis::Client::open(redis_url) { - Ok(c) => c, - Err(e) => { eprintln!("[redis] bad url: {e}"); return; } - }; - match crate::redis_events::make_pubsub_with_kea(&client).await { - Ok(pubsub) => { - let (mut rx, _handle) = crate::redis_events::start_keyevent_listener(pubsub); - tokio::spawn(async move { +// use actix_web::http::header; +// use actix_web::http::header::HeaderValue; +// use actix_web::body::BoxBody; +// use url::form_urlencoded; - while let Some(ev) = rx.recv().await { - // LEVENT 5,6 - match ev.kind { - Set => println!("[redis] db{} SET {}", ev.db, ev.key), - Del | Unlink => println!("[redis] db{} DEL {}", ev.db, ev.key), - Expired => println!("[redis] db{} EXPIRED {}", ev.db, ev.key), - Other(kind) => println!("[redis] db{} {} {}", ev.db, kind, ev.key), - } - - // TODO !!!!!!!!!!!!!!! - hub.do_send(ev.clone()); // ev: RedisEvent - - } - }); - } - Err(e) => eprintln!("[redis] pubsub init error: {e}"), - } -} -*/ // #[tokio::main] #[actix_web::main] @@ -176,13 +140,11 @@ async fn main() -> anyhow::Result<()> { let redis = std::sync::Arc::new(tokio::sync::Mutex::new(redis)); let redis_data = web::Data::new(redis.clone()); - // ====================================== // starting Hub let hub = WsHub::default().start(); let hub_data = web::Data::new(hub.clone()); // starting Logger tokio::spawn(start_redis_logger("redis://127.0.0.1/".to_string(), hub.clone())); - // ============================================ let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port); let payload_config = PayloadConfig::new(CONFIG.payload_size_limit.bytes() as usize); @@ -225,6 +187,59 @@ async fn main() -> anyhow::Result<()> { })) .route("/ws", web::get().to(handlers_ws::handler)) // WebSocket + + +/* + +.service( + web::resource("/ws") + + +.wrap(middleware::from_fn(|mut req: ServiceRequest, next: Next| async move { + // Уже есть Authorization? + let has_auth = req.headers().contains_key(header::AUTHORIZATION); + + if !has_auth { + // ?token=... + if let Some(token) = form_urlencoded::parse(req.query_string().as_bytes()) + .find(|(k, _)| k == "token") + .map(|(_, v)| v.into_owned()) + { + if !token.is_empty() { + let value = format!("Bearer {}", token); + req.headers_mut().insert( + header::AUTHORIZATION, + HeaderValue::from_str(&value) + .map_err(|_| actix_web::error::ErrorBadRequest("Invalid token header"))?, + ); + } + } + } + + next.call(req).await +})) + // затем твой interceptor: +// .wrap(middleware::from_fn(interceptor)) + .route(web::get().to(handlers_ws::handler)) + +) + +*/ + + + + + + + + + + + + + + + }) .bind(socket)? .run(); diff --git a/src/redis_events.rs b/src/redis_events.rs index 1abaac1f78..88a89ca859 100644 --- a/src/redis_events.rs +++ b/src/redis_events.rs @@ -1,11 +1,3 @@ -/* -TODO: Со * - -Сперва по точному совпадению -Потом перебором по * -*/ - - use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_stream::StreamExt; @@ -22,7 +14,7 @@ use redis::{ }; #[derive(Debug, Clone, Serialize)] -pub enum RedisEventKind { +pub enum RedisEventAction { Set, // Insert or Update Del, // Delete Unlink, // async Delete @@ -37,7 +29,8 @@ use actix::Message; pub struct RedisEvent { pub db: u32, pub key: String, - pub kind: RedisEventKind, +// pub value: String, + pub action: RedisEventAction, } @@ -86,6 +79,7 @@ pub fn start_keyevent_listener( } let mut stream = pubsub.on_message(); + while let Some(msg) = stream.next().await { let channel = match msg.get_channel::() { Ok(c) => c, @@ -98,12 +92,12 @@ pub fn start_keyevent_listener( // "__keyevent@0__:set" → event="set", db=0; payload = key let event = channel.rsplit(':').next().unwrap_or(""); - let kind = match event { - "set" => RedisEventKind::Set, - "del" => RedisEventKind::Del, - "unlink" => RedisEventKind::Unlink, - "expired" => RedisEventKind::Expired, - other => RedisEventKind::Other(other.to_string()), + let action = match event { + "set" => RedisEventAction::Set, + "del" => RedisEventAction::Del, + "unlink" => RedisEventAction::Unlink, + "expired" => RedisEventAction::Expired, + other => RedisEventAction::Other(other.to_string()), }; let db = channel.find('@') @@ -112,7 +106,7 @@ pub fn start_keyevent_listener( .and_then(|s| s.parse::().ok()) .unwrap_or(0); - let ev = RedisEvent { db, key: payload.clone(), kind }; + let ev = RedisEvent { db, key: payload.clone(), action }; if tx.send(ev).is_err() { break; } // closed } diff --git a/src/redis_lib.rs b/src/redis_lib.rs index 44d7b3e3d2..eb587706f6 100644 --- a/src/redis_lib.rs +++ b/src/redis_lib.rs @@ -136,6 +136,7 @@ pub async fn redis_read( })) } + /// TTL sec /// redis_save(&mut conn, "key", "val", Some(Ttl::Sec(300)), Some(SaveMode::Insert)).await?; /// diff --git a/src/ws_hub.rs b/src/ws_hub.rs index 8ec468a858..bc8f5b94e8 100644 --- a/src/ws_hub.rs +++ b/src/ws_hub.rs @@ -14,14 +14,14 @@ fn subscription_matches(sub_key: &str, key: &str) -> bool { /// Message from Hub to Session (JSON-string) +use crate::redis_events::RedisEvent; + #[derive(Message, Clone, Debug)] #[rtype(result = "()")] pub struct ServerMessage { pub event: RedisEvent, } -use crate::redis_events::RedisEvent; - /// Count of active sessions #[derive(Message)] #[rtype(result = "usize")] @@ -35,6 +35,7 @@ pub struct WsHub { next_id: SessionId, } +/// Init WsHub impl Default for WsHub { fn default() -> Self { Self { @@ -51,7 +52,7 @@ impl Actor for WsHub { - +/// Connect #[derive(Message)] #[rtype(result = "SessionId")] pub struct Connect { @@ -71,7 +72,7 @@ impl Handler for WsHub { } } - +/// Disconnect #[derive(Message)] #[rtype(result = "()")] pub struct Disconnect { @@ -99,6 +100,7 @@ impl Handler for WsHub { } } +/// SubscribeList #[derive(Message)] #[rtype(result = "Vec")] pub struct SubscribeList { @@ -125,6 +127,7 @@ impl Handler for WsHub { } } +/// Count of IDs impl Handler for WsHub { type Result = usize; @@ -133,8 +136,7 @@ impl Handler for WsHub { } } -// Subscriptions - +/// Subscribe #[derive(Message)] #[rtype(result = "()")] pub struct Subscribe { @@ -149,6 +151,7 @@ impl Handler for WsHub { } } +/// Unsubscribe #[derive(Message)] #[rtype(result = "()")] pub struct Unsubscribe { @@ -182,6 +185,9 @@ impl Handler for WsHub { } } + + + #[derive(Message)] #[rtype(result = "HashMap>")] pub struct TestGetSubs; @@ -211,6 +217,7 @@ impl WsHub { } } +/// Send Messages impl Handler for WsHub { type Result = (); @@ -218,6 +225,11 @@ impl Handler for WsHub { let targets = self.subscribers_for(&msg.key); if targets.is_empty() { return; } + // TODO: redis_read + // conn: &mut MultiplexedConnection, + let value = redis::cmd("GET").arg(&msg.key).query_async(conn).await?; + + let payload = ServerMessage { event: msg.clone() }; for sid in targets {