Release 0.11.0

.github/workflows/rust.yml

@@ -13,11 +13,10 @@ jobs:
       # Docker Hub name will be {image-prefix}{component}
       image-prefix: "samply/"
       components: '[ "beam-broker", "beam-proxy" ]'
-      #architectures: '[ "amd64" ]'
+      architectures: '[ "amd64", "arm64" ]'
       #profile: debug
       test-via-script: true
       features: '[ "", "sockets" ]'
-      push-to: ${{ (github.ref_protected == true || github.event_name == 'workflow_dispatch') && 'dockerhub' || 'ghcr' }}
     secrets:
       DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
       DOCKERHUB_TOKEN: ${{ secrets.DOCKERHUB_TOKEN }}

.gitignore

@@ -3,4 +3,6 @@
 artifacts/
 dev/pki/*
 !dev/pki/pki
-**/mitmproxy/
+**/mitmproxy/
+dev/logs/
+.zed/

CHANGELOG.md

@@ -1,3 +1,19 @@
+
+# Samply.Beam 0.11.0 - 2026-07-06
+
+## Minor changes
+
+* Improved resilience by retrying timed-out broker requests.
+* Certificates are now fetched in parallel, reducing startup time.
+* Broker and proxy logs can now be written to disk.
+* Added an endpoint to allow senders and recipients to retrieve a task with a known task-id.
+
+## Bugfixes
+
+* Fixed an overly broad signer check that matched bare id suffixes (e.g. `ulm.broker` for `neu-ulm.broker`). The check now requires a `.`-delimited label boundary.
+* Fixed socket tasks lingering after one side has already given up on the connection.
+* Fixed a rare certificate getter initialization issue causing it to be initialized more than once leading to a crash.
+
 # Samply.Beam 0.10.0 - 2025-05-26
 
 ## Breaking changes

Cargo.toml

@@ -1,10 +1,17 @@
 [workspace]
 members = ["proxy", "broker", "shared", "tests", "beam-lib"]
 resolver = "2"
-package.version = "0.10.0"
+package.version = "0.11.0"
 
 [workspace.dependencies]
 beam-lib = { path = "./beam-lib", features = [ "strict-ids" ] }
+rsa = "0.10.0-rc.18"
+chacha20poly1305 = "0.11"
+rand = "0.10"
+# Command Line Interface
+clap = { version = "4", features = ["env", "derive"] }
+reqwest = { version = "0.13", default-features = false }
+futures = { version = "0.3" }
 
 [profile.release]
 #opt-level = "z"     # Optimize for size.

README.md

@@ -4,9 +4,9 @@
 
 Samply.Beam is a distributed task broker designed for efficient communication across strict network environments. It provides most commonly used communication patterns across strict network boundaries, end-to-end encryption and signatures, as well as certificate management and validation on top of an easy to use REST API. In addition to task/response semantics, Samply.Beam supports high-performance applications with encrypted low-level direct socket connections.
 
-## Latest version: Samply.Beam 0.10.0 – 2025-05-26
+## Latest version: Samply.Beam 0.11.0 – 2026-07-06
 
