feat: get task by id

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/<task_id>`  
+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. If the caller is the task's creator, it can't decrypt the body and will instead return "<encrypted>" as a body.
+
 ### Create a result
 
 Create or update a result of a task. Currently, the body is restricted to 10MB in size.

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<T: DeserializeOwned + 'static>(&self, task_id: &MsgId, blocking: &BlockingOptions) -> Result<Option<TaskRequest<T>>> {
+        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<T: Serialize + 'static>(&self, task: &TaskRequest<T>) -> Result<()> {
         let url = self.beam_proxy_url

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<TasksState>,
+    Path(task_id): Path<MsgId>,
+    mut block: HowLongToBlock,
+    msg: MsgSigned<MsgEmpty>,
+) -> Result<Response, StatusCode> {
+    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<M: Msg> {
     // fn new() -> Self;
     fn from(&self) -> Option<&AppOrProxyId>;

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)

shared/src/lib.rs

@@ -227,6 +227,7 @@ pub trait DecryptableMsg: Msg + Serialize + Sized {
         my_id: &AppOrProxyId,
         my_priv_key: &RsaPrivateKey,
     ) -> Result<Self::Output, SamplyBeamError> {
+
         let Some(Encrypted {
             encrypted,
             encryption_keys,
@@ -235,7 +236,7 @@ pub trait DecryptableMsg: Msg + Serialize + Sized {
             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 +249,15 @@ 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("<encrypted>".to_string()));
+            } else {
+                return Err(SamplyBeamError::SignEncryptError("Decryption error: This client cannot be found in 'to' list".into()));
+            }
+        };
+
         let encrypted_decryption_key = &encryption_keys[to_array_index];
 
         // Cryptographic Operations
@@ -849,4 +855,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::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 <encrypted> body
+        let as_creator = msg_encr
+            .clone()
+            .decrypt(&p1_id, &p1_private)
+            .expect("Creator must receive the task with <encrypted> body");
+        assert_eq!(as_creator.body.body.as_deref(), Some("<encrypted>"));
+
+        // Non-sender or non-reciever is rejected
+        assert!(msg_encr.decrypt(&p3_id, &p3_private).is_err());
+   }
 }

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::<beam_lib::RawString>(&id, &block).await?.map(|t| t.id), Some(id));
+    assert_eq!(client2().get_task::<beam_lib::RawString>(&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::<beam_lib::RawString>(&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<T: Serialize + 'static>(body: T) -> Result<MsgId> {
+    post_task_to(body, vec![APP2.clone()]).await
+}
+
+pub async fn post_task_to<T: Serialize + 'static>(body: T, to: Vec<beam_lib::AddressingId>) -> Result<MsgId> {
     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,