diff --git a/README.md b/README.md index e8aa9f6763..2132c06479 100644 --- a/README.md +++ b/README.md @@ -12,13 +12,34 @@ The service is exposed as REST and WebSocket API. - service posts a process status ## Key -Key is a string that consists of one or multiple segments separated by some separator. -Example: foo/bar/baz. -It is possible to use wildcard keys to list or subscribe to values with this prefix. +Key is a string that consists of one or multiple segments separated by ‘/’. Example: foo/bar/baz. +Key may not end with ‘/’ +Segment may not contain special characters (‘*’, ‘?’, ‘[’, ‘]’,‘\’,‘\x00..\xF1’,‘\x7F’,‘"’,‘'’) +Segment may not be empty +Key segment may be private (prefixed with ‘$’) + + Query + +May not contain special characters (‘*’, ‘?’, ‘[’, ‘]’,‘\’,‘\x00..\xF1’,‘\x7F’,‘"’,‘'’) +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] -Key may contain a special section (guard) $that separates public and private data. “Private” data is available when querying or subscribing by exact key. -Example foo/bar/$/private, this value can be queried by foo/bar/$/private or foo/bar/$/but not by foo/bar/ ## Data “Data” is an arbitrary JSON document. diff --git a/scripts/TEST.html b/scripts/TEST.html index 8cd50f79b5..a525eca7f1 100644 --- a/scripts/TEST.html +++ b/scripts/TEST.html @@ -46,6 +46,7 @@ border-radius: 6px; min-height: 100px; white-space: pre-wrap; + overflow-wrap: anywhere; } diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index 3e1a69e1f5..050093fcf7 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -6,6 +6,129 @@ source ./pulse_lib.sh TOKEN=$(./token.sh claims.json) ZP="00000000-0000-0000-0000-000000000001/TESTS" + +echo "--------- if-match ----------" + + put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS/2" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS/3$" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/4" "Value_1" "HULY-TTL: 2" + get "00000000-0000-0000-0000-000000000001/TESTS" + get "00000000-0000-0000-0000-000000000001/TESTS/" + 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 + +echo "--------- Deprecated symbols ----------" + + put "00000000-0000-0000-0000-000000000001/'TESTS" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TES?TS" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS*" "Value_1" "HULY-TTL: 2" + put "00000000-0000-0000-0000-000000000001/TESTS/" "Value_1" "HULY-TTL: 2" + echo "--------- if-match ----------" delete ${ZP} @@ -42,17 +165,6 @@ echo "================> UPDATE PUT If-Match" - - - - - - - - - - - put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 3" echo "sleep 1 sec" sleep 1 diff --git a/scripts/pulse_lib.sh b/scripts/pulse_lib.sh index c79edec450..022d8cb87c 100755 --- a/scripts/pulse_lib.sh +++ b/scripts/pulse_lib.sh @@ -32,6 +32,9 @@ api() { *) echo -en "${GRAY}${status}${N}" ;; esac if [ -n "$etag" ]; then echo -n -e " ${F}${etag}${N}" ; fi + + body=$(echo "$body" | sed 's/{/\\n{/g') + if [ -n "$body" ]; then echo -e "\n ${GRAY}[${body}]${N}" ; else echo -e " ${L}(no body)${N}" ; fi rm -f "$tmpfile" } diff --git a/src/handlers_http.rs b/src/handlers_http.rs index 7786627bc8..bba2cc8ddc 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -8,8 +8,6 @@ use tracing::{error, trace}; use uuid::Uuid; use crate::ws_owner; -type ObjectPath = web::Path<(String, String)>; - use crate::redis::{ Ttl, SaveMode, RedisArray, @@ -46,38 +44,35 @@ pub fn map_handler_error(err: impl std::fmt::Display) -> Error { /// list - -// #[derive(Deserialize)] pub async fn list( req: HttpRequest, - path: web::Path, - query: web::Query>, + path: web::Path<(String, Option)>, redis: web::Data>>, ) -> Result { ws_owner::workspace_owner(&req)?; // Check workspace - let workspace = path.into_inner(); - let prefix = query.get("prefix").map(|s| s.as_str()); + let (workspace, key) = path.into_inner(); - trace!(workspace, prefix, "list request"); + // trace!(workspace, prefix, "list request"); async move || -> anyhow::Result { let mut conn = redis.lock().await; - let entries = redis_list(&mut *conn, &workspace, prefix).await?; + let entries = redis_list(&mut *conn, &workspace, key.as_deref()).await?; Ok(HttpResponse::Ok().json(entries)) }().await.map_err(map_handler_error) } -/// get / (test) + +/// get pub async fn get( req: HttpRequest, - path: ObjectPath, + path: web::Path<(String, String)>, redis: web::Data>>, ) -> Result { @@ -85,7 +80,7 @@ pub async fn get( let (workspace, key) = path.into_inner(); - trace!(workspace, key, "get request"); + // trace!(workspace, key, "get request"); async move || -> anyhow::Result { @@ -104,10 +99,9 @@ pub async fn get( /// put - pub async fn put( req: HttpRequest, - path: ObjectPath, + path: web::Path<(String, String)>, body: web::Bytes, redis: web::Data>>, ) -> Result { @@ -118,6 +112,8 @@ pub async fn put( async move || -> anyhow::Result { + if !req.query_string().is_empty() { return Err(anyhow!("Query parameters are not allowed")); } + let mut conn = redis.lock().await; // TTL logic @@ -151,11 +147,10 @@ pub async fn put( -// delete - +/// delete pub async fn delete( req: HttpRequest, - path: ObjectPath, + path: web::Path<(String, String)>, redis: web::Data>>, ) -> Result { diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 216fcf686a..af20f02411 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -1,3 +1,27 @@ +use redis::aio::MultiplexedConnection; +use std::sync::Arc; +use tokio::sync::Mutex; + +use actix::fut; +use actix::ActorFutureExt; + +use crate::redis::{ + Ttl, SaveMode, + RedisArray, + redis_save, + redis_read, + redis_delete, + redis_list, + error +}; + +use std::future::Future; +use serde_json::{Value, Map, json}; + + +type JsonMap = Map; + +// ================== use actix::{Actor, StreamHandler, AsyncContext, ActorContext}; use actix_web::{web, HttpRequest, HttpResponse, Error}; use actix_web_actors::ws; @@ -10,22 +34,59 @@ use std::collections::HashSet; #[serde(rename_all = "lowercase", tag = "type")] pub enum WsCommand { Put { + #[serde(default)] + correlation: Option, key: String, data: String, - #[serde(default)] - correlation: Option, + #[serde(rename = "expiresAt")] + #[serde(default)] expires_at: Option, + + #[serde(rename = "TTL")] + #[serde(default)] + ttl: Option, + + #[serde(rename = "ifMatch")] + #[serde(default)] + if_match: Option, + + #[serde(rename = "ifNoneMatch")] + #[serde(default)] + if_none_match: Option, }, - Delete { - key: String, + + Get { #[serde(default)] correlation: Option, - }, - Sub { key: String, }, + + List { + #[serde(default)] + correlation: Option, + key: Option, + }, + + Delete { + #[serde(default)] + correlation: Option, + key: String, + + #[serde(rename = "ifMatch")] + #[serde(default)] + if_match: Option, + }, + + Sub { + #[serde(default)] + correlation: Option, + key: String, + }, + Unsub { + #[serde(default)] + correlation: Option, key: String, }, } @@ -35,8 +96,10 @@ pub enum WsCommand { pub struct WsSession { pub workspace: String, pub subscriptions: HashSet, // новые поля + pub redis: Arc>, // вот он, тот же тип что и в HTTP API } + /// Actor External trait: must be in separate impl block impl Actor for WsSession { type Context = ws::WebsocketContext; @@ -72,48 +135,224 @@ impl StreamHandler> for WsSession { /// All logic in one impl impl WsSession { + fn wait_and_send( + &mut self, + ctx: &mut ws::WebsocketContext, + fut: F, + mut base: JsonMap, + ) + where + F: std::future::Future> + 'static, + { + ctx.wait( + fut::wrap_future(fut).map(move |res, _actor: &mut Self, ctx| { + match res { + Ok(extra) => { + base.extend(extra); + } + Err(err) => { + base.insert("type".into(), json!("error")); + base.insert("message".into(), json!(err)); + } + } + ctx.text(Value::Object(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, correlation } => { - println!("PUT {} = {} (expires_at: {:?})", key, data, expires_at); - ctx.text(format!("OK PUT {}{}", key, Self::correlation_suffix(&correlation))); - // Здесь — сохранить в Redis - } - WsCommand::Delete { key, correlation } => { + + WsCommand::Put { key, data, expires_at, ttl, if_match, if_none_match, correlation } => { + + println!("PUT {} = {} (expires_at: {:?}) (ttl: {:?}) ws={:?}", key, data, expires_at, ttl, self.workspace); + + let redis = self.redis.clone(); + let workspace = self.workspace.clone(); + + let mut base = JsonMap::new(); + base.insert("action".into(), json!("put")); + base.insert("workspace".into(), json!(&self.workspace)); + 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 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 + }; + + // 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; + + redis_save(&mut *conn, &workspace, &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) + + }; + + self.wait_and_send(ctx, fut, base); + } + + + WsCommand::Delete { key, correlation, if_match } => { println!("DELETE {}", key); - ctx.text(format!("OK DELETE {}{}", key, Self::correlation_suffix(&correlation))); - // Здесь — удалить из Redis + + let redis = self.redis.clone(); + let workspace = self.workspace.clone(); + + let mut base = JsonMap::new(); + base.insert("action".into(), json!("delete")); + base.insert("workspace".into(), json!(&self.workspace)); + 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 fut = async move { + + let mut conn = redis.lock().await; + + let deleted = redis_delete(&mut *conn, &workspace, &key) + .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()) + } + + }; + + self.wait_and_send(ctx, fut, base); } - WsCommand::Sub { key } => { - println!("SUB {}", key); + + WsCommand::Get { key, correlation } => { + println!("GET {}{:?}", key, correlation); + + let redis = self.redis.clone(); + let workspace = self.workspace.clone(); + + let mut base = JsonMap::new(); + base.insert("action".into(), json!("get")); + base.insert("workspace".into(), json!(&self.workspace)); + base.insert("key".into(), json!(&key)); + if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } + + let fut = async move { + + let mut conn = redis.lock().await; + + let data_opt = redis_read(&mut *conn, &workspace, &key) + .await + .map_err(|e| e.to_string())?; + + 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); + } + + WsCommand::List { key, correlation } => { + println!("LIST {:?}{:?}", key, correlation); + + let redis = self.redis.clone(); + let workspace = self.workspace.clone(); + + let mut base = JsonMap::new(); + base.insert("action".into(), json!("get")); + base.insert("workspace".into(), json!(&self.workspace)); + if let Some(x) = &key { base.insert("key".into(), json!(x)); } + if let Some(x) = &correlation { base.insert("correlation".into(), json!(x)); } + + let fut = async move { + + let mut conn = redis.lock().await; + + let data = redis_list(&mut *conn, &workspace, key.as_deref()) + .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); + } + + WsCommand::Sub { key, correlation } => { + println!("SUB {}{:?}", key, correlation); ctx.text(format!("OK SUB {}", key)); // Здесь — подписка (в будущем pub/sub) } - WsCommand::Unsub { key } => { - println!("UNSUB {}", key); + + + WsCommand::Unsub { key, correlation } => { + println!("UNSUB {}{:?}", key, correlation); ctx.text(format!("OK UNSUB {}", key)); // Здесь — отписка } + } } - fn correlation_suffix(corr: &Option) -> String { - match corr { - Some(c) => format!(" [correlation: {}]", c), - None => "".to_string(), - } - // - // corr.as_ref() - // .map(|c| format!(" [correlation: {}]", c)) - // .unwrap_or_default() - // - } - } -pub async fn handler(req: HttpRequest, stream: web::Payload, path: web::Path) -> Result { + +pub async fn handler( + req: HttpRequest, + stream: web::Payload, + path: web::Path, + redis: web::Data>>, +) -> Result { let workspace = path.into_inner(); - let session = WsSession { workspace, subscriptions: HashSet::new() }; + let session = WsSession { + workspace, + subscriptions: HashSet::new(), + redis: redis.get_ref().clone(), + }; ws::start(session, &req, stream) } diff --git a/src/main.rs b/src/main.rs index 99d2b56c4a..3717dccb27 100644 --- a/src/main.rs +++ b/src/main.rs @@ -105,10 +105,11 @@ async fn main() -> anyhow::Result<()> { .service( web::scope("/api") .wrap(middleware::from_fn(interceptor)) - .route("/{workspace}", web::get().to(handlers_http::list)) - .route("/{workspace}/{key:.*}",web::get().to(handlers_http::get)) - .route("/{workspace}/{key:.*}",web::put().to(handlers_http::put)) - .route("/{workspace}/{key:.*}",web::delete().to(handlers_http::delete)) + .route("/{workspace}/", web::get().to(handlers_http::list)) + .route("/{workspace}/{key:.+/}", web::get().to(handlers_http::list)) + .route("/{workspace}/{key:.+}", web::get().to(handlers_http::get)) + .route("/{workspace}/{key:.+}", web::put().to(handlers_http::put)) + .route("/{workspace}/{key:.+}", web::delete().to(handlers_http::delete)) ) .route("/status", web::get().to(async || "ok")) .route("/ws/{workspace}", web::get().to(handlers_ws::handler)) // WebSocket diff --git a/src/redis.rs b/src/redis.rs index 29590a5b84..9c8edfcbd4 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -2,6 +2,7 @@ use crate::config::{CONFIG, RedisMode}; use std::time::{SystemTime, UNIX_EPOCH}; +#[derive(serde::Serialize)] pub enum Ttl { Sec(usize), // EX At(u64), // EXAT (timestamp in seconds) @@ -32,51 +33,76 @@ pub struct RedisArray { pub etag: String, // md5 hash (data) } -fn error(code: u16, msg: impl Into) -> redis::RedisResult { +/// 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 ))) } + +/// Check for redis-deprecated symbols +pub fn redis_deprecate_symbols(s: &str) -> redis::RedisResult<()> { + if s.chars().any(|c| matches!( c, + '*' | '?' | '[' | ']' | '\\' | + '\0'..='\x1F' | '\x7F' | + '"' | '\'' // | ' ' + )) { + error(412, "Deprecated symbols in workspace or key") + } else { + Ok(()) + } +} + + /// redis_list(&connection,workspace,prefix) pub async fn redis_list( conn: &mut MultiplexedConnection, workspace: &str, - prefix: Option<&str>, + key: Option<&str>, ) -> redis::RedisResult> { + + let pattern = if let Some(k) = key { + if !k.ends_with('/') { return error(412, "Key must end with slash"); } + Some(format!("{k}*")) + } else { + None + }; + + redis_deprecate_symbols(&workspace)?; + if let Some(k) = key { redis_deprecate_symbols(k)?; } + let mut cursor = 0; let mut results = Vec::new(); - let pattern = prefix.map(|p| format!("{}*", p)); loop { let mut cmd = redis::cmd("HSCAN"); cmd.arg(workspace).arg(cursor); if let Some(ref p) = pattern { - cmd.arg("MATCH").arg(p); - } + cmd.arg("MATCH").arg(p); + } + // cmd.arg("COUNT").arg(100); let (next_cursor, items): (u64, Vec<(String, String)>) = cmd.query_async(conn).await?; - for (key, value) in items { + for (k, v) in items { + + // Check for $-security path + if let Some(prefix) = key { + if k[prefix.len()..].contains('$') { continue; } + } + // TTL - let ttl_vec: Vec = redis::cmd("HTTL") - .arg(workspace) - .arg("FIELDS") - .arg(1) - .arg(&key) - .query_async(conn) - .await?; - + let ttl_vec: Vec = redis::cmd("HTTL").arg(workspace).arg("FIELDS").arg(1).arg(&k).query_async(conn).await?; let ttl = ttl_vec.get(0).copied().unwrap_or(-3); - if ttl >= 0 { results.push(RedisArray { workspace: workspace.to_string(), - key, - data: value.clone(), + key: k, + data: v.clone(), expires_at: ttl as u64, - etag: hex::encode(md5::compute(&value).0), + etag: hex::encode(md5::compute(&v).0), }); } } @@ -97,6 +123,10 @@ pub async fn redis_read( key: &str, ) -> redis::RedisResult> { + redis_deprecate_symbols(&workspace)?; + redis_deprecate_symbols(&key)?; + if key.ends_with('/') { return error(412, "Key must not end with a slash"); } + let data: Option = redis::cmd("HGET").arg(workspace).arg(key).query_async(conn).await?; let Some(data) = data else { return Ok(None); }; @@ -137,6 +167,10 @@ pub async fn redis_save( mode: Option, ) -> RedisResult<()> { + redis_deprecate_symbols(&workspace)?; + redis_deprecate_symbols(&key)?; + 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, @@ -201,6 +235,10 @@ pub async fn redis_delete( key: &str, ) -> redis::RedisResult { + redis_deprecate_symbols(&workspace)?; + redis_deprecate_symbols(&key)?; + if key.ends_with('/') { return error(412, "Key must not end with a slash"); } + let deleted: i32 = redis::cmd("HDEL") .arg(workspace) .arg(key) @@ -268,3 +306,5 @@ pub async fn redis_connect() -> anyhow::Result { Ok(conn) } + +