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 {