From da15ed3e234388faf538e09ffe8a0653831c659b Mon Sep 17 00:00:00 2001 From: janskiba Date: Fri, 6 Jun 2025 07:34:13 +0000 Subject: [PATCH 01/10] feat: get task by id --- broker/src/serve_tasks.rs | 26 ++++++++++++++++++++++++++ proxy/src/serve_tasks.rs | 1 + 2 files changed, 27 insertions(+) diff --git a/broker/src/serve_tasks.rs b/broker/src/serve_tasks.rs index 74d1668a..27bdc0df 100644 --- a/broker/src/serve_tasks.rs +++ b/broker/src/serve_tasks.rs @@ -38,6 +38,7 @@ pub(crate) fn router() -> Router { let state = TasksState::default(); Router::new() .route("/v1/tasks", get(get_tasks).post(post_task)) + .route("/v1/tasks/{task_id}", get(get_task_by_id)) .route("/v1/tasks/{task_id}/results", get(get_results_for_task)) .route("/v1/tasks/{task_id}/results/{app_id}", put(put_result)) .with_state(state) @@ -210,6 +211,31 @@ async fn get_tasks( }) } +async fn get_task_by_id( + State(state): State, + Path(task_id): Path, + mut block: HowLongToBlock, + msg: MsgSigned, +) -> Result { + if !(block.wait_count.is_none() || block.wait_count == Some(1)) { + return Err(StatusCode::BAD_REQUEST); + } + block.wait_count = Some(1); + let Some(task) = state.task_manager + .wait_for_tasks(&block, |task| task.id() == &task_id && (msg.get_from() == task.get_from() || task.get_to().contains(msg.get_from()))) + .await? + .next() + else { + return Err(StatusCode::NOT_FOUND); + }; + let body = serde_json::to_vec(&*task) + .map_err(|e| { + warn!("Failed to serialize task: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(([(header::CONTENT_TYPE, HeaderValue::from_static("application/json"))], body).into_response()) +} + trait MsgFilterTrait { // fn new() -> Self; fn from(&self) -> Option<&AppOrProxyId>; diff --git a/proxy/src/serve_tasks.rs b/proxy/src/serve_tasks.rs index f2e4b9a5..fb7f36d1 100644 --- a/proxy/src/serve_tasks.rs +++ b/proxy/src/serve_tasks.rs @@ -39,6 +39,7 @@ pub(crate) fn router(client: &SamplyHttpClient, config: &'static Config) -> Rout Router::new() // We need both path variants so the server won't send us into a redirect loop (/tasks, /tasks/, ...) .route("/v1/tasks", get(handler_task).post(handler_task)) + .route("/v1/tasks/{task_id}", get(handler_task)) .route("/v1/tasks/{task_id}/results", get(handler_task)) .route("/v1/tasks/{task_id}/results/{app_id}", put(handler_task)) .with_state(state) From ce92465f4871a870ca4a221a9ba7c450d125c0b4 Mon Sep 17 00:00:00 2001 From: Tobias Kussel Date: Wed, 1 Jul 2026 13:29:40 +0200 Subject: [PATCH 02/10] Add beamlib interface for get-task-by-id --- beam-lib/src/http_util.rs | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/beam-lib/src/http_util.rs b/beam-lib/src/http_util.rs index 78b05dce..2d9f9be4 100644 --- a/beam-lib/src/http_util.rs +++ b/beam-lib/src/http_util.rs @@ -140,6 +140,33 @@ impl BeamClient { } } + /// Retrieve a task by its id. + /// Not existing tasks and non-authorized access return both `None` + /// As only a single task can ever be returned, `blocking.wait_count` must be unset or `1`. + /// The generic parameter `T` represents the expected task body type. + pub async fn get_task(&self, task_id: &MsgId, blocking: &BlockingOptions) -> Result>> { + let url = self.beam_proxy_url + .join(&format!("/v1/tasks/{task_id}?{}", blocking.to_query())) + .expect("The proxy url is valid"); + let response_result = self.client + .get(url) + .send() + .await; + let response = match response_result { + Ok(res) => res, + Err(e) => return if e.is_timeout() { + Ok(None) + } else { + Err(e.into()) + }, + }.handle_invalid_receivers().await?; + match response.status() { + StatusCode::NOT_FOUND | StatusCode::GATEWAY_TIMEOUT => Ok(None), + StatusCode::OK => Ok(Some(response.json().await?)), + status => Err(BeamError::UnexpectedStatus(status)), + } + } + /// Post a beam task with a serializeable body. pub async fn post_task(&self, task: &TaskRequest) -> Result<()> { let url = self.beam_proxy_url From 28c41e159dd3e1d4541f63f257ef967936e87ec0 Mon Sep 17 00:00:00 2001 From: Tobias Kussel Date: Wed, 1 Jul 2026 13:30:25 +0200 Subject: [PATCH 03/10] Add test for get-task-by-id --- tests/src/task_test.rs | 73 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 72 insertions(+), 1 deletion(-) diff --git a/tests/src/task_test.rs b/tests/src/task_test.rs index 957da30b..a6309806 100644 --- a/tests/src/task_test.rs +++ b/tests/src/task_test.rs @@ -74,12 +74,83 @@ async fn test_polling_tasks_yields_more_than_specified_wait_count() -> Result<() Ok(()) } +#[tokio::test] +async fn test_get_task_by_id() -> Result<()> { + let id = post_task(()).await?; + let block = BlockingOptions::from_count(1); + // One is sender, one reciever so both calls should work + assert_eq!(client1().get_task::<()>(&id, &block).await?.map(|t| t.id), Some(id)); + assert_eq!(client2().get_task::<()>(&id, &block).await?.map(|t| t.id), Some(id)); + Ok(()) +} + +#[tokio::test] +async fn test_get_task_by_id_unauthorized() -> Result<()> { + // From APP1 to APP1, App2 unauthorized + let id = post_task_to((), vec![APP1.clone()]).await?; + let block = BlockingOptions::from_time(Duration::from_secs(1)); + assert_eq!(client1().get_task::<()>(&id, &block).await?.map(|t| t.id), Some(id)); + assert!(client2().get_task::<()>(&id, &block).await?.is_none(), "Unauthorized app was able to read task"); + Ok(()) +} + +#[tokio::test] +async fn test_get_task_by_id_not_found() -> Result<()> { + let block = BlockingOptions::from_time(Duration::from_secs(1)); + assert!(client1().get_task::<()>(&MsgId::new(), &block).await?.is_none()); + Ok(()) +} + +#[tokio::test] +async fn test_get_task_by_id_long_poll() -> Result<()> { + let id = MsgId::new(); + let getter = async { + let block = BlockingOptions { wait_time: Some(Duration::from_secs(5)), wait_count: Some(1) }; + let task = client1().get_task::<()>(&id, &block).await?; + assert_eq!(task.map(|t| t.id), Some(id), "Long-poll did not return the task before time out"); + Ok::<_, anyhow::Error>(()) + }; + let poster = async { + // Wait with sending until long-poll is established + tokio::time::sleep(Duration::from_millis(500)).await; + client1().post_task(&TaskRequest { + id, + from: APP1.clone(), + to: vec![APP2.clone()], + body: (), + ttl: "10s".to_string(), + failure_strategy: beam_lib::FailureStrategy::Discard, + metadata: serde_json::Value::Null, + }).await?; + Ok::<_, anyhow::Error>(()) + }; + tokio::try_join!(getter, poster)?; + Ok(()) +} + +#[tokio::test] +async fn test_get_task_by_id_rejects_invalid_wait_count() -> Result<()> { + use reqwest::{Method, StatusCode}; + let id = post_task(()).await?; + let res = client1() + .raw_beam_request(Method::GET, &format!("/v1/tasks/{id}?wait_count=2")) + .send() + .await?; + assert_eq!(res.status(), StatusCode::BAD_REQUEST); + Ok(()) +} + +// Default APP1 -> APP2 pub async fn post_task(body: T) -> Result { + post_task_to(body, vec![APP2.clone()]).await +} + +pub async fn post_task_to(body: T, to: Vec) -> Result { let id = MsgId::new(); client1().post_task(&TaskRequest { id, from: APP1.clone(), - to: vec![APP2.clone()], + to, body, ttl: "10s".to_string(), failure_strategy: beam_lib::FailureStrategy::Discard, From 7f1fe01510352cd25aa4710b4366c8297c219fd2 Mon Sep 17 00:00:00 2001 From: Tobias Kussel Date: Wed, 1 Jul 2026 13:32:58 +0200 Subject: [PATCH 04/10] Document get-task-by-id --- README.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/README.md b/README.md index b4eca17d..233af624 100644 --- a/README.md +++ b/README.md @@ -354,6 +354,29 @@ Date: Mon, 27 Jun 2022 14:05:59 GMT ) ``` +### Retrieve a task by ID + +Retrieve a task by its ID. The caller must be the task's creator or one of its recipients. + +Method: `GET` +URL: `/v1/tasks/` +Parameters: + +- [long polling](#long-polling-api-access) is supported, but since at most one task is ever returned, `wait_count` must be omitted or `1` (any other value returns `400 Bad Request`). + +Returns the single task, cf. [here](#task): + +``` +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "id": ... +} +``` + +If no task with that ID is visible to the caller because a) it does not exist or b) the caller is neither the sender nor a recipient, the broker returns `404 Not Found`. Both conditions leading to the same return value is deliberate to avoid unauthorized message enumeration. + ### Create a result Create or update a result of a task. Currently, the body is restricted to 10MB in size. From d71dafcb7b337dbcef8f05fbb7942711ccd5d57c Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 10:37:38 +0000 Subject: [PATCH 05/10] Return empty body if get-by-id caller is sender --- README.md | 2 +- shared/src/lib.rs | 76 ++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 66 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 233af624..9c3e8551 100644 --- a/README.md +++ b/README.md @@ -375,7 +375,7 @@ Content-Type: application/json } ``` -If no task with that ID is visible to the caller because a) it does not exist or b) the caller is neither the sender nor a recipient, the broker returns `404 Not Found`. Both conditions leading to the same return value is deliberate to avoid unauthorized message enumeration. +If no task with that ID is visible to the caller because a) it does not exist or b) the caller is neither the sender nor a recipient, the broker returns `404 Not Found`. Both conditions leading to the same return value is deliberate to avoid unauthorized message enumeration. If the caller is the task's creator, it can't decrypt the body and will instead return all other fields of the task. ### Create a result diff --git a/shared/src/lib.rs b/shared/src/lib.rs index e16f459d..5b61e0ba 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -227,15 +227,13 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { my_id: &AppOrProxyId, my_priv_key: &RsaPrivateKey, ) -> Result { - let Some(Encrypted { - encrypted, - encryption_keys, - }) = self.get_encryption() else { - // We have something that is not encryptable + + // Message can not be encrypted, e.g. MsgEmpty + if self.get_encryption().is_none() { return Ok(self.convert_self(String::new())); - }; + } - let to_array_index: usize = self + let to_array_index = self .get_to() .iter() .position(|entry| { @@ -248,10 +246,20 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { None => false, }; matched - }) // TODO remove expect! - .ok_or(SamplyBeamError::SignEncryptError( - "Decryption error: This client cannot be found in 'to' list".into(), - ))?; + }); + let Some(to_array_index) = to_array_index else { + if self.get_from().proxy_id() == my_id.proxy_id() { + return Ok(self.convert_self("null".to_string())); + } else { + return Err(SamplyBeamError::SignEncryptError("Decryption error: This client cannot be found in 'to' list".into())); + } + }; + + let Encrypted { + encrypted, + encryption_keys, + } = self.get_encryption().unwrap(); // Checked above (l 233) that it is Some + let encrypted_decryption_key = &encryption_keys[to_array_index]; // Cryptographic Operations @@ -849,4 +857,50 @@ mod tests { assert_eq!(msg_p1_decr, msg_p2_decr); assert_eq!(msg, msg_p1_decr); } + + #[test] + fn decrypt_task_as_creator() { + // Task sender should get an empty body + beam_lib::set_broker_id("broker.samply.de".to_string()); + let p1_id = AppOrProxyId::App(AppId::new("app.proxy1.broker.samply.de").unwrap()); + let p2_id = AppOrProxyId::App(AppId::new("app.proxy2.broker.samply.de").unwrap()); + let p3_id = AppOrProxyId::App(AppId::new("app.proxy3.broker.samply.de").unwrap()); + let msg = MsgTaskRequest { + id: MsgId::new(), + from: p1_id.clone(), + to: vec![p2_id.clone()], + body: "Testbody".into(), + expire: SystemTime::now() + Duration::from_secs(60), + failure_strategy: FailureStrategy::Discard, + results: HashMap::new(), + metadata: "".into(), + }; + + let mut rng = rand::thread_rng(); + let rsa_length: usize = 2048; + let p1_private = RsaPrivateKey::new(&mut rng, rsa_length).unwrap(); + let p2_private = RsaPrivateKey::new(&mut rng, rsa_length).unwrap(); + let p3_private = RsaPrivateKey::new(&mut rng, rsa_length).unwrap(); + let p2_public = RsaPublicKey::from(&p2_private); + + // Encrypted for proxy2 only. + let msg_encr = msg.encrypt(&vec![p2_public]).expect("Could not encrypt message"); + + // Proxy2 can decrypt + let as_recipient = msg_encr + .clone() + .decrypt(&p2_id, &p2_private) + .expect("Recipient must be able to decrypt"); + assert_eq!(as_recipient.body.body.as_deref(), Some("Testbody")); + + // Proxy1 gets empty body + let as_creator = msg_encr + .clone() + .decrypt(&p1_id, &p1_private) + .expect("Creator must receive the task without a body"); + assert_eq!(as_creator.body.body.as_deref(), Some("null")); + + // Non-sender or non-reciever is rejected + assert!(msg_encr.decrypt(&p3_id, &p3_private).is_err()); + } } From 6a7427f9739c593b00de64dc692770b2e5a4c986 Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 10:55:20 +0000 Subject: [PATCH 06/10] Fix new rng interface call --- shared/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 5b61e0ba..0297731a 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -876,7 +876,7 @@ mod tests { metadata: "".into(), }; - let mut rng = rand::thread_rng(); + let mut rng = rand::rng(); let rsa_length: usize = 2048; let p1_private = RsaPrivateKey::new(&mut rng, rsa_length).unwrap(); let p2_private = RsaPrivateKey::new(&mut rng, rsa_length).unwrap(); From f8fc2ca36440fa9009c41c5f8b1018eb48045b92 Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 13:58:33 +0000 Subject: [PATCH 07/10] Revert splitting of get_encryption handling --- shared/src/lib.rs | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 0297731a..62106e58 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -228,10 +228,13 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { my_priv_key: &RsaPrivateKey, ) -> Result { - // Message can not be encrypted, e.g. MsgEmpty - if self.get_encryption().is_none() { + let Some(Encrypted { + encrypted, + encryption_keys, + }) = self.get_encryption() else { + // We have something that is not encryptable return Ok(self.convert_self(String::new())); - } + }; let to_array_index = self .get_to() @@ -255,11 +258,6 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { } }; - let Encrypted { - encrypted, - encryption_keys, - } = self.get_encryption().unwrap(); // Checked above (l 233) that it is Some - let encrypted_decryption_key = &encryption_keys[to_array_index]; // Cryptographic Operations From 49851fa48546c1a5d7f8c810b7c89e3d4f001d4e Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 13:59:19 +0000 Subject: [PATCH 08/10] Choose better body text for tasks fetched by sender --- shared/src/lib.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 62106e58..56f5488c 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -252,7 +252,7 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { }); let Some(to_array_index) = to_array_index else { if self.get_from().proxy_id() == my_id.proxy_id() { - return Ok(self.convert_self("null".to_string())); + return Ok(self.convert_self("".to_string())); } else { return Err(SamplyBeamError::SignEncryptError("Decryption error: This client cannot be found in 'to' list".into())); } @@ -896,7 +896,7 @@ mod tests { .clone() .decrypt(&p1_id, &p1_private) .expect("Creator must receive the task without a body"); - assert_eq!(as_creator.body.body.as_deref(), Some("null")); + assert_eq!(as_creator.body.body.as_deref(), Some("")); // Non-sender or non-reciever is rejected assert!(msg_encr.decrypt(&p3_id, &p3_private).is_err()); From 5e07b1a1c81a2e05c665df75e79019c2cd8a8594 Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 14:43:49 +0000 Subject: [PATCH 09/10] Use raw string in task-by-id tests --- tests/src/task_test.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/src/task_test.rs b/tests/src/task_test.rs index a6309806..316cc7e7 100644 --- a/tests/src/task_test.rs +++ b/tests/src/task_test.rs @@ -79,8 +79,8 @@ async fn test_get_task_by_id() -> Result<()> { let id = post_task(()).await?; let block = BlockingOptions::from_count(1); // One is sender, one reciever so both calls should work - assert_eq!(client1().get_task::<()>(&id, &block).await?.map(|t| t.id), Some(id)); - assert_eq!(client2().get_task::<()>(&id, &block).await?.map(|t| t.id), Some(id)); + assert_eq!(client1().get_task::(&id, &block).await?.map(|t| t.id), Some(id)); + assert_eq!(client2().get_task::(&id, &block).await?.map(|t| t.id), Some(id)); Ok(()) } @@ -106,7 +106,7 @@ async fn test_get_task_by_id_long_poll() -> Result<()> { let id = MsgId::new(); let getter = async { let block = BlockingOptions { wait_time: Some(Duration::from_secs(5)), wait_count: Some(1) }; - let task = client1().get_task::<()>(&id, &block).await?; + let task = client1().get_task::(&id, &block).await?; assert_eq!(task.map(|t| t.id), Some(id), "Long-poll did not return the task before time out"); Ok::<_, anyhow::Error>(()) }; From 9eddf7020b6c14c54cf6abc1f42c19d318cbc933 Mon Sep 17 00:00:00 2001 From: tobiaskussel Date: Thu, 2 Jul 2026 14:46:28 +0000 Subject: [PATCH 10/10] Neaten documentation of task-by-id --- README.md | 2 +- shared/src/lib.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 9c3e8551..d4b23edd 100644 --- a/README.md +++ b/README.md @@ -375,7 +375,7 @@ Content-Type: application/json } ``` -If no task with that ID is visible to the caller because a) it does not exist or b) the caller is neither the sender nor a recipient, the broker returns `404 Not Found`. Both conditions leading to the same return value is deliberate to avoid unauthorized message enumeration. If the caller is the task's creator, it can't decrypt the body and will instead return all other fields of the task. +If no task with that ID is visible to the caller because a) it does not exist or b) the caller is neither the sender nor a recipient, the broker returns `404 Not Found`. Both conditions leading to the same return value is deliberate to avoid unauthorized message enumeration. If the caller is the task's creator, it can't decrypt the body and will instead return "" as a body. ### Create a result diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 56f5488c..8e1a1584 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -891,11 +891,11 @@ mod tests { .expect("Recipient must be able to decrypt"); assert_eq!(as_recipient.body.body.as_deref(), Some("Testbody")); - // Proxy1 gets empty body + // Proxy1 gets body let as_creator = msg_encr .clone() .decrypt(&p1_id, &p1_private) - .expect("Creator must receive the task without a body"); + .expect("Creator must receive the task with body"); assert_eq!(as_creator.body.body.as_deref(), Some("")); // Non-sender or non-reciever is rejected