diff --git a/scripts/TEST.html b/scripts/TEST.html index 2070094d09..8ecf37a3c8 100644 --- a/scripts/TEST.html +++ b/scripts/TEST.html @@ -99,6 +99,7 @@ // let ws = new WebSocket(`ws://localhost:8095/ws`); + ws.onopen = () => { output.textContent = "✅ WebSocket connected."; }; @@ -122,6 +123,7 @@ } catch (e) { output.textContent += "\n\n⚠️ Invalid JSON:\n" + e.message; } + } function place(event) { textarea.value = (event || window.event).target.getAttribute("data"); sendMessage(); } diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index 563d21de2f..b04a1fa74f 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -11,6 +11,19 @@ TOKEN=$(./token.sh claims.json) ZP="00000000-0000-0000-0000-000000000001/TESTS" +echo "--------- authorization_test ----------" +TOKEN="" + put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" +TOKEN=$(./token.sh claims_system.json) + put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" +TOKEN=$(./token.sh claims_wrong_ws.json) + put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" +TOKEN=$(./token.sh claims.json) + put "00000000-0000-0000-0000-000000000002/TESTS" "Value_1" "HULY-TTL: 2" +exit + + + echo "--------- if-match ----------" put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" @@ -23,106 +36,6 @@ echo "--------- if-match ----------" get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/" -exit -Key - - - - - -Key is a string that consists of one or multiple segments separated by ‘/’. Example: foo/bar/baz. - - - - - -Segment may not contain special characters (‘$’, ‘*’, ‘?’) - - - -Key may not end with ‘/’ - - - -Segment may not be empty - - - -Key segment may be private (prefixed with ‘$’) - - - -Query - - - - - -May not contain special characters (‘*’, ‘?’) - - - -It is possible to use prefix, for listings / subscriptions (prefix ends with segment separator ‘/’) - - - - - -GET/SUBSCRIBE/.. a/b → single key - - - -GET/SUBSCRIBE/.. a/b/c/ → multiple - - - -If multiple - - - - - -select all keys starting with prefix - - - -skip keys, containing private segments to the right from the prefix - - - -example - - - - - - 1. /a/b/$c/$d, 2. /a/b/c, 3. /a/b/$c, 4. /a/b/$c/$d/e - - - -/ → [2] - - - -/a/b/ → [2] - - - -/a/b/$c/ → [3] - - - -/a/b/$c/$d/ → [4] - - - -/a/b/$c/$d → (1) - - - - - - exit diff --git a/src/handlers_http.rs b/src/handlers_http.rs index 38463dc934..16ff8ace0e 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -1,55 +1,61 @@ -use redis::aio::MultiplexedConnection; -use std::sync::Arc; -use std::collections::HashMap; -use tokio::sync::Mutex; +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +use crate::workspace_owner::workspace_check; use anyhow::anyhow; +use redis::aio::MultiplexedConnection; +use std::collections::HashMap; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::Mutex; use tracing::{error, trace}; use uuid::Uuid; -use crate::workspace_owner::workspace_check; use crate::redis_lib::{ - Ttl, SaveMode, - RedisArray, - redis_save, - redis_read, - redis_delete, - redis_list, + RedisArray, SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save, }; use actix_web::{ - HttpRequest, HttpResponse, error, Error, + Error, HttpRequest, HttpResponse, error, web::{self, Data, Json, Query}, }; - pub fn map_handler_error(err: impl std::fmt::Display) -> Error { - let msg = err.to_string(); if let Some(detail) = msg.split(" - ExtensionError: ").nth(1) { if let Some((code, text)) = detail.split_once(": ") { - let text = format!("{} {}", code, text); + let text = format!("{} {}", code, text); return match code { "400" => actix_web::error::ErrorBadRequest(text), "404" => actix_web::error::ErrorNotFound(text), "412" => actix_web::error::ErrorPreconditionFailed(text), "500" => actix_web::error::ErrorInternalServerError(text), - _ => actix_web::error::ErrorInternalServerError("unexpected error"), + _ => actix_web::error::ErrorInternalServerError("unexpected error"), }; } } actix_web::error::ErrorInternalServerError("internal error") } - /// list pub async fn list( req: HttpRequest, path: web::Path, redis: web::Data>>, ) -> Result { - workspace_check(&req)?; // Check workspace let key = path.into_inner(); @@ -57,24 +63,22 @@ pub async fn list( trace!(key, "list request"); async move || -> anyhow::Result { - let mut conn = redis.lock().await; - let entries = redis_list(&mut *conn, &key).await?; + let entries = redis_list(&mut *conn, &key).await?; Ok(HttpResponse::Ok().json(entries)) - - }().await.map_err(map_handler_error) + }() + .await + .map_err(map_handler_error) } - /// get pub async fn get( req: HttpRequest, path: web::Path, redis: web::Data>>, ) -> Result { - workspace_check(&req)?; // Check workspace let key = path.into_inner(); @@ -82,21 +86,21 @@ pub async fn get( // trace!(key, "get request"); async move || -> anyhow::Result { - let mut conn = redis.lock().await; - Ok( - redis_read(&mut *conn, &key).await? - .map(|entry| HttpResponse::Ok() + Ok(redis_read(&mut *conn, &key) + .await? + .map(|entry| { + HttpResponse::Ok() .insert_header(("ETag", &*entry.etag)) - .json(entry)) - .unwrap_or_else(|| HttpResponse::NotFound().body("empty")) - ) - - }().await.map_err(map_handler_error) + .json(entry) + }) + .unwrap_or_else(|| HttpResponse::NotFound().body("empty"))) + }() + .await + .map_err(map_handler_error) } - /// put pub async fn put( req: HttpRequest, @@ -104,54 +108,72 @@ pub async fn put( body: web::Bytes, redis: web::Data>>, ) -> Result { - workspace_check(&req)?; // Check workspace let key: String = path.into_inner(); async move || -> anyhow::Result { - - if !req.query_string().is_empty() { return Err(anyhow!("Query parameters are not allowed")); } + if !req.query_string().is_empty() { + return Err(anyhow!("Query parameters are not allowed")); + } let mut conn = redis.lock().await; - // TTL logic - let mut ttl = None; - if let Some(x) = req.headers().get("HULY-TTL") { - let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-TTL header"))?; - let secs = s.parse::().map_err(|_| anyhow!("Invalid TTL value in HULY-TTL header"))?; - ttl = Some(Ttl::Sec(secs)); - } else if let Some(x) = req.headers().get("HULY-EXPIRE-AT") { - let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-EXPIRE-AT header"))?; - let ts = s.parse::().map_err(|_| anyhow!("Invalid EXPIRE-AT value in HULY-EXPIRE-AT header"))?; - ttl = Some(Ttl::At(ts)); - } + // TTL logic + let mut ttl = None; + if let Some(x) = req.headers().get("HULY-TTL") { + let s = x.to_str().map_err(|_| anyhow!("Invalid HULY-TTL header"))?; + let secs = s + .parse::() + .map_err(|_| anyhow!("Invalid TTL value in HULY-TTL header"))?; + ttl = Some(Ttl::Sec(secs)); + } else if let Some(x) = req.headers().get("HULY-EXPIRE-AT") { + let s = x + .to_str() + .map_err(|_| anyhow!("Invalid HULY-EXPIRE-AT header"))?; + let ts = s + .parse::() + .map_err(|_| anyhow!("Invalid EXPIRE-AT value in HULY-EXPIRE-AT header"))?; + ttl = Some(Ttl::At(ts)); + } - // MODE logic - let mut mode = Some(SaveMode::Upsert); - if let Some(h) = req.headers().get("If-Match") { // `If-Match: *` - update only if the key exists - let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?; - if s == "*" { mode = Some(SaveMode::Update); } // `If-Match: *` — update only if exist - else { mode = Some(SaveMode::Equal(s.to_string())); } // `If-Match: ` — update only if current - } else if let Some(h) = req.headers().get("If-None-Match") { // `If-None-Match: *` — insert only if does not exist - let s = h.to_str().map_err(|_| anyhow!("Invalid If-None-Match header"))?; - if s == "*" { mode = Some(SaveMode::Insert); } else { return Err(anyhow!("If-None-Match must be '*'")); } - } + // MODE logic + let mut mode = Some(SaveMode::Upsert); + if let Some(h) = req.headers().get("If-Match") { + // `If-Match: *` - update only if the key exists + let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?; + if s == "*" { + mode = Some(SaveMode::Update); + } + // `If-Match: *` — update only if exist + else { + mode = Some(SaveMode::Equal(s.to_string())); + } // `If-Match: ` — update only if current + } else if let Some(h) = req.headers().get("If-None-Match") { + // `If-None-Match: *` — insert only if does not exist + let s = h + .to_str() + .map_err(|_| anyhow!("Invalid If-None-Match header"))?; + if s == "*" { + mode = Some(SaveMode::Insert); + } else { + return Err(anyhow!("If-None-Match must be '*'")); + } + } redis_save(&mut *conn, &key, &body[..], ttl, mode).await?; - return Ok(HttpResponse::Ok().body("DONE")); - - }().await.map_err(map_handler_error) + return Ok(HttpResponse::Ok().body("DONE")); + }() + .await + .map_err(map_handler_error) } - /// delete pub async fn delete( req: HttpRequest, path: web::Path, redis: web::Data>>, ) -> Result { - workspace_check(&req)?; // Check workspace let key: String = path.into_inner(); @@ -161,7 +183,21 @@ pub async fn delete( async move || -> anyhow::Result { let mut conn = redis.lock().await; - let deleted = redis_delete(&mut *conn, &key).await?; + // MODE logic + let mut mode = Some(SaveMode::Upsert); + if let Some(h) = req.headers().get("If-Match") { + // `If-Match: *` - delete only if the key exists + let s = h.to_str().map_err(|_| anyhow!("Invalid If-Match header"))?; + if s == "*" { + mode = Some(SaveMode::Update); + } + // `If-Match: *` — return error if not exist + else { + mode = Some(SaveMode::Equal(s.to_string())); + } // `If-Match: ` — delete only if current + } + + let deleted = redis_delete(&mut *conn, &key, mode).await?; let response = match deleted { true => HttpResponse::NoContent().finish(), @@ -169,6 +205,7 @@ pub async fn delete( }; Ok(response) - }().await.map_err(map_handler_error) + }() + .await + .map_err(map_handler_error) } - diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 88dccb7516..bbadf6513e 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -1,43 +1,70 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +use actix::prelude::*; use uuid::Uuid; -use actix::{prelude::*}; use crate::ws_hub::{ - WsHub, ServerMessage, SessionId, - Connect, Disconnect, - Subscribe, Unsubscribe, UnsubscribeAll, - SubscribeList, + Connect, Disconnect, ServerMessage, SessionId, Subscribe, SubscribeList, Unsubscribe, + UnsubscribeAll, WsHub, }; +use actix::{ + Actor, ActorContext, ActorFutureExt, AsyncContext, Handler, StreamHandler, WrapFuture, fut, +}; +use actix_web::{Error, HttpRequest, HttpResponse, web}; +use actix_web_actors::ws; use redis::aio::MultiplexedConnection; +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use std::collections::HashSet; use std::sync::Arc; use tokio::sync::Mutex; -use serde_json::{Value, Map, json}; -use actix::{ - Actor, - StreamHandler, - AsyncContext, - ActorContext, - fut, - ActorFutureExt, - Handler, WrapFuture -}; -use actix_web::{web, HttpRequest, HttpResponse, Error}; -use actix_web_actors::ws; -use serde::Deserialize; -use std::collections::HashSet; use crate::redis_lib::{ - Ttl, SaveMode, - RedisArray, + RedisArray, SaveMode, Ttl, deprecated_symbol, error, redis_delete, redis_list, redis_read, redis_save, - redis_read, - redis_delete, - redis_list, - error, - deprecated_symbol, }; -type JsonMap = Map; +use serde::Serialize; + +#[derive(Serialize, Default)] +struct ReturnBase<'a> { + action: &'a str, + + #[serde(skip_serializing_if = "Option::is_none")] + key: Option<&'a str>, + + #[serde(skip_serializing_if = "Option::is_none")] + data: Option<&'a str>, + + #[serde(skip_serializing_if = "Option::is_none")] + correlation: Option<&'a str>, + + #[serde(rename = "TTL", skip_serializing_if = "Option::is_none")] + ttl: Option, + + #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")] + expires_at: Option, + + #[serde(rename = "ifMatch", skip_serializing_if = "Option::is_none")] + if_match: Option<&'a str>, + + #[serde(rename = "ifNoneMatch", skip_serializing_if = "Option::is_none")] + if_none_match: Option<&'a str>, +} /// WsCommand - commands enum (put, delete, sub, unsub) #[derive(Deserialize, Debug)] @@ -66,6 +93,16 @@ pub enum WsCommand { if_none_match: Option, }, + Delete { + #[serde(default)] + correlation: Option, + key: String, + + #[serde(rename = "ifMatch")] + #[serde(default)] + if_match: Option, + }, + Get { #[serde(default)] correlation: Option, @@ -78,16 +115,6 @@ pub enum WsCommand { key: String, }, - Delete { - #[serde(default)] - correlation: Option, - key: String, - - #[serde(rename = "ifMatch")] - #[serde(default)] - if_match: Option, - }, - Sub { #[serde(default)] correlation: Option, @@ -117,7 +144,6 @@ pub struct WsSession { pub claims: Option, } - /// Actor External trait: must be in separate impl block impl Actor for WsSession { type Context = ws::WebsocketContext; @@ -131,33 +157,35 @@ impl Actor for WsSession { self.hub .send(Connect { addr: recipient }) .into_actor(self) - .map(|res, act, _ctx| { - match res { - Ok(id) => { - act.id = id; - tracing::info!("WebSocket connected: {id}"); - } - Err(e) => { - tracing::error!("WebSocket failed connect to hub: {e}"); - _ctx.stop(); - } + .map(|res, act, _ctx| match res { + Ok(id) => { + act.id = id; + tracing::info!("WebSocket connected: {id}"); + } + Err(e) => { + tracing::error!("WebSocket failed connect to hub: {e}"); + _ctx.stop(); } }) .wait(ctx); // waiting for ID } fn stopped(&mut self, _ctx: &mut Self::Context) { - if self.id != 0 { self.hub.do_send(Disconnect { session_id: self.id }); } - tracing::info!("WebSocket disconnected: {:?}",&self.id); + if self.id != 0 { + self.hub.do_send(Disconnect { + session_id: self.id, + }); + } + tracing::info!("WebSocket disconnected: {:?}", &self.id); } - } impl actix::Handler for WsSession { type Result = (); fn handle(&mut self, msg: ServerMessage, ctx: &mut Self::Context) { - let json = serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); + let json = + serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); ctx.text(json); } } @@ -166,16 +194,12 @@ impl actix::Handler for WsSession { impl StreamHandler> for WsSession { fn handle(&mut self, msg: Result, ctx: &mut Self::Context) { match msg { - Ok(ws::Message::Text(text)) => { - // println!("Message: {}", text); - match serde_json::from_str::(&text) { - Ok(cmd) => self.handle_command(cmd, ctx), - Err(err) => ctx.text(format!("Invalid JSON: {}", err)), - } - } + Ok(ws::Message::Text(text)) => match serde_json::from_str::(&text) { + Ok(cmd) => self.handle_command(cmd, ctx), + Err(err) => ctx.text(format!("Invalid JSON: {}", err)), + }, Ok(ws::Message::Ping(msg)) => ctx.pong(&msg), Ok(ws::Message::Close(reason)) => { - // println!("Closing WS: {:?}", reason); ctx.close(reason); ctx.stop(); } @@ -184,287 +208,352 @@ impl StreamHandler> for WsSession { } } +use crate::workspace_owner::check_workspace_core; + /// 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(()) + check_workspace_core(claims, key) } - fn wait_and_send( + fn fut_send( &mut self, ctx: &mut ws::WebsocketContext, - fut: F, - mut base: JsonMap, - ) - where - F: std::future::Future> + 'static, - { + fut: impl Future> + 'static, + mut base: Value, + ) { ctx.wait( fut::wrap_future(fut).map(move |res, _actor: &mut Self, ctx| { + // if !base.is_object() { base = json!({ "base": base }); } + let obj = base.as_object_mut().unwrap(); match res { - Ok(extra) => { base.extend(extra); } - Err(err) => { base.insert("error".into(), json!(err)); } + Ok(Value::Object(extra)) => { + obj.extend(extra); + } + Ok(v) => { + obj.insert("extra".into(), v); + } + Err(err) => { + obj.insert("error".into(), json!(err)); + } } - ctx.text(Value::Object(base).to_string()); - }) + ctx.text(base.to_string()); + }), ); } /// When valid JSON recieved for WsSession fn handle_command(&mut self, cmd: WsCommand, ctx: &mut ws::WebsocketContext) { match cmd { + WsCommand::Put { + key, + data, + expires_at, + ttl, + if_match, + if_none_match, + correlation, + } => { + tracing::info!( + "PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}", + &key, + &data, + &expires_at, + &ttl, + &correlation + ); - WsCommand::Put { key, data, expires_at, ttl, if_match, if_none_match, correlation } => { + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } - tracing::info!("PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}", &key, &data, &expires_at, &ttl, &correlation); + let redis = self.redis.clone(); - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let base = serde_json::json!(ReturnBase { + action: "put", + key: Some(key.as_str()), + data: Some(data.as_str()), + correlation: correlation.as_deref(), + ttl, + expires_at, + if_match: if_match.as_deref(), + if_none_match: if_none_match.as_deref(), + }); - let redis = self.redis.clone(); - - let mut base = JsonMap::new(); - base.insert("action".into(), json!("put")); - base.insert("key".into(), json!(&key)); - base.insert("data".into(), json!(&data)); - if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } - if let Some(x) = &expires_at { base.insert("expiresAt".into(), json!(x)); } - if let Some(x) = &ttl { base.insert("TTL".into(), json!(x)); } - if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); } - if let Some(x) = &if_none_match { base.insert("ifNoneMatch".into(),json!(x)); } - - let fut = async move { - - // TTL logic + let fut = async move { + // TTL logic let real_ttl = if let Some(secs) = ttl { - Some(Ttl::Sec(secs as usize)) - } else if let Some(timestamp) = expires_at { - Some(Ttl::At(timestamp)) - } else { - None - }; + Some(Ttl::Sec(secs as usize)) + } else if let Some(timestamp) = expires_at { + Some(Ttl::At(timestamp)) + } else { + None + }; - // MODE logic - let mut mode = Some(SaveMode::Upsert); - if let Some(s) = if_match { // `If-Match: *` - update only if the key exists - if s == "*" { // `If-Match: *` — update only if exist - mode = Some(SaveMode::Update); - } else { // `If-Match: ` — update only if current - mode = Some(SaveMode::Equal(s.to_string())); - } - } else if let Some(s) = if_none_match { // `If-None-Match: *` — insert only if does not exist - if s == "*" { - mode = Some(SaveMode::Insert); - } else { - return Err::("ifNoneMatch must contain only '*'".into()); - } - } + // MODE logic + let mut mode = Some(SaveMode::Upsert); + if let Some(s) = if_match { + // `If-Match: *` - update only if the key exists + if s == "*" { + // `If-Match: *` — update only if exist + mode = Some(SaveMode::Update); + } else { + // `If-Match: ` — update only if current + mode = Some(SaveMode::Equal(s.to_string())); + } + } else if let Some(s) = if_none_match { + // `If-None-Match: *` — insert only if does not exist + if s == "*" { + mode = Some(SaveMode::Insert); + } else { + return Err("ifNoneMatch must contain only '*'".into()); + } + } - let mut conn = redis.lock().await; + let mut conn = redis.lock().await; - redis_save(&mut *conn, &key, &data, real_ttl, mode) - .await - .map_err(|e| e.to_string())?; + redis_save(&mut *conn, &key, &data, real_ttl, mode) + .await + .map_err(|e| e.to_string())?; - let mut extra = JsonMap::new(); - extra.insert("response".into(), json!("OK")); - Ok::(extra) + Ok(json!({"result": "OK"})) + }; - }; + self.fut_send(ctx, fut, base); + } - self.wait_and_send(ctx, fut, base); - } - - - WsCommand::Delete { key, correlation, if_match } => { + WsCommand::Delete { + key, + correlation, + if_match, + } => { tracing::info!("DELETE {} correlation:{:?}", &key, &correlation); - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } - let redis = self.redis.clone(); + let redis = self.redis.clone(); - let mut base = JsonMap::new(); - base.insert("action".into(), json!("delete")); - base.insert("key".into(), json!(&key)); - if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } - if let Some(x) = &if_match { base.insert("ifMatch".into(), json!(x)); } + let base = serde_json::json!(ReturnBase { + action: "delete", + key: Some(key.as_str()), + correlation: correlation.as_deref(), + if_match: if_match.as_deref(), + ..Default::default() + }); - let fut = async move { + let fut = async move { + let mut conn = redis.lock().await; - let mut conn = redis.lock().await; + // MODE logic + let mut mode = Some(SaveMode::Upsert); + if let Some(s) = if_match { + // `If-Match: *` - delete only if the key exists + if s == "*" { + // `If-Match: *` — return error if not exist + mode = Some(SaveMode::Update); + } else { + // `If-Match: ` — update only if current + mode = Some(SaveMode::Equal(s.to_string())); + } + } - let deleted = redis_delete(&mut *conn, &key).await.map_err(|e| e.to_string())?; + let deleted = redis_delete(&mut *conn, &key, mode) + .await + .map_err(|e| e.to_string())?; - if deleted { - let mut extra = JsonMap::new(); - extra.insert("response".into(), json!("OK")); - Ok::(extra) - } else { - Err::("not found".into()) - } + if deleted { + Ok(json!({"result": "OK"})) + } else { + Err("not found".into()) + } + }; - }; - - self.wait_and_send(ctx, fut, base); + self.fut_send(ctx, fut, base); } WsCommand::Get { key, correlation } => { tracing::info!("GET {} correlation:{:?}", &key, &correlation); - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } - let redis = self.redis.clone(); + let redis = self.redis.clone(); - let mut base = JsonMap::new(); - base.insert("action".into(), json!("get")); - base.insert("key".into(), json!(&key)); - if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } + let base = serde_json::json!(ReturnBase { + action: "get", + key: Some(key.as_str()), + correlation: correlation.as_deref(), + ..Default::default() + }); - let fut = async move { + let fut = async move { + let mut conn = redis.lock().await; - let mut conn = redis.lock().await; + let data_opt = redis_read(&mut *conn, &key) + .await + .map_err(|e| e.to_string())?; - let data_opt = redis_read(&mut *conn, &key) - .await - .map_err(|e| e.to_string())?; + match data_opt { + Some(data) => { + let data_value = + serde_json::to_value(&data).map_err(|e| e.to_string())?; + Ok(json!({"result": data_value})) + } + None => Err("not found".into()), + } + }; - match data_opt { - Some(data) => { - let mut extra = JsonMap::new(); - let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?; - extra.insert("response".into(), data_value); - Ok::(extra) - } - None => Err::("not found".into()) - } - }; - - self.wait_and_send(ctx, fut, base); + self.fut_send(ctx, fut, base); } WsCommand::List { key, correlation } => { tracing::info!("LIST {:?} correlation: {:?}", &key, &correlation); - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } - let redis = self.redis.clone(); + let redis = self.redis.clone(); - let mut base = JsonMap::new(); - base.insert("action".into(), json!("get")); - base.insert("key".into(), json!(&key)); - if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } + let base = serde_json::json!(ReturnBase { + action: "list", + key: Some(key.as_str()), + correlation: correlation.as_deref(), + ..Default::default() + }); - let fut = async move { + let fut = async move { + let mut conn = redis.lock().await; + let data = redis_list(&mut *conn, &key) + .await + .map_err(|e| e.to_string())?; + Ok(json!({ "result": data })) + }; - let mut conn = redis.lock().await; - - let data = redis_list(&mut *conn, &key).await.map_err(|e| e.to_string())?; - - let mut extra = JsonMap::new(); - let data_value = serde_json::to_value(&data).map_err(|e| e.to_string())?; - extra.insert("response".into(), data_value); - Ok::(extra) - }; - - self.wait_and_send(ctx, fut, base); + self.fut_send(ctx, fut, base); } + WsCommand::Sub { key, correlation } => { + // LEVENT 3 + tracing::info!("SUB {} correlation: {:?}", &key, &correlation); - WsCommand::Sub { key, correlation } => { - // LEVENT 3 - tracing::info!("SUB {} correlation: {:?}", &key, &correlation); + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { self.ws_error(ctx, e); return; } + let mut obj = serde_json::json!(ReturnBase { + action: "sub", + key: Some(key.as_str()), + correlation: correlation.as_deref(), + ..Default::default() + }); - let mut obj = JsonMap::new(); - obj.insert("action".into(), json!("sub")); - obj.insert("key".into(), json!(key)); - if let Some(c) = correlation { obj.insert("correlation".into(), json!(c)); } + let map = obj.as_object_mut().unwrap(); - if deprecated_symbol(&key) { - obj.insert("error".into(), json!("Deprecated symbol in key")); - } else { - self.hub.do_send(Subscribe { session_id: self.id, key: key.clone() }); - } + if deprecated_symbol(&key) { + map.insert("error".into(), json!("Deprecated symbol in key")); + } else { + self.hub.do_send(Subscribe { + session_id: self.id, + key: key.clone(), + }); + map.insert("result".into(), json!("OK")); + } - ctx.text(Value::Object(obj).to_string()); - } - - WsCommand::Unsub { key, correlation } => { - // LEVENT 4 - tracing::info!("UNSUB {} correlation: {:?}", &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)); - if let Some(c) = correlation { obj.insert("correlation".into(), json!(c)); } - - if key == "*" { - self.hub.do_send(UnsubscribeAll { session_id: self.id }); - } else { - if deprecated_symbol(&key) { - obj.insert("error".into(), json!("Deprecated symbol in key")); - } else { - self.hub.do_send(Unsubscribe { session_id: self.id, key: key.clone() }); - } - }; - - ctx.text(Value::Object(obj).to_string()); - } - - WsCommand::Sublist { correlation } => { - tracing::info!("SUBLIST correlation: {:?}", &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)); } - - let hub = self.hub.clone(); - let id = self.id; - - let fut = async move { - let keys = hub.send(SubscribeList { session_id: id }).await.unwrap_or_default(); - let mut extra = JsonMap::new(); - extra.insert("response".into(), serde_json::to_value(&keys).map_err(|e| e.to_string())? ); - Ok::(extra) - }; - - self.wait_and_send(ctx, fut, base); + ctx.text(obj.to_string()); } - // End of commands + WsCommand::Unsub { key, correlation } => { + // LEVENT 4 + tracing::info!("UNSUB {} correlation: {:?}", &key, &correlation); + + let mut obj = serde_json::json!(ReturnBase { + action: "unsub", + key: Some(key.as_str()), + correlation: correlation.as_deref(), + ..Default::default() + }); + + let map = obj.as_object_mut().unwrap(); + + if key == "*" { + self.hub.do_send(UnsubscribeAll { + session_id: self.id, + }); + map.insert("result".into(), json!("OK")); + } else { + // Check workspace + if let Err(e) = self.workspace_check_ws(&key) { + self.ws_error(ctx, e); + return; + } + + if deprecated_symbol(&key) { + map.insert("error".into(), json!("Deprecated symbol in key")); + } else { + map.insert("result".into(), json!("OK")); + self.hub.do_send(Unsubscribe { + session_id: self.id, + key: key.clone(), + }); + } + }; + + ctx.text(obj.to_string()); + } + + WsCommand::Sublist { correlation } => { + tracing::info!("SUBLIST correlation: {:?}", &correlation); + + // w/o Check workspace! + + let base = serde_json::json!(ReturnBase { + action: "list", + correlation: correlation.as_deref(), + ..Default::default() + }); + + let hub = self.hub.clone(); + let id = self.id; + + let fut = async move { + let keys = hub + .send(SubscribeList { session_id: id }) + .await + .unwrap_or_default(); + Ok(json!({ "result": keys })) + }; + + self.fut_send(ctx, fut, base); + } // End of commands } } - } // ---- auth -use actix_web::{HttpMessage,error}; -use url::form_urlencoded; -use jsonwebtoken::{decode, DecodingKey, Validation, Algorithm}; use crate::CONFIG; +use actix_web::{HttpMessage, error}; +use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode}; +use url::form_urlencoded; pub async fn handler( req: HttpRequest, @@ -472,23 +561,24 @@ pub async fn handler( redis: 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()) - }); + 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 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"))?; - + 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) } diff --git a/src/main.rs b/src/main.rs index db6be5c846..b66a84b544 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,10 +23,14 @@ use actix_web::{ App, Error, HttpMessage, HttpRequest, HttpResponse, HttpServer, body::MessageBody, dev::{ServiceRequest, ServiceResponse}, + error::ErrorBadRequest, + http::header::{AUTHORIZATION, HeaderValue}, middleware::{self, Next}, web::{self, Data, PayloadConfig}, }; +use url::form_urlencoded; + use actix_web_actors::ws; use tracing::info; @@ -41,15 +45,10 @@ use crate::redis_lib::redis_connect; mod workspace_owner; -// == =hub === mod redis_events; mod ws_hub; +use crate::ws_hub::{ServerMessage, TestGetSubs, WsHub}; use actix::prelude::*; -use crate::ws_hub::{WsHub, ServerMessage, - TestGetSubs, -}; - -// === /hub === use config::CONFIG; @@ -70,54 +69,55 @@ fn initialize_tracing(level: tracing::Level) { .init(); } -// #[allow(dead_code)] async fn interceptor( - request: ServiceRequest, + mut request: ServiceRequest, next: Next, ) -> Result, Error> { + // Authorization/token patch + if request.headers().get(AUTHORIZATION).is_none() { + if let Some(qs) = request.uri().query() { + if let Some(token) = form_urlencoded::parse(qs.as_bytes()) + .find(|(k, _)| k == "token") + .map(|(_, v)| v.into_owned()) + { + let auth_value = HeaderValue::from_str(&format!("Bearer {}", token)) + .map_err(|_| ErrorBadRequest("Malformed token"))?; + request.headers_mut().insert(AUTHORIZATION, auth_value); + } + } + } let secret = SecretString::new(CONFIG.token_secret.clone().into_boxed_str()); - let claims = request.extract_claims(&secret)?; - // TODO: сделать это здесь - request.extensions_mut().insert(claims.to_owned()); - // TODO потом исправить hulyrs: extract_claims - next.call(request).await } - - -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== -// ===================================================================================== 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) { Ok(c) => c, - Err(e) => { eprintln!("[redis] bad url: {e}"); return; } + 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); while let Some(ev) = rx.recv().await { - - 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), - } + /* + 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()); } @@ -126,14 +126,6 @@ pub async fn start_redis_logger(redis_url: String, hub: Addr) { } } - - -// use actix_web::http::header; -// use actix_web::http::header::HeaderValue; -// use actix_web::body::BoxBody; -// use url::form_urlencoded; - - // #[tokio::main] #[actix_web::main] async fn main() -> anyhow::Result<()> { @@ -146,12 +138,14 @@ async fn main() -> anyhow::Result<()> { let redis_data = web::Data::new(redis.clone()); // starting Hub - // let hub = WsHub::default().start(); let hub = WsHub::new(redis.clone()).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())); + 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); @@ -167,85 +161,37 @@ async fn main() -> anyhow::Result<()> { App::new() .app_data(payload_config.clone()) .app_data(redis_data.clone()) - .app_data(hub_data.clone()) + .app_data(hub_data.clone()) .wrap(middleware::Logger::default()) .wrap(cors) .service( web::scope("/api") .wrap(middleware::from_fn(interceptor)) - .route("/{key:.+/}", web::get().to(handlers_http::list)) + .route("/{key:.+/}", web::get().to(handlers_http::list)) .route("/{key:.+}", web::get().to(handlers_http::get)) - .route("/{key:.+}", web::put().to(handlers_http::put)) - .route("/{key:.+}", web::delete().to(handlers_http::delete)) + .route("/{key:.+}", web::put().to(handlers_http::put)) + .route("/{key:.+}", web::delete().to(handlers_http::delete)), ) .route("/status", web::get().to(async || "ok")) - - // .route("/stat", web::get().to(ws_hub::stat)) - .route("/stat2", web::get().to(|hub: web::Data>| async move { - let count = hub.send(crate::ws_hub::Count).await.unwrap_or(0); - HttpResponse::Ok().json(serde_json::json!({ "connections": count })) - })) - - .route("/subs", web::get().to(|hub: web::Data>| async move { - match hub.send(TestGetSubs).await { - Ok(subs) => HttpResponse::Ok().json(subs), - Err(_) => HttpResponse::InternalServerError().body("Failed to get subscriptions"), - } - })) - - .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)) - -) - -*/ - - - - - - - - - - - - - - - + .route( + "/stat2", + web::get().to(|hub: web::Data>| async move { + let count = hub.send(crate::ws_hub::Count).await.unwrap_or(0); + HttpResponse::Ok().json(serde_json::json!({ "connections": count })) + }), + ) + .route( + "/subs", + web::get().to(|hub: web::Data>| async move { + match hub.send(TestGetSubs).await { + Ok(subs) => HttpResponse::Ok().json(subs), + Err(_) => { + HttpResponse::InternalServerError().body("Failed to get subscriptions") + } + } + }), + ) + .route("/ws", web::get().to(handlers_ws::handler)) // WebSocket }) .bind(socket)? .run(); diff --git a/src/redis_events.rs b/src/redis_events.rs index 88a89ca859..2d3c2150df 100644 --- a/src/redis_events.rs +++ b/src/redis_events.rs @@ -1,24 +1,35 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_stream::StreamExt; use serde::Serialize; - use redis::{ - self, - AsyncCommands, - RedisResult, - Client, - aio::{PubSub, ConnectionLike}, + self, AsyncCommands, Client, RedisResult, + aio::{ConnectionLike, PubSub}, }; #[derive(Debug, Clone, Serialize)] pub enum RedisEventAction { - Set, // Insert or Update - Del, // Delete - Unlink, // async Delete - Expired, // TTL Delete + Set, // Insert or Update + Del, // Delete + Unlink, // async Delete + Expired, // TTL Delete Other(String), } @@ -29,17 +40,21 @@ use actix::Message; pub struct RedisEvent { pub db: u32, pub key: String, -// pub value: String, + // pub value: String, pub action: RedisEventAction, } - /// Notifications: keyevent + generic + expired = "Egx" (no keyspace) async fn try_enable_keyspace_notifications(conn: &mut C) -> RedisResult<()> where C: ConnectionLike + Send, { - let _: String = redis::cmd("CONFIG").arg("SET").arg("notify-keyspace-events").arg("E$gx").query_async(conn).await?; + let _: String = redis::cmd("CONFIG") + .arg("SET") + .arg("notify-keyspace-events") + .arg("E$gx") + .query_async(conn) + .await?; Ok(()) } @@ -82,33 +97,46 @@ pub fn start_keyevent_listener( while let Some(msg) = stream.next().await { let channel = match msg.get_channel::() { - Ok(c) => c, - Err(e) => { eprintln!("[redis_events] bad channel: {e}"); continue; } + Ok(c) => c, + Err(e) => { + eprintln!("[redis_events] bad channel: {e}"); + continue; + } }; let payload = match msg.get_payload::() { - Ok(p) => p, - Err(e) => { eprintln!("[redis_events] bad payload: {e}"); continue; } + Ok(p) => p, + Err(e) => { + eprintln!("[redis_events] bad payload: {e}"); + continue; + } }; // "__keyevent@0__:set" → event="set", db=0; payload = key let event = channel.rsplit(':').next().unwrap_or(""); - let action = match event { - "set" => RedisEventAction::Set, - "del" => RedisEventAction::Del, - "unlink" => RedisEventAction::Unlink, - "expired" => RedisEventAction::Expired, - other => RedisEventAction::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('@') - .and_then(|at| channel.get(at + 1..)) - .and_then(|rest| rest.find("__:").map(|end| &rest[..end])) - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); + let db = channel + .find('@') + .and_then(|at| channel.get(at + 1..)) + .and_then(|rest| rest.find("__:").map(|end| &rest[..end])) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); - let ev = RedisEvent { db, key: payload.clone(), action }; + let ev = RedisEvent { + db, + key: payload.clone(), + action, + }; - if tx.send(ev).is_err() { break; } // closed + if tx.send(ev).is_err() { + break; + } // closed } }); diff --git a/src/redis_lib.rs b/src/redis_lib.rs index eb587706f6..415d24bf77 100644 --- a/src/redis_lib.rs +++ b/src/redis_lib.rs @@ -1,3 +1,18 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + use crate::config::{CONFIG, RedisMode}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -5,21 +20,20 @@ use std::time::{SystemTime, UNIX_EPOCH}; #[derive(serde::Serialize)] pub enum Ttl { Sec(usize), // EX - At(u64), // EXAT (timestamp in seconds) + At(u64), // EXAT (timestamp in seconds) } #[derive(Debug)] pub enum SaveMode { - Upsert, // default: set or overwrite - Insert, // only if not exists (NX) - Update, // only if exists (XX) + Upsert, // default: set or overwrite + Insert, // only if not exists (NX) + Update, // only if exists (XX) Equal(String), // only if md5 matches provided } use redis::{ - AsyncCommands, RedisResult, - ToRedisArgs, - Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, aio::MultiplexedConnection + AsyncCommands, Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, RedisResult, + ToRedisArgs, aio::MultiplexedConnection, }; use url::Url; @@ -30,24 +44,28 @@ pub struct RedisArray { pub key: String, pub data: String, pub expires_at: u64, // sec to expire TTL - pub etag: String, // md5 hash (data) + pub etag: String, // md5 hash (data) } /// return Error pub fn error(code: u16, msg: impl Into) -> redis::RedisResult { let msg = msg.into(); let full = format!("{}: {}", code, msg); - Err(redis::RedisError::from(( redis::ErrorKind::ExtensionError, "", full ))) + Err(redis::RedisError::from(( + redis::ErrorKind::ExtensionError, + "", + full, + ))) } /// Check for redis-deprecated symbols pub fn deprecated_symbol(s: &str) -> bool { - s.chars().any(|c| matches!( - c, - '*' | '?' | '[' | ']' | '\\' | - '\0'..='\x1F' | '\x7F' | - '"' | '\'' - )) + s.chars().any(|c| { + matches!( + c, + '*' | '?' | '[' | ']' | '\\' | '\0'..='\x1F' | '\x7F' | '"' | '\'' + ) + }) } pub fn deprecated_symbol_error(s: &str) -> redis::RedisResult<()> { @@ -63,70 +81,84 @@ pub async fn redis_list( conn: &mut MultiplexedConnection, key: &str, ) -> redis::RedisResult> { - deprecated_symbol_error(key)?; - if !key.ends_with('/') { return error(412, "Key must end with slash"); } + if !key.ends_with('/') { + return error(412, "Key must end with slash"); + } let pattern = format!("{key}*"); let mut cursor = 0u64; let mut results = Vec::new(); loop { - let mut cmd = redis::cmd("SCAN"); + let mut cmd = redis::cmd("SCAN"); cmd.arg(cursor); cmd.arg("MATCH").arg(&pattern); // cmd.arg("COUNT").arg(100); // Optionally adjust batch size let (next_cursor, keys): (u64, Vec) = cmd.query_async(conn).await?; - for k in keys { + for k in keys { + // Check for $-security path + if k.strip_prefix(key).map_or(false, |s| s.contains('$')) { + continue; + } - // Check for $-security path - if k.strip_prefix(key).map_or(false, |s| s.contains('$')) { continue; } + // Get value + let value: Option = redis::cmd("GET").arg(&k).query_async(conn).await?; + let Some(value) = value else { + continue; + }; // Old and deleted - // Get value - let value: Option = redis::cmd("GET").arg(&k).query_async(conn).await?; - let Some(value) = value else { continue; }; // Old and deleted + // Get TTL + let ttl: i64 = redis::cmd("TTL").arg(&k).query_async(conn).await?; + if ttl >= 0 { + results.push(RedisArray { + key: k, + data: value.clone(), + expires_at: ttl as u64, + etag: hex::encode(md5::compute(&value).0), + }); + } + } - // Get TTL - let ttl: i64 = redis::cmd("TTL").arg(&k).query_async(conn).await?; - if ttl >= 0 { - results.push(RedisArray { - key: k, - data: value.clone(), - expires_at: ttl as u64, - etag: hex::encode(md5::compute(&value).0), - }); - } - } - - if next_cursor == 0 { break;} - cursor = next_cursor; + if next_cursor == 0 { + break; + } + cursor = next_cursor; } Ok(results) } - /// redis_read(&connection,key) #[allow(dead_code)] pub async fn redis_read( conn: &mut MultiplexedConnection, key: &str, ) -> redis::RedisResult> { - deprecated_symbol_error(key)?; - if key.ends_with('/') { return error(412, "Key must not end with a slash"); } + if key.ends_with('/') { + return error(412, "Key must not end with a slash"); + } let data: Option = redis::cmd("GET").arg(key).query_async(conn).await?; - let Some(data) = data else { return Ok(None); }; + let Some(data) = data else { + return Ok(None); + }; let ttl: i64 = redis::cmd("TTL").arg(key).query_async(conn).await?; - if ttl == -1 { return error(500, "TTL not set"); } - if ttl == -2 { return error(500, "Key not found"); } - if ttl < 0 { return error(500, "Unknown TTL error"); } + if ttl == -1 { + return error(500, "TTL not set"); + } + if ttl == -2 { + return error(500, "Key not found"); + } + if ttl < 0 { + return error(500, "Unknown TTL error"); + } Ok(Some(RedisArray { key: key.to_string(), @@ -136,7 +168,6 @@ pub async fn redis_read( })) } - /// TTL sec /// redis_save(&mut conn, "key", "val", Some(Ttl::Sec(300)), Some(SaveMode::Insert)).await?; /// @@ -155,25 +186,33 @@ pub async fn redis_save( ttl: Option, mode: Option, ) -> RedisResult<()> { - deprecated_symbol_error(&key)?; - if key.ends_with('/') { return error(412, "Key must not end with a slash"); } + if key.ends_with('/') { + return error(412, "Key must not end with a slash"); + } // TTL logic let sec = match ttl { - Some(Ttl::Sec(secs)) => secs, - Some(Ttl::At(timestamp)) => { - let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(); - if timestamp <= now { - return error(400, "TTL timestamp exceeds MAX_TTL limit"); - } - (timestamp - now) as usize - } - None => CONFIG.max_ttl, + Some(Ttl::Sec(secs)) => secs, + Some(Ttl::At(timestamp)) => { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + if timestamp <= now { + return error(400, "TTL timestamp exceeds MAX_TTL limit"); + } + (timestamp - now) as usize + } + None => CONFIG.max_ttl, }; - if sec == 0 { return error(400, "TTL must be > 0"); } - if sec > CONFIG.max_ttl { return error(412, "TTL exceeds MAX_TTL"); } + if sec == 0 { + return error(400, "TTL must be > 0"); + } + if sec > CONFIG.max_ttl { + return error(412, "TTL exceeds MAX_TTL"); + } let mut cmd = redis::cmd("SET"); cmd.arg(key).arg(value).arg("EX").arg(sec); @@ -182,25 +221,39 @@ pub async fn redis_save( let mode = mode.unwrap_or(SaveMode::Upsert); match mode { - SaveMode::Upsert => {} // none - SaveMode::Insert => { cmd.arg("NX"); } // if NOT Exist + SaveMode::Insert => { + cmd.arg("NX"); + } // if NOT Exist - SaveMode::Update => { cmd.arg("XX"); } // if Exist + SaveMode::Update => { + cmd.arg("XX"); + } // if Exist - SaveMode::Equal(ref expected_md5) => { // if md5 === actual_md5 - let current_value: Option = redis::cmd("GET").arg(key).query_async(conn).await?; + SaveMode::Equal(ref expected_md5) => { + // if md5 === actual_md5 + let current_value: Option = + redis::cmd("GET").arg(key).query_async(conn).await?; if let Some(existing) = current_value { let actual_md5 = hex::encode(md5::compute(&existing).0); - if &actual_md5 != expected_md5 { return error(412, format!("md5 mismatch, current: {}, expected: {}", actual_md5, expected_md5)); } - } else { return error(404, "Equal: key does not exist"); } + if &actual_md5 != expected_md5 { + return error( + 412, + format!( + "md5 mismatch, current: {}, expected: {}", + actual_md5, expected_md5 + ), + ); + } + } else { + return error(404, "Equal: key does not exist"); + } } } + // execute let result: Option = cmd.query_async(conn).await?; -// // execute -// cmd.query_async::(&mut *conn).await?; if result.is_none() { match mode { @@ -213,28 +266,54 @@ pub async fn redis_save( Ok(()) } - /// redis_delete(&connection,key) -#[allow(dead_code)] pub async fn redis_delete( conn: &mut MultiplexedConnection, key: &str, -) -> redis::RedisResult { - + mode: Option, // <— добавили +) -> RedisResult { deprecated_symbol_error(key)?; - if key.ends_with('/') { return error(412, "Key must not end with a slash"); } + if key.ends_with('/') { + return error(412, "Key must not end with a slash"); + } - let deleted: i32 = redis::cmd("DEL") - .arg(key) - .query_async(conn) - .await?; + let mode = mode.unwrap_or(SaveMode::Upsert); + + match mode { + SaveMode::Equal(ref expected_md5) => { + let current: Option = redis::cmd("GET").arg(key).query_async(conn).await?; + match current { + None => return error(404, "Equal: key does not exist"), + Some(val) => { + let actual_md5 = hex::encode(md5::compute(&val).0); + if &actual_md5 != expected_md5 { + return error( + 412, + format!( + "md5 mismatch, current: {}, expected: {}", + actual_md5, expected_md5 + ), + ); + } + } + } + } + SaveMode::Insert => { + return error(412, "Insert mode is not supported for delete"); + } + SaveMode::Update | SaveMode::Upsert => {} + } + + let deleted: i32 = redis::cmd("DEL").arg(key).query_async(conn).await?; + + if deleted == 0 && matches!(mode, SaveMode::Equal(_)) { + return error(404, "Delete: key does not exist"); + } Ok(deleted > 0) } - - /// redis_connect() pub async fn redis_connect() -> anyhow::Result { let default_port = match CONFIG.redis_mode { @@ -292,4 +371,3 @@ pub async fn redis_connect() -> anyhow::Result { Ok(conn) } - diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs index 24fade1ebe..3f369c9cdc 100644 --- a/src/workspace_owner.rs +++ b/src/workspace_owner.rs @@ -1,43 +1,62 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +use actix_web::{Error, HttpMessage, HttpRequest}; use hulyrs::services::jwt::Claims; use uuid::Uuid; -use actix_web::{ Error, HttpMessage, HttpRequest, error }; -/// Checking workspace in Authorization -pub fn workspace_check(req: &HttpRequest) -> Result<(), Error> { - let extensions = req.extensions(); - - // Get key - let key = req - .match_info() - .get("key") - .ok_or_else(|| error::ErrorBadRequest("Missing key in URL path"))?; - - // Get workspace - let path_ws = match key.find('/') { - Some(x) if x > 0 => &key[..x], - _ => return Err(error::ErrorBadRequest("Invalid key: missing workspace")), - }; - - let claims = extensions - .get::() - .ok_or_else(|| error::ErrorUnauthorized("Missing auth claims"))?; - - // is_system - allowed to all +// common checker +pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> { if claims.is_system() { return Ok(()); } - // else - check workplace let jwt_workspace = claims .workspace .as_ref() - .ok_or_else(|| error::ErrorForbidden("Missing workspace in token"))?; - - let path_ws_uuid = Uuid::parse_str(path_ws).map_err(|_| error::ErrorBadRequest("Invalid workspace UUID"))?; + .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(error::ErrorForbidden("Workspace mismatch")); + return Err("Workspace mismatch"); } Ok(()) } + +/// HTTP API +pub fn workspace_check(req: &HttpRequest) -> Result<(), actix_web::Error> { + let key = req + .match_info() + .get("key") + .ok_or_else(|| actix_web::error::ErrorBadRequest("Missing key in URL path"))?; + let claims = req + .extensions() + .get::() + .cloned() + .ok_or_else(|| actix_web::error::ErrorUnauthorized("Missing auth claims"))?; + + match check_workspace_core(&claims, key) { + Ok(()) => Ok(()), + Err(msg) => Err(actix_web::error::ErrorUnauthorized(msg)), + } +} diff --git a/src/ws_hub.rs b/src/ws_hub.rs index e9611ac505..21b3f71d44 100644 --- a/src/ws_hub.rs +++ b/src/ws_hub.rs @@ -1,7 +1,24 @@ -use std::collections::{ HashMap, HashSet }; +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +use std::collections::{HashMap, HashSet}; fn subscription_matches(sub_key: &str, key: &str) -> bool { - if sub_key == key { return true; } + if sub_key == key { + return true; + } if sub_key.ends_with('/') && key.starts_with(sub_key) { let rest = &key[sub_key.len()..]; return !rest.contains('$'); @@ -9,16 +26,16 @@ fn subscription_matches(sub_key: &str, key: &str) -> bool { false } -use crate::redis_events::{ RedisEvent, RedisEventAction }; +use crate::redis_events::{RedisEvent, RedisEventAction}; use serde::Serialize; #[derive(Message, Clone, Serialize, Debug)] #[rtype(result = "()")] pub struct ServerMessage { #[serde(flatten)] - pub event: RedisEvent, // поля RedisEvent «вливаются» в корень JSON + pub event: RedisEvent, #[serde(skip_serializing_if = "Option::is_none")] - pub value: Option, // будет только при Set + pub value: Option, } /// Count of active sessions @@ -61,7 +78,7 @@ impl Handler for WsHub { type Result = SessionId; fn handle(&mut self, msg: Connect, _ctx: &mut Context) -> Self::Result { - // LEVENT 1 + // LEVENT 1 let id = self.next_id; self.next_id = self.next_id.wrapping_add(1); self.sessions.insert(id, msg.addr); @@ -81,7 +98,7 @@ impl Handler for WsHub { type Result = (); fn handle(&mut self, msg: Disconnect, _ctx: &mut Context) { - // LEVENT 2 + // LEVENT 2 // Delete all subscribes self.subs.retain(|_key, session_ids| { @@ -110,7 +127,8 @@ impl Handler for WsHub { fn handle(&mut self, msg: SubscribeList, _ctx: &mut Context) -> Self::Result { // Collect all keys with my session_id - let list = self.subs + let list = self + .subs .iter() .filter_map(|(key, sessions)| { if sessions.contains(&msg.session_id) { @@ -162,7 +180,9 @@ impl Handler for WsHub { fn handle(&mut self, msg: Unsubscribe, _ctx: &mut Context) { if let Some(set) = self.subs.get_mut(&msg.key) { set.remove(&msg.session_id); - if set.is_empty() { self.subs.remove(&msg.key); } + if set.is_empty() { + self.subs.remove(&msg.key); + } } } } @@ -183,9 +203,6 @@ impl Handler for WsHub { } } - - - #[derive(Message)] #[rtype(result = "HashMap>")] pub struct TestGetSubs; @@ -194,7 +211,8 @@ impl Handler for WsHub { type Result = MessageResult; fn handle(&mut self, _msg: TestGetSubs, _ctx: &mut Context) -> Self::Result { - let s: HashMap> = self.subs + let s: HashMap> = self + .subs .iter() .map(|(key, ids)| (key.clone(), ids.iter().copied().collect())) .collect(); @@ -215,12 +233,12 @@ impl WsHub { } } -use actix::prelude::*; use actix::ActorFutureExt; use actix::fut::ready; +use actix::prelude::*; +use redis::aio::MultiplexedConnection; use std::sync::Arc; use tokio::sync::Mutex; -use redis::aio::MultiplexedConnection; impl Handler for WsHub { type Result = ResponseActFuture; @@ -231,7 +249,8 @@ impl Handler for WsHub { return Box::pin(actix::fut::ready(()).into_actor(self)); } - let recipients: Vec> = targets.into_iter() + let recipients: Vec> = targets + .into_iter() .filter_map(|sid| self.sessions.get(&sid).cloned()) .collect(); @@ -242,9 +261,11 @@ impl Handler for WsHub { Box::pin( async move { let value = if need_get { - let mut conn = redis.lock().await; - match redis::cmd("GET").arg(&event.key).query_async::>(&mut *conn).await + match redis::cmd("GET") + .arg(&event.key) + .query_async::>(&mut *conn) + .await { Ok(v) => v, Err(e) => { @@ -262,7 +283,7 @@ impl Handler for WsHub { let _ = rcpt.do_send(payload.clone()); } } - .into_actor(self) + .into_actor(self), ) } }