-This new major version includes better error messages for expired certificates and a small api change. Please check the [Changelog](CHANGELOG.md) for details.
+This new version improves broker request retries, speeds up certificate fetching, adds file logging and a get-task-by-id endpoint, and fixes socket cleanup. Please check the [Changelog](CHANGELOG.md) for details.
 
 Find info on all previous versions in the [Changelog](CHANGELOG.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/Cargo.toml

@@ -9,7 +9,7 @@ license = "Apache-2.0"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 uuid = { version = "1", features = ["v4", "serde"] }
-reqwest = { version = "0.12", features = ["json"], default-features = false, optional = true }
+reqwest = { workspace = true, features = ["json"], default-features = false, optional = true }
 thiserror = { version = "2.0", optional = true }
 
 [features]

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

beam-lib/src/ids.rs

@@ -68,7 +68,7 @@ impl AppOrProxyId {
     }
 
     pub fn can_be_signed_by(&self, other: &impl AsRef<str>) -> bool {
-        self.as_ref().ends_with(other.as_ref())
+        signer_matches(self.as_ref(), other.as_ref())
     }
 
     pub fn hide_broker(&self) -> &str {
@@ -150,7 +150,7 @@ macro_rules! impl_id {
 
             #[cfg(feature = "strict-ids")]
             pub fn can_be_signed_by(&self, other: &impl AsRef<str>) -> bool {
-                self.as_ref().ends_with(other.as_ref())
+                signer_matches(self.as_ref(), other.as_ref())
             }
         }
 
@@ -271,6 +271,18 @@ impl Display for BeamIdError {
     }
 }
 
+/// Returns true if `signer` is allowed to sign on behalf of `id`, i.e. `id`
+/// equals `signer` or `signer` is a suffix of `id` at a `.`-delimited label
+/// boundary. A bare `str::ends_with` is insufficient here: it would let a proxy
+/// `ulm.broker` sign for the unrelated `neu-ulm.broker`.
+#[cfg(feature = "strict-ids")]
+fn signer_matches(id: &str, signer: &str) -> bool {
+    id == signer
+        || id
+            .strip_suffix(signer)
+            .is_some_and(|prefix| prefix.ends_with('.'))
+}
+
 #[cfg(feature = "strict-ids")]
 fn check_valid_id_part(id: &str) -> Result<(), BeamIdError> {
     for char in id.chars() {
@@ -335,6 +347,25 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_can_be_signed_by_label_boundary() {
+        set_broker_id("broker.samply.de".to_string());
+        let victim_proxy = ProxyId::new("neu-ulm.broker.samply.de").unwrap();
+        let victim_app: AppOrProxyId =
+            AppId::new("app.neu-ulm.broker.samply.de").unwrap().into();
+        let attacker = ProxyId::new("ulm.broker.samply.de").unwrap();
+
+        // Legitimate: a proxy signs for itself and its own apps.
+        assert!(victim_proxy.can_be_signed_by(&victim_proxy));
+        assert!(victim_app.can_be_signed_by(&victim_proxy));
+
+        // Attack: `ulm.broker.samply.de` is a bare-suffix of the unrelated
+        // `neu-ulm.broker.samply.de`, but the preceding char is `-`, not a label
+        // boundary -> must be rejected.
+        assert!(!victim_proxy.can_be_signed_by(&attacker));
+        assert!(!victim_app.can_be_signed_by(&attacker));
+    }
+
     #[test]
     fn test_app_or_proxy_id() {
         let app_id_str = "app.proxy1.broker.samply.de";

broker/Cargo.toml

@@ -8,13 +8,14 @@ documentation = "https://github.com/samply/beam"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-shared = { path = "../shared", features = ["config-for-central"] }
+shared = { path = "../shared" }
 beam-lib = { workspace = true }
+clap.workspace = true
 
 tokio = { version = "1", features = ["full"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-axum = { version = "0.8", features = [ "query" ] }
+axum = { version = "0.8", features = ["query", "macros"] }
 #axum-macros = "0.3.7"
 dashmap =  "6.0"
 
@@ -26,11 +27,11 @@ tracing = "0.1"
 
 # Server-sent Events (SSE) support
 async-stream = "0.3"
-futures-core = { version = "0.3", default-features = false }
+futures.workspace = true
 once_cell = "1"
 # Socket dependencies
 bytes = { version = "1", optional = true }
-axum-extra = { version = "0.10", features = ["typed-header"] }
+axum-extra = { version = "0.12", features = ["typed-header"] }
 hyper = { version = "1", default-features = false, optional = true}
 hyper-util = { version = "0.1", default-features = false, features = ["tokio"], optional = true}
 

shared/src/config_broker.rs → broker/src/config.rs

@@ -5,7 +5,7 @@ use crate::{
 };
 use axum::http::Uri;
 use clap::Parser;
-use reqwest::Url;
+use shared::{logger::LogOptions, openssl::x509::X509, reqwest::{self, Url}};
 use std::str::FromStr;
 use tracing::info;
 
@@ -14,9 +14,9 @@ use tracing::info;
     name("🌈 Samply.Beam.Broker"),
     version,
     arg_required_else_help(true),
-    after_help(crate::config_shared::CLAP_FOOTER)
+    after_help(crate::CLAP_FOOTER)
 )]
-struct CliArgs {
+pub struct CliArgs {
     /// Local bind address
     #[clap(long, env, value_parser, default_value_t = SocketAddr::from_str("0.0.0.0:8080").unwrap())]
     bind_addr: SocketAddr,
@@ -53,23 +53,23 @@ struct CliArgs {
     #[clap(long, env, value_parser)]
     monitoring_api_key: Option<String>,
 
-    /// (included for technical reasons)
-    #[clap(long, hide(true))]
-    test_threads: Option<String>,
+    #[clap(flatten)]
+    pub log_options: LogOptions,
 }
 
+#[derive(Debug)]
 pub struct Config {
     pub bind_addr: SocketAddr,
     pub pki_address: Url,
     pub pki_realm: String,
     pub pki_token: String,
-    pub tls_ca_certificates_dir: Option<PathBuf>,
+    pub tls_ca_certificates: Vec<reqwest::Certificate>,
     pub monitoring_api_key: Option<String>,
+    pub rootcert: X509,
 }
 
-impl crate::config::Config for Config {
-    fn load() -> Result<Self, SamplyBeamError> {
-        let cli_args = CliArgs::parse();
+impl Config {
+    pub fn load(cli_args: CliArgs) -> Result<Self, SamplyBeamError> {
         beam_lib::set_broker_id(cli_args.broker_url.host().unwrap().to_string());
         let pki_token = read_to_string(&cli_args.pki_apikey_file)
             .map_err(|e| {
@@ -88,8 +88,11 @@ impl crate::config::Config for Config {
             pki_address: cli_args.pki_address,
             pki_realm: cli_args.pki_realm,
             pki_token,
-            tls_ca_certificates_dir: cli_args.tls_ca_certificates_dir,
+            tls_ca_certificates: shared::crypto::load_certificates_from_dir(cli_args.tls_ca_certificates_dir).map_err(|e| {
+                SamplyBeamError::ConfigurationFailed(format!("Unable to read from TLS CA directory: {e:#}"))
+            })?,
             monitoring_api_key: cli_args.monitoring_api_key,
+            rootcert: shared::crypto::load_certificates_from_file(cli_args.rootcert_file)?,
         };
         Ok(config)
     }