diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 7bd5493c..23c5e662 100644 --- a/.github/workflows/rust.yml +++ b/.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 }} diff --git a/.gitignore b/.gitignore index 980c54b3..2483001e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,6 @@ artifacts/ dev/pki/* !dev/pki/pki -**/mitmproxy/ \ No newline at end of file +**/mitmproxy/ +dev/logs/ +.zed/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 900c3dbf..d504cb14 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Cargo.toml b/Cargo.toml index 4d3e043d..533b079b 100644 --- a/Cargo.toml +++ b/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. diff --git a/README.md b/README.md index 1b1212a4..b0be12fd 100644 --- a/README.md +++ b/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/` +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 "" as a body. + ### Create a result Create or update a result of a task. Currently, the body is restricted to 10MB in size. diff --git a/beam-lib/Cargo.toml b/beam-lib/Cargo.toml index 922319dd..75b44cfd 100644 --- a/beam-lib/Cargo.toml +++ b/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] 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 diff --git a/beam-lib/src/ids.rs b/beam-lib/src/ids.rs index 60614691..3c883b2e 100644 --- a/beam-lib/src/ids.rs +++ b/beam-lib/src/ids.rs @@ -68,7 +68,7 @@ impl AppOrProxyId { } pub fn can_be_signed_by(&self, other: &impl AsRef) -> 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) -> 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"; diff --git a/broker/Cargo.toml b/broker/Cargo.toml index cf813349..0c94f774 100644 --- a/broker/Cargo.toml +++ b/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} diff --git a/shared/src/config_broker.rs b/broker/src/config.rs similarity index 79% rename from shared/src/config_broker.rs rename to broker/src/config.rs index 727aec81..e5a7156e 100644 --- a/shared/src/config_broker.rs +++ b/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, - /// (included for technical reasons) - #[clap(long, hide(true))] - test_threads: Option, + #[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, + pub tls_ca_certificates: Vec, pub monitoring_api_key: Option, + pub rootcert: X509, } -impl crate::config::Config for Config { - fn load() -> Result { - let cli_args = CliArgs::parse(); +impl Config { + pub fn load(cli_args: CliArgs) -> Result { 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) } diff --git a/broker/src/crypto.rs b/broker/src/crypto.rs index 1f12c371..aeaef187 100644 --- a/broker/src/crypto.rs +++ b/broker/src/crypto.rs @@ -3,16 +3,18 @@ use std::{future::Future, mem::discriminant, sync::Arc}; use axum::http::{header, method, uri::Scheme, Method, Request, StatusCode, Uri}; use serde::{Deserialize, Serialize}; use shared::{ - async_trait, config, crypto::{parse_crl, CertificateCache, CertificateCacheUpdate, GetCerts}, errors::SamplyBeamError, http_client::{self, SamplyHttpClient}, openssl::x509::X509Crl, reqwest::{self, Url} + async_trait, crypto::{parse_crl, CertificateCache, CertificateCacheUpdate, GetCerts}, errors::SamplyBeamError, http_client::{self, SamplyHttpClient}, openssl::x509::X509Crl, reqwest::{self, Url} }; use std::time::Duration; use tokio::{sync::RwLock, time::timeout}; use tracing::{debug, error, warn, info}; -use crate::serve_health::{Health, VaultStatus}; +use crate::{config::Config, serve_health::{Health, VaultStatus}}; pub struct GetCertsFromPki { pki_realm: String, + vault_token: String, + vault_address: Url, hyper_client: SamplyHttpClient, health: Arc>, } @@ -36,30 +38,18 @@ struct PkiListResponse { impl GetCertsFromPki { pub(crate) fn new( health: Arc>, + config: &Config, ) -> Result { - let mut certs: Vec = Vec::new(); - if let Some(dir) = &config::CONFIG_CENTRAL.tls_ca_certificates_dir { - for file in std::fs::read_dir(dir).map_err(|e| { - SamplyBeamError::ConfigurationFailed(format!( - "Unable to read CA certificates: {}", - e - )) - })? { - if let Ok(file) = file { - certs.push(file.path().to_str().unwrap().into()); - } - } - debug!("Loaded local certificates: {}", certs.join(" ")); - } - let hyper_client = http_client::build( - &config::CONFIG_SHARED.tls_ca_certificates, + let hyper_client = http_client::builder( + &config.tls_ca_certificates, Some(Duration::from_secs(30)), Some(Duration::from_secs(20)), - )?; - let pki_realm = config::CONFIG_CENTRAL.pki_realm.clone(); + ).build()?; Ok(Self { - pki_realm, + pki_realm: config.pki_realm.clone(), + vault_address: config.pki_address.clone(), + vault_token: config.pki_token.clone(), hyper_client, health, }) @@ -86,7 +76,7 @@ impl GetCertsFromPki { } async fn check_vault_health_helper(&self) -> Result<(), SamplyBeamError> { - let url = pki_url_builder("sys/health"); + let url = self.vault_address.join(&format!("/v1/sys/health")).unwrap(); debug!("Checking Vault's health at URL {url}"); let health = self.hyper_client.get(url).send().await; let Ok(resp) = health else { @@ -117,7 +107,7 @@ impl GetCertsFromPki { api_path: &str, max_tries: Option, ) -> Result { - let uri = pki_url_builder(api_path); + let uri = self.vault_address.join(&format!("/v1/{api_path}")).unwrap(); debug!("Samply.PKI: Vault request to {uri}"); let max_tries = max_tries.unwrap_or(u32::MAX); for tries in 0..max_tries { @@ -126,7 +116,7 @@ impl GetCertsFromPki { } let resp = self.hyper_client .request(method.clone(), uri.clone()) - .header("X-Vault-Token", &config::CONFIG_CENTRAL.pki_token) + .header("X-Vault-Token", &self.vault_token) .header("User-Agent", env!("SAMPLY_USER_AGENT")) .send() .await; @@ -193,7 +183,7 @@ impl GetCerts for GetCertsFromPki { let resp = self .resilient_vault_request( &Method::from_bytes("LIST".as_bytes()).unwrap(), - &format!("{}/certs", &config::CONFIG_CENTRAL.pki_realm), + &format!("{}/certs", &self.pki_realm), Some(100), ) .await?; @@ -253,7 +243,3 @@ impl GetCerts for GetCertsFromPki { parse_crl(&resp.bytes().await?).map(Some) } } - -fn pki_url_builder(location: &str) -> Url { - config::CONFIG_CENTRAL.pki_address.join(&format!("/v1/{location}")).unwrap() -} diff --git a/broker/src/main.rs b/broker/src/main.rs index 723df949..afda99ed 100644 --- a/broker/src/main.rs +++ b/broker/src/main.rs @@ -2,6 +2,7 @@ mod banner; mod crypto; +mod config; mod serve; mod serve_health; mod serve_pki; @@ -13,39 +14,38 @@ mod compare_client_server_version; use std::{collections::HashMap, sync::Arc, time::Duration}; +use clap::Parser; use crypto::GetCertsFromPki; use serve_health::{Health, InitStatus}; use once_cell::sync::Lazy; -use shared::{config::CONFIG_CENTRAL, *, errors::SamplyBeamError}; +use shared::{errors::SamplyBeamError, openssl::x509::X509, *}; use tokio::sync::RwLock; use tracing::{error, info, warn}; +use crate::{config::CliArgs, serve::BrokerState}; + #[tokio::main] pub async fn main() -> anyhow::Result<()> { - shared::logger::init_logger()?; + let args = CliArgs::parse(); + let _log_guard = shared::logger::init_logger(&args.log_options)?; banner::print_banner(); + let config = config::Config::load(args)?; let health = Arc::new(RwLock::new(Health::default())); - let cert_getter = GetCertsFromPki::new(health.clone())?; + let cert_getter = GetCertsFromPki::new(health.clone(), &config)?; shared::crypto::init_cert_getter(cert_getter); - tokio::task::spawn(init_broker_ca_chain(health.clone())); - #[cfg(debug_assertions)] - if shared::examples::print_example_objects() { - return Ok(()); - } - - Lazy::force(&config::CONFIG_CENTRAL); // Initialize config + tokio::task::spawn(init_broker_ca_chain(health.clone(), config.rootcert.clone())); - serve::serve(health).await?; + serve::serve(BrokerState::new(health, config)).await?; Ok(()) } -async fn init_broker_ca_chain(health: Arc>) { +async fn init_broker_ca_chain(health: Arc>, rootcert: X509) { { health.write().await.initstatus = InitStatus::FetchingIntermediateCert } - shared::crypto::init_ca_chain().await.expect("Failed to init broker ca chain"); + shared::crypto::init_ca_chain(&rootcert).await.expect("Failed to init broker ca chain"); health.write().await.initstatus = InitStatus::Done; } diff --git a/broker/src/serve.rs b/broker/src/serve.rs index 6ea6c054..bcd8c062 100644 --- a/broker/src/serve.rs +++ b/broker/src/serve.rs @@ -1,7 +1,7 @@ use std::{collections::HashMap, net::SocketAddr, sync::Arc}; use axum::{ - extract::{DefaultBodyLimit, Path, Query}, + extract::{DefaultBodyLimit, FromRef, Path, Query}, http::{header, StatusCode}, response::IntoResponse, routing::{get, post}, @@ -9,7 +9,7 @@ use axum::{ }; use serde::Deserialize; use shared::{ - config, EncryptedMsgTaskRequest, EncryptedMsgTaskResult, HasWaitId, HowLongToBlock, Msg, + EncryptedMsgTaskRequest, EncryptedMsgTaskResult, HasWaitId, HowLongToBlock, Msg, MsgEmpty, MsgId, MsgSigned, EMPTY_VEC_APPORPROXYID, }; use tokio::{ @@ -20,12 +20,13 @@ use tokio::{ }; use tracing::{debug, info, trace, warn}; -use crate::{banner, crypto, serve_health::Health, serve_health, serve_pki, serve_tasks, compare_client_server_version}; +use crate::{banner, compare_client_server_version, config::Config, crypto, serve_health::{self, Health}, serve_pki, serve_tasks}; -pub(crate) async fn serve(health: Arc>) -> anyhow::Result<()> { +pub(crate) async fn serve(broker_state: BrokerState) -> anyhow::Result<()> { + let bind_addr = broker_state.config.bind_addr; let app = serve_tasks::router() .merge(serve_pki::router()) - .merge(serve_health::router(health)); + .merge(serve_health::router(broker_state)); #[cfg(feature = "sockets")] let app = app.merge(crate::serve_sockets::router()); // Middleware needs to be set last @@ -34,12 +35,24 @@ pub(crate) async fn serve(health: Arc>) -> anyhow::Result<()> { .layer(axum::middleware::map_response(banner::set_server_header)) .layer(DefaultBodyLimit::disable()); - info!( - "Startup complete. Listening for requests on {}", - config::CONFIG_CENTRAL.bind_addr - ); - axum::serve(TcpListener::bind(&config::CONFIG_CENTRAL.bind_addr).await?, app.into_make_service_with_connect_info::()) + info!("Startup complete. Listening for requests on {bind_addr}"); + axum::serve(TcpListener::bind(bind_addr).await?, app.into_make_service_with_connect_info::()) .with_graceful_shutdown(shared::graceful_shutdown::wait_for_signal()) .await?; Ok(()) } + +#[derive(Debug, FromRef, Clone)] +pub struct BrokerState { + pub health: Arc>, + pub config: &'static Config, +} + +impl BrokerState { + pub fn new(health: Arc>, config: Config) -> Self { + Self { + health, + config: Box::leak(Box::new(config)), + } + } +} \ No newline at end of file diff --git a/broker/src/serve_health.rs b/broker/src/serve_health.rs index 585e38fa..fb188b89 100644 --- a/broker/src/serve_health.rs +++ b/broker/src/serve_health.rs @@ -1,15 +1,15 @@ use std::{collections::HashMap, convert::Infallible, marker::PhantomData, sync::Arc, time::{Duration, SystemTime}}; -use axum::{extract::{Path, State}, http::StatusCode, response::{sse::{Event, KeepAlive}, Response, Sse}, routing::get, Json, Router}; +use axum::{extract::{Path, State}, http::StatusCode, response::{sse::{Event, KeepAlive, KeepAliveStream}, Response, Sse}, routing::get, Json, Router}; use axum_extra::{headers::{authorization::Basic, Authorization}, TypedHeader}; use beam_lib::ProxyId; -use futures_core::Stream; +use futures::Stream; use serde::{Serialize, Deserialize}; -use shared::{crypto_jwt::Authorized, Msg, config::CONFIG_CENTRAL}; +use shared::{crypto_jwt::Authorized, Msg}; use tokio::sync::{Mutex, OwnedMutexGuard, RwLock}; use tracing::info; -use crate::compare_client_server_version::log_version_mismatch; +use crate::{compare_client_server_version::log_version_mismatch, serve::BrokerState}; #[derive(Serialize)] struct HealthOutput { @@ -70,13 +70,13 @@ impl ProxyStatus { } } -pub(crate) fn router(health: Arc>) -> Router { +pub(crate) fn router(state: BrokerState) -> Router { Router::new() .route("/v1/health", get(handler)) .route("/v1/health/proxies/{proxy_id}", get(proxy_health)) .route("/v1/health/proxies", get(get_all_proxies)) .route("/v1/control", get(get_control_tasks).layer(axum::middleware::from_fn(log_version_mismatch))) - .with_state(health) + .with_state(state) } // GET /v1/health @@ -104,11 +104,11 @@ async fn get_all_proxies(State(state): State>>) -> Json>>, + State(BrokerState { health, config }): State, Path(proxy): Path, auth: TypedHeader> ) -> Result<(StatusCode, Json), StatusCode> { - let Some(ref monitoring_key) = CONFIG_CENTRAL.monitoring_api_key else { + let Some(ref monitoring_key) = config.monitoring_api_key else { return Err(StatusCode::NOT_IMPLEMENTED); }; @@ -116,7 +116,7 @@ async fn proxy_health( return Err(StatusCode::UNAUTHORIZED) } - if let Some(reported_back) = state.read().await.proxies.get(&proxy) { + if let Some(reported_back) = health.read().await.proxies.get(&proxy) { if let Ok(last_disconnect) = reported_back.online_guard.try_lock().as_deref().copied() { Ok((StatusCode::SERVICE_UNAVAILABLE, Json(serde_json::json!({ "last_disconnect": last_disconnect @@ -132,7 +132,7 @@ async fn proxy_health( async fn get_control_tasks( State(state): State>>, proxy_auth: Authorized, -) -> Result, StatusCode> { +) -> Result>, StatusCode> { let proxy_id = proxy_auth.get_from().proxy_id(); // Once this is freed the connection will be removed from the map of connected proxies again // This ensures that when the connection is dropped and therefore this response future the status of this proxy will be updated diff --git a/broker/src/serve_pki.rs b/broker/src/serve_pki.rs index 2e124e67..31cceafb 100644 --- a/broker/src/serve_pki.rs +++ b/broker/src/serve_pki.rs @@ -11,7 +11,6 @@ use axum::{ }; use serde::{Deserialize, Serialize}; use shared::{ - config::CONFIG_CENTRAL, crypto_jwt::Authorized, errors::{CertificateInvalidReason, SamplyBeamError}, }; diff --git a/broker/src/serve_sockets.rs b/broker/src/serve_sockets.rs index f31c0c33..51caae93 100644 --- a/broker/src/serve_sockets.rs +++ b/broker/src/serve_sockets.rs @@ -4,9 +4,9 @@ use axum::{extract::{Path, Request, State}, http::{header, request::Parts, Heade use bytes::BufMut; use hyper_util::rt::TokioIo; use serde::{Serialize, Serializer, ser::SerializeSeq}; -use shared::{config::{CONFIG_CENTRAL, CONFIG_SHARED}, crypto_jwt::Authorized, expire_map::LazyExpireMap, serde_helpers::DerefSerializer, Encrypted, HasWaitId, HowLongToBlock, Msg, MsgEmpty, MsgId, MsgSigned, MsgSocketRequest}; +use shared::{format_to_without_broker, crypto_jwt::Authorized, expire_map::LazyExpireMap, serde_helpers::DerefSerializer, Encrypted, HasWaitId, HowLongToBlock, Msg, MsgEmpty, MsgId, MsgSigned, MsgSocketRequest}; use tokio::sync::{RwLock, broadcast::{Sender, self}, oneshot}; -use tracing::{debug, log::error, warn}; +use tracing::{debug, info, log::error, warn}; use crate::task_manager::{TaskManager, Task}; @@ -69,6 +69,7 @@ async fn post_socket_request( state: State, msg: MsgSigned>, ) -> Result { + info!(from = %msg.get_from().hide_broker(), to = %format_to_without_broker(&msg.get_to()), id = %msg.msg.id, "Submitted socket request"); let msg_id = msg.wait_id(); state.task_manager.post_task(msg)?; @@ -90,6 +91,7 @@ async fn connect_socket( Ok(msg) => msg.msg, Err(e) => return Ok(e.into_response()), }; + info!(from = %msg.get_from().hide_broker(), id = %task_id, "Connected to socket request"); { let task = state.task_manager.get(&task_id)?; // Allowed to connect are the issuer of the task and the recipient @@ -98,6 +100,12 @@ async fn connect_socket( } } + let tm = state.task_manager.clone(); + // Drop the task if any side that connected to it loses interest (aka drops the connection) + let _guard = DropGuard::new(move || { + // We don't care if the task expired by now + _ = tm.remove(&task_id); + }); let Some(conn) = parts.extensions.remove::() else { warn!("Failed to upgrade connection: {:#?}", parts.headers); return Err(StatusCode::UPGRADE_REQUIRED); @@ -115,8 +123,6 @@ async fn connect_socket( debug!("Socket expired because nobody connected"); return Err(StatusCode::GONE); }; - // We don't care if the task expired by now - _ = state.task_manager.remove(&task_id); tokio::spawn(async move { let (socket1, socket2) = match tokio::try_join!(conn, other_con) { Ok(sockets) => sockets, @@ -137,3 +143,21 @@ async fn connect_socket( (header::CONNECTION, HeaderValue::from_static("upgrade")) ], StatusCode::SWITCHING_PROTOCOLS).into_response()) } + +struct DropGuard { + on_drop: Option, +} + +impl DropGuard { + fn new(on_drop: F) -> Self { + Self { on_drop: Some(on_drop) } + } +} + +impl Drop for DropGuard { + fn drop(&mut self) { + if let Some(f) = self.on_drop.take() { + f(); + } + } +} diff --git a/broker/src/serve_tasks.rs b/broker/src/serve_tasks.rs index 6d047a99..27bdc0df 100644 --- a/broker/src/serve_tasks.rs +++ b/broker/src/serve_tasks.rs @@ -12,13 +12,11 @@ use axum::{ Json, Router, }; use beam_lib::AppOrProxyId; -use futures_core::{stream, Stream}; +use futures::{stream, Stream}; use serde::Deserialize; use beam_lib::WorkStatus; use shared::{ - config, errors::SamplyBeamError, sse_event::SseEventType, - EncryptedMsgTaskRequest, EncryptedMsgTaskResult, HasWaitId, HowLongToBlock, Msg, MsgEmpty, - MsgId, MsgSigned, MsgTaskRequest, MsgTaskResult, EMPTY_VEC_APPORPROXYID, serde_helpers::DerefSerializer, + EMPTY_VEC_APPORPROXYID, EncryptedMsgTaskRequest, EncryptedMsgTaskResult, HasWaitId, HowLongToBlock, Msg, MsgEmpty, MsgId, MsgSigned, MsgTaskRequest, MsgTaskResult, errors::SamplyBeamError, format_to_without_broker, serde_helpers::DerefSerializer, sse_event::SseEventType }; use tokio::{ sync::{ @@ -40,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) @@ -212,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>; @@ -339,12 +363,10 @@ async fn post_task( State(state): State, msg: MsgSigned, ) -> Result<(StatusCode, impl IntoResponse), StatusCode> { - // let id = MsgId::new(); - // msg.id = id; - // TODO: Check if ID is taken + info!(id = %msg.msg.id, from = %msg.msg.from.hide_broker(), to = %format_to_without_broker(&msg.msg.to), "New task"); trace!( - "Client {} with IP {addr} is creating task {:?}", - msg.msg.from, msg + "Client {} with IP {addr} is creating task {msg:?}", + msg.msg.from, ); let id = msg.msg.id; state.task_manager.post_task(msg)?; @@ -361,6 +383,7 @@ async fn put_result( State(state): State, result: MsgSigned, ) -> Result { + info!(for = %result.msg.task, from = %result.msg.from.hide_broker(), status = ?result.msg.status, "New result"); trace!("Called: Task {:?}, {:?} by {addr}", task_id, result); if task_id != result.msg.task { return Err(( @@ -376,7 +399,6 @@ async fn put_result( )); } - let status = if state.task_manager.put_result(&task_id, result)? { StatusCode::NO_CONTENT } else { diff --git a/broker/src/task_manager.rs b/broker/src/task_manager.rs index fded3089..dc6d5a13 100644 --- a/broker/src/task_manager.rs +++ b/broker/src/task_manager.rs @@ -4,7 +4,7 @@ use std::{ use axum::{response::{IntoResponse, sse::Event, Sse}, Json, http::StatusCode}; use dashmap::DashMap; -use futures_core::Stream; +use futures::Stream; use once_cell::sync::Lazy; use serde::Serialize; use serde_json::json; diff --git a/dev/beamdev b/dev/beamdev index e0c10519..a100f3ef 100755 --- a/dev/beamdev +++ b/dev/beamdev @@ -167,8 +167,9 @@ function stop { } function clean { - docker compose down + docker compose --profile "*" down rm -fv pki/*.pem pki/*.json pki/pki.secret + sudo rm -fv logs/*/*.log pki/pki clean } @@ -275,6 +276,14 @@ case "$1" in shift start_bg $@ ;; + start_cert_manager) + shift + clean + pki/pki devsetup + echo "$VAULT_TOKEN" > ./pki/pki.secret + build $@ + docker compose --profile "cert-manager" up --no-build --no-recreate + ;; restart) shift build $@ @@ -296,6 +305,6 @@ case "$1" in defaults ;; *) - echo "Usage: $0 [--tag SOMETAG, e.g. develop, to use an existing image] build|start|start_bg|restart|stop|clean|defaults|demo|noop" + echo "Usage: $0 [--tag SOMETAG, e.g. develop, to use an existing image] build|start|start_bg|start_cert_manager|restart|stop|clean|defaults|demo|noop" ;; esac diff --git a/dev/docker-compose.yml b/dev/docker-compose.yml index 4928a8f7..d0a6bed7 100644 --- a/dev/docker-compose.yml +++ b/dev/docker-compose.yml @@ -1,7 +1,7 @@ version: "3.7" services: vault: - image: hashicorp/vault + image: hashicorp/vault:1.21 ports: - 127.0.0.1:8200:8200 environment: @@ -31,7 +31,10 @@ services: BIND_ADDR: 0.0.0.0:8080 MONITORING_API_KEY: ${BROKER_MONITORING_KEY} RUST_LOG: ${RUST_LOG} + LOG_DIR: /logs # ALL_PROXY: http://mitmproxy:8080 + volumes: + - ./logs/broker:/logs secrets: - pki.secret - dummy.pem @@ -60,7 +63,10 @@ services: PRIVKEY_FILE: /run/secrets/proxy1.pem BIND_ADDR: 0.0.0.0:8081 RUST_LOG: ${RUST_LOG} + LOG_DIR: /logs # ALL_PROXY: http://mitmproxy:8080 + volumes: + - ./logs/proxy1:/logs secrets: - proxy1.pem - root.crt.pem @@ -82,10 +88,50 @@ services: PRIVKEY_FILE: /run/secrets/proxy2.pem BIND_ADDR: 0.0.0.0:8082 RUST_LOG: ${RUST_LOG} + LOG_DIR: /logs # ALL_PROXY: http://mitmproxy:8080 + volumes: + - ./logs/proxy2:/logs secrets: - proxy2.pem - root.crt.pem + + mailhog: + image: mailhog/mailhog + ports: + - "1025:1025" # SMTP port + - "8025:8025" # Web UI + profiles: + - cert-manager + + cert-ui: + image: samply/beam-cert-manager + ports: + - 8000:8000 + - 3000:3000 + environment: + BROKER_URL: http://broker:8080 + VAULT_URL: http://vault:8200 + CSR_DIR: /pki + SMTP_URL: smtp://mailhog:1025 + DB_DIR: /pki/db + BROKER_MONITORING_KEY: ${BROKER_MONITORING_KEY} + PKI_DEFAULT_ROLE: hd-dot-dktk-dot-com + PUBLIC_BASE_URL: http://localhost:3000 + ADMIN_ADDR: 0.0.0.0:8000 + BROKER_ID: broker + RUST_LOG: ${RUST_LOG} + volumes: + - ./pki:/pki + secrets: + - pki.secret + depends_on: + - vault + - broker + - mailhog + profiles: + - cert-manager + secrets: pki.secret: file: ./pki/pki.secret diff --git a/dev/pki/pki b/dev/pki/pki index 1d422b16..d301291e 100755 --- a/dev/pki/pki +++ b/dev/pki/pki @@ -17,7 +17,7 @@ function start() { } function clean() { - rm -vf *.pem *.json *.secret + rm -vf *.pem *.json *.secret *.csr docker compose down } @@ -95,17 +95,22 @@ function request() { application=$1 cn=$2 ttl=$3 - data="{\"common_name\": \"$cn\", \"ttl\": \"$ttl\"}" - echo $data + # Beam only supports multiple certificates per proxy when they share one key. + # devsetup enrolls some proxies repeatedly to create duplicate certs, so reuse + # an existing key on re-enrollment and generate a fresh one only the first time; + # otherwise senders may encrypt to a key the proxy no longer holds. + if [ -s "${application}.priv.pem" ] && openssl pkey -in "${application}.priv.pem" -noout 2>/dev/null; then + openssl req -new -key "${application}.priv.pem" -out ${application}.csr -subj "/CN=${cn}" 2>/dev/null + else + openssl req -new -newkey rsa:2048 -nodes -keyout ${application}.priv.pem -out ${application}.csr -subj "/CN=${cn}" 2>/dev/null + fi + data=$(jq -Rs '{common_name: "'$cn'", ttl: "'$ttl'", csr: .}' < ${application}.csr) echo "Creating Certificate for domain $cn" curl --header "X-Vault-Token: $VAULT_TOKEN" \ --request POST \ --data "$data" \ --no-progress-meter \ - $VAULT_ADDR/v1/samply_pki/issue/hd-dot-dktk-dot-com | jq > ${application}.json - cat ${application}.json | jq -r .data.certificate > ${application}.crt.pem - cat ${application}.json | jq -r .data.ca_chain[] > ${application}.chain.pem - cat ${application}.json | jq -r .data.private_key > ${application}.priv.pem + $VAULT_ADDR/v1/samply_pki/sign/hd-dot-dktk-dot-com | jq > ${application}.json echo "Success: PEM files stored to ${application}*.pem" } @@ -118,7 +123,7 @@ function setup() { #touch root.crt.pem # see https://github.com/docker/compose/issues/8305 start while ! [ "$(curl -s $VAULT_ADDR/v1/sys/health | jq -r .sealed)" == "false" ]; do echo "Waiting ..."; sleep 0.1; done - docker compose exec -T vault sh -c "https_proxy=$http_proxy apk add --no-cache bash curl jq" + docker compose exec -T vault sh -c "https_proxy=$http_proxy apk add --no-cache bash curl jq openssl" docker compose exec -T vault sh -c "VAULT_TOKEN=$VAULT_TOKEN http_proxy= HTTP_PROXY= PROXY1_ID=$PROXY1_ID PROXY2_ID=$PROXY2_ID /pki/pki init" docker compose exec -T vault sh -c "VAULT_TOKEN=$VAULT_TOKEN http_proxy= HTTP_PROXY= PROXY1_ID=$PROXY1_ID PROXY2_ID=$PROXY2_ID /pki/pki request_proxy $PROXY1_ID_SHORT" "24h" docker compose exec -T vault sh -c "VAULT_TOKEN=$VAULT_TOKEN http_proxy= HTTP_PROXY= PROXY1_ID=$PROXY1_ID PROXY2_ID=$PROXY2_ID /pki/pki request_proxy $PROXY2_ID_SHORT" "24h" diff --git a/proxy/Cargo.toml b/proxy/Cargo.toml index a4064e42..c30835e7 100644 --- a/proxy/Cargo.toml +++ b/proxy/Cargo.toml @@ -8,13 +8,15 @@ 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-proxy"] } +shared = { path = "../shared" } beam-lib = { workspace = true } tokio = { version = "1", features = ["full"] } axum = { version = "0.8", features = ["macros"] } bytes = { version = "1" } httpdate = "1.0" +clap.workspace = true +regex = "1" # Error handling anyhow = "1" @@ -28,25 +30,26 @@ serde = "1" serde_json = "1" # Encryption handling -rsa = "0.9" +rsa.workspace = true # Server-sent Events (SSE) support -tokio-util = { version = "0.7", features = ["io"] } -futures = "0.3" -async-sse = "5.1" +tokio-util = "0.7" +futures.workspace = true +sse-stream = "0.2.3" async-stream = "0.3" # Socket dependencies -chacha20poly1305 = { version = "0.10", features = ["stream"], optional = true } +aead-stream = { version = "0.6", features = ["alloc"], optional = true } +chacha20poly1305 = { workspace = true, features = ["bytes"], optional = true } dashmap = { version = "6.0", optional = true} hyper = { version = "1", default-features = false, optional = true } hyper-util = { version = "0.1", default-features = false, features = ["tokio"], optional = true} [features] -sockets = ["dep:chacha20poly1305", "dep:dashmap", "tokio-util/codec", "tokio-util/compat", "shared/sockets", "shared/expire_map", "dep:hyper", "dep:hyper-util"] +sockets = ["dep:aead-stream", "dep:chacha20poly1305", "dep:dashmap", "tokio-util/codec", "tokio-util/compat", "shared/sockets", "shared/expire_map", "dep:hyper", "dep:hyper-util"] [build-dependencies] build-data = "0" [dev-dependencies] -rand = "0.8.5" +rand.workspace = true diff --git a/proxy/src/auth.rs b/proxy/src/auth.rs index 96737bad..7d7b0e77 100644 --- a/proxy/src/auth.rs +++ b/proxy/src/auth.rs @@ -1,22 +1,25 @@ use std::collections::HashMap; use axum::{ - extract::{FromRequest, FromRequestParts}, + extract::{FromRef, FromRequest, FromRequestParts}, http::{header::{self, HeaderName}, request::Parts, Request, StatusCode}, }; use beam_lib::{AppId, AppOrProxyId}; -use shared::{ - config, config_proxy -}; use tracing::{debug, Span, debug_span, warn}; +use crate::config::Config; + pub(crate) struct AuthenticatedApp(pub(crate) AppId); -impl FromRequestParts for AuthenticatedApp { +impl FromRequestParts for AuthenticatedApp +where + &'static Config: FromRef, + S: Sync, +{ type Rejection = (StatusCode, [(HeaderName, &'static str); 1]); - async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { + async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { const SCHEME: &str = "ApiKey"; const UNAUTH_ERR: (StatusCode, [(HeaderName, &str); 1]) = ( StatusCode::UNAUTHORIZED, @@ -33,7 +36,8 @@ impl FromRequestParts for AuthenticatedApp { warn!(auth_str, "Invalid app id"); return Err(UNAUTH_ERR); }; - let Some(api_key_actual) = config::CONFIG_PROXY.api_keys.get(&client_id) else { + let config = <&'static Config>::from_ref(state); + let Some(api_key_actual) = config.api_keys.get(&client_id) else { warn!("App {client_id} not registered in proxy"); return Err(UNAUTH_ERR); }; @@ -43,7 +47,7 @@ impl FromRequestParts for AuthenticatedApp { return Err(UNAUTH_ERR); } debug!("Request authenticated (ClientID {})", client_id); - Span::current().record("from", AppOrProxyId::App(client_id.clone()).hide_broker()); + Span::current().record("from", client_id.hide_broker_name()); Ok(Self(client_id)) } else { warn!("No auth header provided"); diff --git a/shared/src/config_proxy.rs b/proxy/src/config.rs similarity index 65% rename from shared/src/config_proxy.rs rename to proxy/src/config.rs index fda55e29..5baee9ae 100644 --- a/shared/src/config_proxy.rs +++ b/proxy/src/config.rs @@ -1,11 +1,12 @@ use clap::Parser; -use openssl::x509::X509; use regex::Regex; use reqwest::Url; +use rsa::{pkcs1::DecodeRsaPrivateKey, pkcs8::DecodePrivateKey, RsaPrivateKey}; +use shared::{errors::SamplyBeamError, jwt_simple::prelude::RS256KeyPair, logger::LogOptions, openssl::x509::X509, reqwest}; use std::{ collections::HashMap, - fs::read_to_string, + fs::{self, read_to_string}, net::SocketAddr, path::{Path, PathBuf}, process::exit, @@ -17,7 +18,6 @@ use serde::Deserialize; use tracing::{debug, info, warn}; use beam_lib::{AppId, ProxyId}; -use crate::errors::SamplyBeamError; #[derive(Clone, Debug)] pub struct Config { @@ -27,6 +27,14 @@ pub struct Config { pub proxy_id: ProxyId, pub api_keys: HashMap, pub tls_ca_certificates: Vec, + pub crypto: ConfigCrypto, + pub rootcert: X509, +} + +#[derive(Debug, Clone)] +pub struct ConfigCrypto { + pub privkey_rs256: RS256KeyPair, + pub privkey_rsa: RsaPrivateKey, } pub type ApiKey = String; @@ -36,7 +44,7 @@ pub type ApiKey = String; name("🌈 Samply.Beam.Proxy"), version, arg_required_else_help(true), - after_help(crate::config_shared::CLAP_FOOTER) + after_help(shared::CLAP_FOOTER) )] pub struct CliArgs { /// Local bind address @@ -63,9 +71,8 @@ pub struct CliArgs { #[clap(long, env, value_parser, default_value = "/run/secrets/root.crt.pem")] rootcert_file: PathBuf, - /// (included for technical reasons) - #[clap(long, hide(true))] - test_threads: Option, + #[clap(flatten)] + pub log_options: LogOptions, } pub const APP_PREFIX: &str = "APP"; @@ -76,6 +83,7 @@ pub const APP_PREFIX: &str = "APP"; fn parse_apikeys(proxy_id: &ProxyId) -> Result, SamplyBeamError> { let env_vars = std::env::vars().collect::>(); let mut api_keys = HashMap::new(); + // TODO: Do we really need a regex for that? let pattern = Regex::new(&format!("{APP_PREFIX}_([A-Za-z0-9-]+)_KEY")).expect("This is a valid regex"); for (env_var_name, secret) in env_vars { if let Some(app_name) = pattern.captures_iter(&env_var_name).next().and_then(|cap| cap.get(1)) { @@ -95,9 +103,8 @@ fn parse_apikeys(proxy_id: &ProxyId) -> Result, SamplyBea Ok(api_keys) } -impl crate::config::Config for Config { - fn load() -> Result { - let cli_args = CliArgs::parse(); +impl Config { + pub fn load(cli_args: CliArgs) -> Result { beam_lib::set_broker_id(cli_args.broker_url.host().unwrap().to_string()); let proxy_id = ProxyId::new(&cli_args.proxy_id).map_err(|e| { SamplyBeamError::ConfigurationFailed(format!( @@ -109,7 +116,7 @@ impl crate::config::Config for Config { if api_keys.is_empty() { return Err(SamplyBeamError::ConfigurationFailed(format!("No API keys have been defined. Please set environment vars à la {0}__KEY=", APP_PREFIX))); } - let tls_ca_certificates = crate::crypto::load_certificates_from_dir( + let tls_ca_certificates = shared::crypto::load_certificates_from_dir( cli_args.tls_ca_certificates_dir, ) .map_err(|e| { @@ -122,15 +129,61 @@ impl crate::config::Config for Config { broker_host_header: uri_to_host_header(&cli_args.broker_url)?, broker_uri: cli_args.broker_url, bind_addr: cli_args.bind_addr, + crypto: load_private_crypto_for_proxy(&cli_args.privkey_file, &proxy_id)?, proxy_id, api_keys, tls_ca_certificates, + rootcert: shared::crypto::load_certificates_from_file(cli_args.rootcert_file)?, }; info!("Successfully read config and API keys from CLI and secrets file."); Ok(config) } } +fn load_private_crypto_for_proxy(privkey_file: &PathBuf, proxy_id: &ProxyId) -> Result { + let privkey_pem = fs::read_to_string(privkey_file) + .map_err(|e| { + SamplyBeamError::ConfigurationFailed(format!( + "Unable to load private key from file {}: {}\n{}", + privkey_file.display(), + e, + get_enrollment_msg(proxy_id.as_ref()) + )) + })? + .trim() + .to_string(); + let privkey_rsa = RsaPrivateKey::from_pkcs1_pem(&privkey_pem) + .or_else(|_| RsaPrivateKey::from_pkcs8_pem(&privkey_pem)) + .map_err(|e| { + SamplyBeamError::ConfigurationFailed(format!( + "Unable to interpret private key PEM as PKCS#1 or PKCS#8: {}", + e + )) + })?; + let privkey_rs256 = RS256KeyPair::from_pem(&privkey_pem).map_err(|e| { + SamplyBeamError::ConfigurationFailed(format!( + "Unable to interpret private key PEM as PKCS#1 or PKCS#8: {}", + e + )) + })?; + Ok(ConfigCrypto { + privkey_rs256, + privkey_rsa, + }) +} + +fn get_enrollment_msg(proxy_id: &str) -> String { + let divider = "***************************************************************************\n + *** Beam Certificate Enrollment Warning ***\n + ***************************************************************************"; + format!( + "{}\nIf you are not yet enrolled in the central certificate store, please execute the beam-enroll companion tool (https://github.com/samply/beam-enroll) by executing:\n docker run --rm -it -v \"$(pwd)\":/data -e PROXY_ID={} samply/beam-enroll\nand follow the steps on the screen.\nAfter your certificate signing request (CSR) has been approved, please restart this Beam.Proxy and this message should disappear.", + divider, + proxy_id + ) +} + + fn uri_to_host_header(uri: &Url) -> Result { let hostname: String = uri .host() diff --git a/proxy/src/crypto.rs b/proxy/src/crypto.rs index 4fb54323..1dc313b6 100644 --- a/proxy/src/crypto.rs +++ b/proxy/src/crypto.rs @@ -1,19 +1,25 @@ +use std::{fs, path::PathBuf}; + use axum::{body::Bytes, http::{header, request, Method, Request, StatusCode, Uri}, response::Response, Json}; -use beam_lib::AppOrProxyId; +use beam_lib::{AppOrProxyId, ProxyId}; +use rsa::{pkcs1::{DecodeRsaPrivateKey, DecodeRsaPublicKey}, pkcs8::DecodePrivateKey, RsaPrivateKey, RsaPublicKey}; use shared::{ - async_trait, config, config_proxy::Config, config_shared::ConfigCrypto, crypto::GetCerts, errors::{CertificateInvalidReason, SamplyBeamError}, http_client::SamplyHttpClient, reqwest, EncryptedMessage, MsgEmpty + async_trait, crypto::{self, asn_str_to_vault_str, get_all_certs_and_clients_by_cname_as_pemstr, get_best_own_certificate, x509_cert_to_x509_public_key, CryptoPublicPortion, GetCerts, ProxyCertInfo}, errors::{CertificateInvalidReason, SamplyBeamError}, http_client::SamplyHttpClient, jwt_simple::prelude::RS256KeyPair, openssl::x509::X509, reqwest, EncryptedMessage, MsgEmpty }; use tracing::{debug, info, warn, error}; -use crate::serve_tasks::sign_request; +use crate::{config::{self, Config}, serve_tasks::sign_request}; pub(crate) struct GetCertsFromBroker { client: SamplyHttpClient, config: Config, - crypto_conf: ConfigCrypto, } impl GetCertsFromBroker { + pub fn new(client: SamplyHttpClient, config: Config) -> Self { + Self { client, config } + } + async fn request(&self, path: &str) -> Result { let uri = Uri::builder() .scheme(self.config.broker_uri.scheme()) @@ -33,7 +39,7 @@ impl GetCertsFromBroker { .expect("To build request successfully") .into_parts(); - let req = sign_request(body, parts, &self.config, &self.crypto_conf) + let req = sign_request(body, parts, &self.config) .await .map_err(|(_, msg)| SamplyBeamError::SignEncryptError(msg.into()))?; Ok(self.client.execute(req).await?.into()) @@ -93,15 +99,33 @@ impl GetCerts for GetCertsFromBroker { } } -pub(crate) fn build_cert_getter( - config: Config, - client: SamplyHttpClient, - crypto_conf: ConfigCrypto, -) -> Result { - let client = client; - Ok(GetCertsFromBroker { - client, - config, - crypto_conf, - }) +pub async fn init_public_crypto_for_proxy( + config: &Config +) -> Result<(ProxyCertInfo, config::ConfigCrypto), SamplyBeamError> { + let (public_info, new_crypto) = load_public_crypto_for_proxy(config).await?; + + let cert_info = ProxyCertInfo::try_from(&public_info.cert)?; + Ok((cert_info, new_crypto)) } + +pub async fn load_public_crypto_for_proxy( + config: &Config, +) -> Result<(CryptoPublicPortion, config::ConfigCrypto), SamplyBeamError> { + let publics: Vec = get_all_certs_and_clients_by_cname_as_pemstr(&config.proxy_id) + .await + .into_iter() + .filter_map(|r| { + r.map_err(|e| debug!("Unable to parse Certificate: {e}")) + .ok() + }) + .collect(); + let public = get_best_own_certificate(publics, &config.crypto.privkey_rsa).ok_or( + SamplyBeamError::SignEncryptError( + "Unable to choose valid, newest certificate for this proxy".into(), + ), + )?; + let serial = asn_str_to_vault_str(public.cert.serial_number())?; + let mut crypto_with_kid = config.crypto.clone(); + crypto_with_kid.privkey_rs256 = crypto_with_kid.privkey_rs256.with_key_id(&serial); + Ok((public, crypto_with_kid)) +} \ No newline at end of file diff --git a/proxy/src/main.rs b/proxy/src/main.rs index 7aae2e4c..1659db7d 100644 --- a/proxy/src/main.rs +++ b/proxy/src/main.rs @@ -5,21 +5,25 @@ use std::time::Duration; use axum::http::{header, HeaderValue, StatusCode}; use beam_lib::AppOrProxyId; +use clap::Parser; use futures::future::Ready; -use futures::{StreamExt, TryStreamExt}; +use futures::StreamExt; use shared::{reqwest, EncryptedMessage, MsgEmpty, PlainMessage}; -use shared::crypto::{get_own_crypto_material, CryptoPublicPortion, ProxyCertInfo}; +use shared::crypto::{CryptoPublicPortion, ProxyCertInfo}; use shared::errors::SamplyBeamError; use shared::http_client::{self, SamplyHttpClient}; -use shared::{config, config_proxy::Config}; +use sse_stream::SseStream; use tokio::time::Instant; use tracing::{debug, error, info, warn}; use tryhard::{backoff_strategies::ExponentialBackoff, RetryFuture, RetryFutureConfig}; -use crate::serve_tasks::sign_request; +use crate::config::{CliArgs, Config}; +use crate::crypto::GetCertsFromBroker; +use crate::serve_tasks::{sign_request, sse_error_is_timeout}; mod auth; mod banner; +mod config; mod crypto; mod serve; mod serve_health; @@ -31,15 +35,32 @@ pub(crate) const PROXY_TIMEOUT: u64 = 120; #[tokio::main] pub async fn main() -> anyhow::Result<()> { - shared::logger::init_logger()?; + let args = CliArgs::parse(); + let _log_guard = shared::logger::init_logger(&args.log_options)?; banner::print_banner(); - let config = config::CONFIG_PROXY.clone(); - let client = http_client::build( - &config::CONFIG_SHARED.tls_ca_certificates, + let config = Config::load(args)?; + let retry_policy = reqwest::retry::for_host(config.broker_uri.host_str().unwrap().to_string()) + .classify_fn(|res| { + if res.method() != reqwest::Method::GET { + return res.success(); + } + if let Some(StatusCode::BAD_GATEWAY | StatusCode::SERVICE_UNAVAILABLE | StatusCode::GATEWAY_TIMEOUT) = res.status() { + res.retryable() + } else if res.error().and_then(|e| e.downcast_ref::()).is_some_and(reqwest::Error::is_timeout) { + res.retryable() + } else { + res.success() + } + }) + .max_extra_load(2.0) + .max_retries_per_request(3); + + let client = http_client::builder( + &config.tls_ca_certificates, Some(Duration::from_secs(PROXY_TIMEOUT)), Some(Duration::from_secs(20)), - )?; + ).retry(retry_policy).build()?; if let Err(err) = retry_notify(|| get_broker_health(&config, &client), |err, dur| { warn!("Still trying to reach Broker: {err}. Retrying in {}s", dur.as_secs()); @@ -50,26 +71,38 @@ pub async fn main() -> anyhow::Result<()> { info!("Connected to Broker: {}", &config.broker_uri); } - if let Err(err) = retry_notify(|| init_crypto(config.clone(), client.clone()), |err, dur| { + shared::crypto::init_cert_getter(GetCertsFromBroker::new( + client.clone(), + config.clone(), + )); + let result = retry_notify(|| init_crypto(&config), |err, dur| { warn!("Still trying to initialize certificate chain: {err}. Retrying in {}s", dur.as_secs()); - }).await { - error!("Giving up on initializing certificate chain: {}", err); - std::process::exit(1); - } else { - debug!("Certificate chain successfully initialized and validated"); - } - spawn_controller_polling(client.clone(), config.clone()); + }).await; + let config = match result { + Err(err) => { + error!("Giving up on initializing certificate chain: {}", err); + std::process::exit(1); + } + Ok(crypto_config) => { + debug!("Certificate chain successfully initialized and validated"); + Box::leak(Box::new(Config { + crypto: crypto_config, + ..config + })) + } + }; + spawn_controller_polling(client.clone(), config); serve::serve(config, client).await?; Ok(()) } fn retry_notify(f: F, on_error: Cb) -> RetryFuture, &E) -> Ready<()>>> -where +where F: FnMut() -> Fut, Fut: Future>, Cb: Fn(&E, Duration) + 'static, - + { tryhard::retry_fn(f) .retries(100) @@ -78,14 +111,8 @@ where .on_retry(Box::new(move |_, b, e| futures::future::ready(on_error(e, b.unwrap_or(Duration::MAX))))) } -async fn init_crypto(config: Config, client: SamplyHttpClient) -> Result<(), SamplyBeamError> { - let private_crypto_proxy = shared::config_shared::load_private_crypto_for_proxy()?; - shared::crypto::init_cert_getter(crypto::build_cert_getter( - config.clone(), - client.clone(), - private_crypto_proxy.clone(), - )?); - shared::crypto::init_ca_chain().await?; +async fn init_crypto(config: &Config) -> Result { + shared::crypto::init_ca_chain(&config.rootcert).await?; let _public_info: Vec<_> = shared::crypto::get_all_certs_and_clients_by_cname_as_pemstr(&config.proxy_id) @@ -96,15 +123,15 @@ async fn init_crypto(config: Config, client: SamplyHttpClient) -> Result<(), Sam .ok() }) .collect(); - let ProxyCertInfo { serial, common_name, .. } = - shared::config_shared::init_public_crypto_for_proxy(private_crypto_proxy).await?; - if common_name != config.proxy_id.to_string() { + let (ProxyCertInfo { serial, common_name, .. }, new_crpto) = + crate::crypto::init_public_crypto_for_proxy(&config).await?; + if &common_name != config.proxy_id.as_ref() { return Err(SamplyBeamError::ConfigurationFailed(format!("Unable to retrieve a certificate matching your Proxy ID. Expected {common_name}, got {}. Please check your configuration", config.proxy_id.as_ref()))); } info!("Certificate retrieved for our proxy ID {common_name} (serial {serial})"); - Ok(()) + Ok(new_crpto) } async fn get_broker_health( @@ -129,7 +156,7 @@ async fn get_broker_health( } } -fn spawn_controller_polling(client: SamplyHttpClient, config: Config) { +fn spawn_controller_polling(client: SamplyHttpClient, config: &'static Config) { const RETRY_INTERVAL: Duration = Duration::from_secs(60); tokio::spawn(async move { let mut retries_this_min = 0; @@ -148,7 +175,7 @@ fn spawn_controller_polling(client: SamplyHttpClient, config: Config) { .expect("To build request successfully") .into_parts(); - let req = sign_request(body, parts, &config, &get_own_crypto_material()).await.expect("Unable to sign request; this should always work"); + let req = sign_request(body, parts, &config).await.expect("Unable to sign request; this should always work"); // In the future this will poll actual control related tasks let res = match client.execute(req).await { Ok(res) if res.status() == StatusCode::CONFLICT => { @@ -172,18 +199,11 @@ fn spawn_controller_polling(client: SamplyHttpClient, config: Config) { continue; } }; - let incoming = res - .bytes_stream() - .map(|result| result.map_err(|error| { - let kind = error.is_timeout().then_some(std::io::ErrorKind::TimedOut).unwrap_or(std::io::ErrorKind::Other); - std::io::Error::new(kind, format!("IO Error: {error}")) - })) - .into_async_read(); - let mut reader = async_sse::decode(incoming); + let mut reader = SseStream::from_byte_stream(res.bytes_stream()); while let Some(ev) = reader.next().await { match ev { Ok(_)=> (), - Err(e) if e.downcast_ref::().unwrap().kind() == std::io::ErrorKind::TimedOut => { + Err(e) if sse_error_is_timeout(&e) => { debug!("SSE connection timed out"); break; }, diff --git a/proxy/src/serve.rs b/proxy/src/serve.rs index 238f062a..19cfc894 100644 --- a/proxy/src/serve.rs +++ b/proxy/src/serve.rs @@ -2,7 +2,7 @@ use std::{fmt::Write, net::SocketAddr}; use axum::extract::DefaultBodyLimit; use shared::{ - config, config_proxy, config_shared, errors::SamplyBeamError, http_client::SamplyHttpClient, + errors::SamplyBeamError, http_client::SamplyHttpClient, }; use tokio::net::TcpListener; use tracing::{debug, error, info, warn}; @@ -10,17 +10,17 @@ use tracing::{debug, error, info, warn}; use crate::{banner, serve_health, serve_tasks}; pub(crate) async fn serve( - config: config_proxy::Config, + config: &'static crate::Config, client: SamplyHttpClient, ) -> anyhow::Result<()> { - let router_tasks = serve_tasks::router(&client); + let router_tasks = serve_tasks::router(&client, config); let router_health = serve_health::router(); let app = router_tasks.merge(router_health); #[cfg(feature = "sockets")] - let app = app.merge(crate::serve_sockets::router(client)); + let app = app.merge(crate::serve_sockets::router(client, config)); // Middleware needs to be set last let app = app .layer(axum::middleware::from_fn(shared::middleware::log)) diff --git a/proxy/src/serve_sockets.rs b/proxy/src/serve_sockets.rs index b5346c56..13cf8d44 100644 --- a/proxy/src/serve_sockets.rs +++ b/proxy/src/serve_sockets.rs @@ -1,30 +1,21 @@ use std::{ - io::{self, Write}, ops::{Deref, DerefMut}, pin::Pin, sync::{atomic::AtomicUsize, Arc}, task::Poll, time::{Duration, Instant, SystemTime} + io, mem::size_of, pin::Pin, sync::{atomic::AtomicUsize, Arc}, time::{Duration, Instant, SystemTime} }; use axum::{ extract::{Path, Request, State}, http::{self, header, HeaderValue, StatusCode}, response::{IntoResponse, Response}, routing::{get, post}, Extension, Json, RequestPartsExt, Router }; +use aead_stream::{DecryptorLE31, EncryptorLE31, Nonce, StreamLE31}; use bytes::{Buf, BufMut, BytesMut}; -use chacha20poly1305::{ - aead::{ - self, - generic_array::{typenum::Unsigned, GenericArray}, - stream::{DecryptorLE31, EncryptorLE31, NewStream, StreamLE31, StreamPrimitive}, - Buffer, Nonce, OsRng, - }, - consts::{U20, U32}, - AeadCore, AeadInPlace, ChaCha20Poly1305, KeyInit, XChaCha20Poly1305, -}; +use chacha20poly1305::{aead::{array::typenum::Unsigned, AeadCore, Generate}, Key, KeyInit, XChaCha20Poly1305}; use dashmap::DashMap; use futures::{stream::IntoAsyncRead, FutureExt, SinkExt, StreamExt, TryStreamExt}; use hyper_util::rt::TokioIo; -use rsa::rand_core::RngCore; use serde::{Deserialize, Serialize}; use serde_json::Value; use beam_lib::AppOrProxyId; use shared::{ - config, ct_codecs::{self, Base64UrlSafeNoPadding, Decoder as B64Decoder, Encoder as B64Encoder}, expire_map::LazyExpireMap, http_client::SamplyHttpClient, reqwest, MessageType, MsgEmpty, MsgId, MsgSocketRequest, Plain + ct_codecs::{self, Base64UrlSafeNoPadding, Decoder as B64Decoder, Encoder as B64Encoder}, expire_map::LazyExpireMap, http_client::SamplyHttpClient, reqwest, MessageType, MsgEmpty, MsgId, MsgSocketRequest, Plain }; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf, ReadHalf, WriteHalf}; use tokio_util::{ @@ -34,15 +25,17 @@ use tokio_util::{ use tracing::{warn, debug}; use crate::{ - auth::AuthenticatedApp, - serve_tasks::{forward_request, handler_task, TasksState, validate_and_decrypt, to_server_error}, + auth::AuthenticatedApp, config::Config, serve_tasks::{forward_request, handler_task, to_server_error, validate_and_decrypt, TasksState} }; type MsgSecretMap = Arc>; +type StreamNonce = Nonce>; +type FrameLen = u32; +const FRAME_LEN_SIZE: usize = size_of::(); +const TAG_SIZE: usize = ::TagSize::USIZE; const TASK_SECRET_CLEANUP_INTERVAL: Duration = Duration::from_secs(5 * 60); -pub(crate) fn router(client: SamplyHttpClient) -> Router { - let config = config::CONFIG_PROXY.clone(); +pub(crate) fn router(client: SamplyHttpClient, config: &'static Config) -> Router { let state = TasksState { client: client.clone(), config, @@ -74,7 +67,7 @@ async fn get_tasks( return Err(http::Response::from(res).map(axum::body::Body::new)); } let enc_json = res.json().await.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())?; - let plain_json = to_server_error(validate_and_decrypt(enc_json).await).map_err(IntoResponse::into_response)?; + let plain_json = to_server_error(validate_and_decrypt(enc_json, state.config).await).map_err(IntoResponse::into_response)?; let tasks: Vec> = serde_json::from_value(plain_json).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())?; let mut out = Vec::with_capacity(tasks.len()); for task in tasks { @@ -238,17 +231,15 @@ async fn connect_socket( } #[derive(Debug, Clone, Copy)] -struct SocketEncKey(GenericArray); +struct SocketEncKey(Key); impl SocketEncKey { fn generate() -> Self { - let mut arr = GenericArray::default(); - OsRng.fill_bytes(&mut arr); - SocketEncKey(arr) + SocketEncKey(Key::generate()) } fn to_b64_str(&self) -> Result { - Base64UrlSafeNoPadding::encode_to_string(self.as_slice()) + Base64UrlSafeNoPadding::encode_to_string(self.0.as_slice()) } } @@ -267,25 +258,9 @@ impl<'de> Deserialize<'de> for SocketEncKey { D: serde::Deserializer<'de>, { let bytes = Base64UrlSafeNoPadding::decode_to_vec(String::deserialize(deserializer)?, None).map_err(serde::de::Error::custom)?; - if bytes.len() != U32::to_usize() { - return Err(serde::de::Error::custom("Key does not match required key length")); - } else { - Ok(SocketEncKey(GenericArray::clone_from_slice(bytes.as_slice()))) - } - } -} - -impl Deref for SocketEncKey { - type Target = GenericArray; - - fn deref(&self) -> &GenericArray { - &self.0 - } -} - -impl DerefMut for SocketEncKey { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.0 + let key = Key::try_from(bytes.as_slice()) + .map_err(|_| serde::de::Error::custom("Key does not match required key length"))?; + Ok(SocketEncKey(key)) } } @@ -303,111 +278,71 @@ struct DecryptorCodec { decryptor: DecryptorLE31, } -impl DecryptorCodec { - const SIZE_OVERHEAD: usize = 4; -} - -impl EncryptorCodec { - const TAG_SIZE: usize = ::TagSize::USIZE; -} - impl Encoder<&[u8]> for EncryptorCodec { type Error = io::Error; fn encode(&mut self, item: &[u8], dst: &mut BytesMut) -> Result<(), Self::Error> { - let mut enc_buf = EncBuffer::new(dst, (item.len() + Self::TAG_SIZE).try_into().expect("item to large")); - enc_buf.extend_from_slice(item).expect("Infallible"); - self.encryptor - .encrypt_next_in_place(b"", &mut enc_buf) - .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Encryption failed")) + let frame_len: FrameLen = item + .len() + .checked_add(TAG_SIZE) + .and_then(|l| l.try_into().ok()) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "item too large"))?; + + dst.reserve(FRAME_LEN_SIZE + frame_len as usize); + let frame_start = dst.len(); + dst.put_u32_le(frame_len); + dst.extend_from_slice(item); + let mut frame = dst.split_off(frame_start + FRAME_LEN_SIZE); + self + .encryptor + .encrypt_next_in_place(&[], &mut frame) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Encryption failed"))?; + dst.unsplit(frame); + Ok(()) } } impl Decoder for DecryptorCodec { - type Item = Vec; + type Item = BytesMut; type Error = io::Error; fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { - if src.len() < Self::SIZE_OVERHEAD { + if src.len() < FRAME_LEN_SIZE { return Ok(None); } - let mut size_slice = [0; Self::SIZE_OVERHEAD]; - size_slice.clone_from_slice(&src[..Self::SIZE_OVERHEAD]); - let size = u32::from_le_bytes(size_slice); - let total_frame_size = size as usize + Self::SIZE_OVERHEAD; + let mut size_slice = [0; FRAME_LEN_SIZE]; + size_slice.copy_from_slice(&src[..FRAME_LEN_SIZE]); + let size = FrameLen::from_le_bytes(size_slice) as usize; + let total_frame_size = size + FRAME_LEN_SIZE; if src.len() < total_frame_size { return Ok(None); } - let plain = self + src.advance(FRAME_LEN_SIZE); + let mut frame = src.split_to(size); + self .decryptor - .decrypt_next(&src[Self::SIZE_OVERHEAD..(Self::SIZE_OVERHEAD + size as usize)]) + .decrypt_next_in_place(&[], &mut frame) .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Decryption failed"))?; - src.advance(total_frame_size); - Ok(Some(plain)) - } -} - -struct EncBuffer<'a> { - buf: &'a mut BytesMut, - /// The index from which the inplace encryption takes place - enc_idx: usize, -} - -impl<'a> EncBuffer<'a> { - fn new(buffer: &'a mut BytesMut, content_len: u32) -> Self { - let enc_idx = buffer.len() + Self::SIZE_OVERHEAD; - buffer.reserve(content_len as usize + Self::SIZE_OVERHEAD); - buffer.extend_from_slice(&u32::to_le_bytes(content_len)); - Self { buf: buffer, enc_idx } - } - - /// Reserved for size of msg - const SIZE_OVERHEAD: usize = (u32::BITS / 8) as usize; -} - -impl<'a> Buffer for EncBuffer<'a> { - // This should only be called to append the tag to the buffer - fn extend_from_slice(&mut self, other: &[u8]) -> aead::Result<()> { - self.buf.extend_from_slice(other); - Ok(()) - } - - // This should only be called when decrypting - fn truncate(&mut self, len: usize) { - warn!("Buffer got truncated. This should never happen as it should be perfectly sized"); - self.buf.truncate(self.enc_idx + len) - } -} - -impl<'a> AsRef<[u8]> for EncBuffer<'a> { - fn as_ref(&self) -> &[u8] { - &self.buf[self.enc_idx..] - } -} - -impl<'a> AsMut<[u8]> for EncBuffer<'a> { - fn as_mut(&mut self) -> &mut [u8] { - &mut self.buf[self.enc_idx..] + Ok(Some(frame)) } } impl EncryptedSocket { - /// Creates a new cipher stream with a 32 byte key and a 16 + 4 byte Nonce - async fn new(mut inner: S, key: &GenericArray) -> io::Result { - let aead = XChaCha20Poly1305::new(key); + /// Creates a new cipher stream from a shared AEAD key and per-direction STREAM nonces. + async fn new(mut inner: S, key: &SocketEncKey) -> io::Result { + let cipher = XChaCha20Poly1305::new(&key.0); // Encryption - let mut enc_nonce = GenericArray::default(); - OsRng.fill_bytes(&mut enc_nonce); - let encryptor = EncryptorLE31::from_aead(aead.clone(), &enc_nonce); + let enc_nonce = StreamNonce::generate(); + let encryptor = EncryptorLE31::from_aead(cipher.clone(), &enc_nonce); inner.write_all(&enc_nonce).await?; // Decryption - let mut dec_nonce = GenericArray::default(); + let mut dec_nonce = StreamNonce::default(); inner.read_exact(dec_nonce.as_mut_slice()).await?; - let decryptor = DecryptorLE31::from_aead(aead, &dec_nonce); + let decryptor = DecryptorLE31::from_aead(cipher, &dec_nonce); let (r, w) = tokio::io::split(inner); let read = FramedRead::new(r, DecryptorCodec { decryptor }); @@ -454,19 +389,17 @@ impl AsyncWrite for EncryptedSocket { #[cfg(test)] mod tests { - use chacha20poly1305::aead::stream::{Decryptor, Encryptor, EncryptorLE31}; - use rand::Rng; + use rand::RngExt; use tokio::net::{TcpListener, TcpStream}; use super::*; #[tokio::test] async fn test_encryption() { - let mut key = GenericArray::default(); - OsRng.fill_bytes(&mut key); + let key = SocketEncKey::generate(); let data: Arc> = (0..13337).map(|_| { - let mut chunk = vec![0; OsRng.gen_range(1..9999)]; - OsRng.fill_bytes(&mut chunk); + let mut chunk = vec![0; rand::rng().random_range(1..9999)]; + rand::rng().fill(&mut chunk[..]); chunk }).collect::>().into(); @@ -503,7 +436,7 @@ mod tests { }); } - async fn client(key: &GenericArray) -> impl AsyncRead + AsyncWrite { + async fn client(key: &SocketEncKey) -> impl AsyncRead + AsyncWrite { // Wait for server to start tokio::time::sleep(Duration::from_millis(100)).await; let stream = TcpStream::connect("127.0.0.1:1337").await.unwrap(); @@ -512,27 +445,17 @@ mod tests { #[test] fn normal_enc() { - let mut key = GenericArray::default(); - OsRng.fill_bytes(&mut key); const N: usize = 2_usize.pow(10); let test_data: &mut [u8; N] = &mut [0; N]; - OsRng.fill_bytes(test_data); - - let mut nonce = GenericArray::::default(); - OsRng.fill_bytes(&mut nonce); - let aead = XChaCha20Poly1305::new(&key); - let client1 = StreamLE31::from_aead(aead, &nonce); - let mut encryptor = Encryptor::from_stream_primitive(client1); - - // let mut nonce = GenericArray::::default(); - // OsRng.fill_bytes(&mut nonce); - let aead = XChaCha20Poly1305::new(&key); - let client2 = StreamLE31::from_aead(aead, &nonce); - let mut decrypter = Decryptor::from_stream_primitive(client2); + rand::rng().fill(&mut test_data[..]); + let key = SocketEncKey::generate(); + let cipher = XChaCha20Poly1305::new(&key.0); + let nonce = StreamNonce::generate(); + let mut encryptor = EncryptorLE31::from_aead(cipher.clone(), &nonce); + let mut decryptor = DecryptorLE31::from_aead(cipher, &nonce); let cipher_text = encryptor.encrypt_next(test_data.as_slice()).unwrap(); - dbg!(cipher_text.len()); - let a = decrypter.decrypt_next(cipher_text.as_slice()).unwrap(); - assert_eq!(test_data, a.as_slice()); + let plain = decryptor.decrypt_next(cipher_text.as_slice()).unwrap(); + assert_eq!(test_data.as_slice(), plain); } } diff --git a/proxy/src/serve_tasks.rs b/proxy/src/serve_tasks.rs index 6c01e10d..fb7f36d1 100644 --- a/proxy/src/serve_tasks.rs +++ b/proxy/src/serve_tasks.rs @@ -8,7 +8,7 @@ use axum::{ body::Bytes, extract::{FromRef, Request, State}, http::{header, request::Parts, HeaderMap, HeaderValue, StatusCode, Uri}, response::{sse::Event, IntoResponse, Response, Sse}, routing::{any, get, put}, Json, RequestExt, Router }; use futures::{ - stream::{StreamExt, TryStreamExt}, + stream::StreamExt, Stream, TryFutureExt, }; use httpdate::fmt_http_date; @@ -17,21 +17,21 @@ use serde::{de::DeserializeOwned, Deserialize, Serialize}; use serde_json::Value; use beam_lib::{AppId, AppOrProxyId, ProxyId}; use shared::{ - config::{self, CONFIG_PROXY}, config_proxy, config_shared::ConfigCrypto, crypto::{self, get_own_crypto_material, CryptoPublicPortion}, crypto_jwt, errors::SamplyBeamError, http_client::SamplyHttpClient, reqwest, sse_event::SseEventType, DecryptableMsg, EncryptableMsg, EncryptedMessage, EncryptedMsgTaskRequest, EncryptedMsgTaskResult, MessageType, Msg, MsgEmpty, MsgId, MsgSigned, MsgTaskRequest, MsgTaskResult, PlainMessage + DecryptableMsg, EncryptableMsg, EncryptedMessage, EncryptedMsgTaskRequest, EncryptedMsgTaskResult, MessageType, Msg, MsgEmpty, MsgId, MsgSigned, MsgTaskRequest, MsgTaskResult, PlainMessage, crypto::{self, CryptoPublicPortion}, crypto_jwt, errors::SamplyBeamError, format_to_without_broker, http_client::SamplyHttpClient, reqwest, sse_event::SseEventType }; -use tokio::io::BufReader; +use sse_stream::SseStream; +use tokio::{io::BufReader, task::id}; use tracing::{debug, error, info, trace, warn}; -use crate::{auth::AuthenticatedApp, PROXY_TIMEOUT}; +use crate::{auth::AuthenticatedApp, config::Config, PROXY_TIMEOUT}; #[derive(Clone, FromRef)] pub(crate) struct TasksState { pub(crate) client: SamplyHttpClient, - pub(crate) config: config_proxy::Config, + pub(crate) config: &'static Config, } -pub(crate) fn router(client: &SamplyHttpClient) -> Router { - let config = config::CONFIG_PROXY.clone(); +pub(crate) fn router(client: &SamplyHttpClient, config: &'static Config) -> Router { let state = TasksState { client: client.clone(), config, @@ -39,6 +39,7 @@ pub(crate) fn router(client: &SamplyHttpClient) -> Router { 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) @@ -62,7 +63,7 @@ const ERR_FAKED_FROM: (StatusCode, &str) = ( pub(crate) async fn forward_request( mut req: Request, - config: &config_proxy::Config, + config: &Config, sender: &AppId, client: &SamplyHttpClient, ) -> Result { @@ -83,7 +84,14 @@ pub(crate) async fn forward_request( HeaderValue::from_static(env!("SAMPLY_USER_AGENT")), ); let (encrypted_msg, parts) = encrypt_request(req, &sender).await?; - let req = sign_request(encrypted_msg, parts, &config, &get_own_crypto_material()).await.map_err(IntoResponse::into_response)?; + match &encrypted_msg { + MessageType::MsgTaskRequest(task) => info!(from = %sender.hide_broker_name(), to = %format_to_without_broker(&task.to), id = %task.id, "Sending task"), + MessageType::MsgTaskResult(result) => info!(from = %sender.hide_broker_name(), for = %result.task, "Submitting result"), + #[cfg(feature = "sockets")] + MessageType::MsgSocketRequest(socket_req) => info!(from = %socket_req.get_from().hide_broker(), to = %format_to_without_broker(&socket_req.get_to()), id = %socket_req.id, "Submitting socket request"), + MessageType::MsgEmpty(..) => {}, + }; + let req = sign_request(encrypted_msg, parts, &config).await.map_err(IntoResponse::into_response)?; trace!("Requesting: {:?}", req); let resp = client.execute(req).await.map_err(|e| { if e.is_timeout() { @@ -103,7 +111,7 @@ pub(crate) async fn forward_request( pub(crate) async fn handler_task( State(client): State, - State(config): State, + State(config): State<&'static Config>, AuthenticatedApp(sender): AuthenticatedApp, headers: HeaderMap, req: Request, @@ -131,7 +139,7 @@ pub(crate) async fn handler_task( async fn handler_tasks_nostream( client: SamplyHttpClient, - config: config_proxy::Config, + config: &Config, sender: AppId, req: Request, ) -> Result { @@ -152,7 +160,7 @@ async fn handler_tasks_nostream( // TODO: Always return application/jwt from server. if !bytes.is_empty() { if let Ok(json) = serde_json::from_slice::(&bytes) { - let json = to_server_error(validate_and_decrypt(json).await)?; + let json = to_server_error(validate_and_decrypt(json, config).await)?; trace!("Decrypted Msg: {:#?}", json); bytes = serde_json::to_vec(&json).unwrap().into(); trace!( @@ -177,7 +185,7 @@ async fn handler_tasks_nostream( async fn handler_tasks_stream( client: SamplyHttpClient, - config: config_proxy::Config, + config: &'static Config, sender: AppId, req: Request, ) -> Result>>, Response> { @@ -193,20 +201,12 @@ async fn handler_tasks_stream( } let outgoing = async_stream::stream! { - let incoming = resp - .bytes_stream() - .map(|result| result.map_err(|error| { - let kind = error.is_timeout().then_some(std::io::ErrorKind::TimedOut).unwrap_or(std::io::ErrorKind::Other); - std::io::Error::new(kind, format!("IO Error: {error}")) - })) - .into_async_read(); - - let mut reader = async_sse::decode(incoming); + let mut reader = SseStream::from_byte_stream(resp.bytes_stream()); while let Some(event) = reader.next().await { let event = match event { - Ok(event)=> event, - Err(e) if e.downcast_ref::().unwrap().kind() == std::io::ErrorKind::TimedOut => { + Ok(event) => event, + Err(e) if sse_error_is_timeout(&e) => { debug!("SSE connection timed out"); break; }, @@ -218,81 +218,78 @@ async fn handler_tasks_stream( continue; } }; - match event { - async_sse::Event::Retry(_dur) => { - error!("Got a retry message from the Broker, which is not yet supported."); + if event.retry.is_some() && event.event.is_none() && event.data.is_none() { + error!("Got a retry message from the Broker, which is not yet supported."); + continue; + } + // Check if this is a message or some control event + let event_type = SseEventType::from_str(event.event.as_deref().unwrap_or("")).expect("Error in Infallible"); + let mut event_as_bytes = event.data.unwrap_or_default().into_bytes(); + let event_as_str = std::str::from_utf8(&event_as_bytes).unwrap_or("(unable to parse)"); + + match &event_type { + SseEventType::DeletedTask | SseEventType::WaitExpired => { + debug!("SSE: Got {event_type} message, forwarding to App."); + yield Ok(Event::default() + .event(event_type) + .data(event_as_str)); + continue; }, - async_sse::Event::Message(event) => { - // Check if this is a message or some control event - let event_type = SseEventType::from_str(event.name()).expect("Error in Infallible"); - let mut event_as_bytes = event.into_bytes(); - let event_as_str = std::str::from_utf8(&event_as_bytes).unwrap_or("(unable to parse)"); - - match &event_type { - SseEventType::DeletedTask | SseEventType::WaitExpired => { - debug!("SSE: Got {event_type} message, forwarding to App."); - yield Ok(Event::default() - .event(event_type) - .data(event_as_str)); - continue; - }, - SseEventType::Error => { - warn!("SSE: The Broker has reported an error: {event_as_str}"); - yield Ok(Event::default() - .event(event_type) - .data(event_as_str)); - continue; - }, - SseEventType::Undefined => { - error!("SSE: Got a message without event type -- discarding."); - continue; - }, - SseEventType::Unknown(s) => { - error!("SSE: Got unknown event type: {s} -- discarding."); - continue; - }, - SseEventType::NewResult => { - debug!("SSE: Got new result"); - } - other => { - info!("Got \"{other}\" event -- parsing."); - } - } - - // Check reply's signature - - if !event_as_bytes.is_empty() { - let Ok(json) = serde_json::from_slice::(&event_as_bytes) else { - warn!("Answer is no valid JSON; discarding: \"{event_as_str}\"."); - // TODO: For some reason, compiler won't accept the following lines, so we can't inform the App about the problem. - // - // warn!("Answer is no valid JSON; returning as-is to client: \"{event_as_str}\"."); - // yield Ok(Event::default() - // .event(SseEventType::Error) - // .data(format!("Broker sent invalid JSON: {event_as_str}"))); - continue; - }; - let json = match validate_and_decrypt(json).await { - Ok(json) => json, - Err(err) => { - warn!("Got an error decrypting Broker's reply: {err}"); - continue; - } - }; - trace!("Decrypted Msg: {:#?}",json); - event_as_bytes = serde_json::to_vec(&json).unwrap(); - trace!( - "Validated and stripped signature: \"{}\"", - std::str::from_utf8(&event_as_bytes).unwrap_or("Unable to parse string as UTF-8") - ); - } - let as_string = std::str::from_utf8(&event_as_bytes).unwrap_or("(garbled_utf8)"); - let event = Event::default() + SseEventType::Error => { + warn!("SSE: The Broker has reported an error: {event_as_str}"); + yield Ok(Event::default() .event(event_type) - .data(as_string); - yield Ok(event); + .data(event_as_str)); + continue; + }, + SseEventType::Undefined => { + error!("SSE: Got a message without event type -- discarding."); + continue; + }, + SseEventType::Unknown(s) => { + error!("SSE: Got unknown event type: {s} -- discarding."); + continue; + }, + SseEventType::NewResult => { + debug!("SSE: Got new result"); } + other => { + info!("Got \"{other}\" event -- parsing."); + } + } + + // Check reply's signature + + if !event_as_bytes.is_empty() { + let Ok(json) = serde_json::from_slice::(&event_as_bytes) else { + warn!("Answer is no valid JSON; discarding: \"{event_as_str}\"."); + // TODO: For some reason, compiler won't accept the following lines, so we can't inform the App about the problem. + // + // warn!("Answer is no valid JSON; returning as-is to client: \"{event_as_str}\"."); + // yield Ok(Event::default() + // .event(SseEventType::Error) + // .data(format!("Broker sent invalid JSON: {event_as_str}"))); + continue; + }; + let json = match validate_and_decrypt(json, config).await { + Ok(json) => json, + Err(err) => { + warn!("Got an error decrypting Broker's reply: {err}"); + continue; + } + }; + trace!("Decrypted Msg: {:#?}",json); + event_as_bytes = serde_json::to_vec(&json).unwrap(); + trace!( + "Validated and stripped signature: \"{}\"", + std::str::from_utf8(&event_as_bytes).unwrap_or("Unable to parse string as UTF-8") + ); } + let as_string = std::str::from_utf8(&event_as_bytes).unwrap_or("(garbled_utf8)"); + let event = Event::default() + .event(event_type) + .data(as_string); + yield Ok(event); } }; // TODO: Somehow return correct error code (not always possible since headers are sent before long request) @@ -300,6 +297,15 @@ async fn handler_tasks_stream( Ok(sse) } +pub(crate) fn sse_error_is_timeout(err: &sse_stream::Error) -> bool { + match err { + sse_stream::Error::Body(err) => err + .downcast_ref::() + .is_some_and(reqwest::Error::is_timeout), + _ => false, + } +} + pub(crate) fn to_server_error(res: Result) -> Result { res.map_err(|e| match e { SamplyBeamError::JsonParseError(e) => { @@ -322,12 +328,11 @@ pub(crate) fn to_server_error(res: Result) -> Result Result { let from = body.get_from(); - let token_without_extended_signature = crypto_jwt::sign_to_jwt(&body, &private_key.privkey_rs256) + let token_without_extended_signature = crypto_jwt::sign_to_jwt(&body, &config.crypto.privkey_rs256) .await .map_err(|e| { error!("Crypto failed: {}", e); @@ -351,7 +356,7 @@ pub async fn sign_request( let digest = crypto_jwt::make_extra_fields_digest(&parts.method, &parts.uri, &headers_mut, sig, &from) .map_err(|_| ERR_INTERNALCRYPTO)?; - let token_with_extended_signature = crypto_jwt::sign_to_jwt(&digest, &private_key.privkey_rs256) + let token_with_extended_signature = crypto_jwt::sign_to_jwt(&digest, &config.crypto.privkey_rs256) .await .map_err(|e| { error!("Crypto failed: {}", e); @@ -377,7 +382,7 @@ pub async fn sign_request( } // This requires rustc 1.77 -pub(crate) async fn validate_and_decrypt(json: Value) -> Result { +pub(crate) async fn validate_and_decrypt(json: Value, config: &Config) -> Result { // It might be possible to use MsgSigned directly instead but there are issues impl Deserialize for MsgSigned #[derive(Deserialize)] struct MsgSignedHelper { @@ -386,7 +391,7 @@ pub(crate) async fn validate_and_decrypt(json: Value) -> Result Result::verify(&signed.jwt) .await? .msg; - Ok(serde_json::to_value(decrypt_msg(msg)?).expect("Should serialize fine")) + match &msg { + MessageType::MsgTaskRequest(task) => info!(from = %task.get_from().hide_broker(), id = %task.id, "New task"), + MessageType::MsgTaskResult(result) => info!(from = %result.get_from().hide_broker(), for = %result.task, "New result"), + #[cfg(feature = "sockets")] + MessageType::MsgSocketRequest(socket_req) => info!(from = %socket_req.get_from().hide_broker(), id = %socket_req.id, "New socket request"), + MessageType::MsgEmpty(..) => {}, + }; + Ok(serde_json::to_value(decrypt_msg(msg, config)?).expect("Should serialize fine")) } Err(e) => Err(SamplyBeamError::JsonParseError(format!( "Failed to parse broker response as a signed encrypted message. Err is {e}" @@ -408,10 +420,10 @@ pub(crate) async fn validate_and_decrypt(json: Value) -> Result(msg: M) -> Result { +fn decrypt_msg(msg: M, config: &Config) -> Result { msg.decrypt( - &AppOrProxyId::Proxy(CONFIG_PROXY.proxy_id.to_owned()), - &crypto::get_own_crypto_material().privkey_rsa, + &AppOrProxyId::Proxy(config.proxy_id.to_owned()), + &config.crypto.privkey_rsa, ) } diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 445bd721..5a0bad06 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -15,25 +15,27 @@ uuid = { version = "1", features = [ ]} serde = { version = "1", features = ["derive"] } serde_json = "1" +clap.workspace = true tokio = { version = "1", features = ["full"] } axum = { version = "0.8", features = [] } bytes = "1.4" +futures.workspace = true -# HTTP client with proxy support -reqwest = { version = "0.12", features = ["stream"] } +# This includes all default features of reqwest but uses native-tls aka openssl as we depend on it for encryption anyways +reqwest = { workspace = true, features = ["stream", "native-tls", "charset", "system-proxy", "http2"] } # Logging tracing = "0.1" -tracing-subscriber = { version = "0.3", features = ["env-filter"] } +tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } # Crypto -rand = "0.8" -rsa = "0.9" -sha2 = "0.10" +rand.workspace = true +rsa.workspace = true +sha2 = "0.11" +chacha20poly1305.workspace = true openssl = "0.10" -chacha20poly1305 = "0.10" -itertools = "0.14.0" +itertools = "0.15.0" jwt-simple = "0.11" # Global variables @@ -42,21 +44,17 @@ once_cell = "1" # Error handling thiserror = "2" -# Command Line Interface -clap = { version = "4", features = ["env", "derive"] } fundu = "2.0" -regex = "1" # expire map dependencies dashmap = { version = "6.0", optional = true} beam-lib = { workspace = true } async-trait = "0.1" +tracing-appender = "0.2.3" [features] expire_map = ["dep:dashmap"] sockets = ["expire_map", "beam-lib/sockets", "reqwest/json"] default = [] -config-for-proxy = [] -config-for-central = [] diff --git a/shared/src/config.rs b/shared/src/config.rs deleted file mode 100644 index bb3e2b7f..00000000 --- a/shared/src/config.rs +++ /dev/null @@ -1,38 +0,0 @@ -use once_cell::sync::{Lazy, OnceCell}; -use tracing::debug; - -use crate::{ - config_broker, config_proxy, - config_shared::{self, ConfigCrypto}, - crypto, - errors::SamplyBeamError, -}; - -pub(crate) trait Config: Sized { - fn load() -> Result; -} - -fn load() -> T where { - T::load() - .unwrap_or_else(|e| { - eprintln!("Unable to start as there was an error reading the config:\n{}\n\nTerminating -- please double-check your startup parameters with --help and refer to the documentation.", e); - std::process::exit(1); - }) -} - -pub static CONFIG_PROXY: Lazy = Lazy::new(|| { - debug!("Loading config CONFIG_PROXY"); - load() -}); - -pub static CONFIG_CENTRAL: Lazy = Lazy::new(|| { - debug!("Loading config CONFIG_CENTRAL"); - load() -}); - -pub static CONFIG_SHARED: Lazy = Lazy::new(|| { - debug!("Loading config CONFIG_SHARED"); - load() -}); - -pub(crate) static CONFIG_SHARED_CRYPTO: OnceCell = OnceCell::new(); diff --git a/shared/src/config_shared.rs b/shared/src/config_shared.rs deleted file mode 100644 index b1b652f8..00000000 --- a/shared/src/config_shared.rs +++ /dev/null @@ -1,225 +0,0 @@ -use beam_lib::ProxyId; -use reqwest::{Certificate, Url}; -use crate::{ - config::CONFIG_SHARED_CRYPTO, - crypto::{ - self, get_all_certs_and_clients_by_cname_as_pemstr, load_certificates_from_dir, - CryptoPublicPortion, GetCerts, ProxyCertInfo, - }, - SamplyBeamError, -}; -use clap::Parser; -use jwt_simple::prelude::RS256KeyPair; -use openssl::{ - asn1::Asn1IntegerRef, - x509::{self, X509}, -}; -use rsa::{pkcs1::DecodeRsaPrivateKey, pkcs8::DecodePrivateKey, RsaPrivateKey}; -use std::{fs::read_to_string, path::PathBuf, rc::Rc, sync::Arc}; -use tracing::{debug, info}; - -pub(crate) const CLAP_FOOTER: &str = "For proxy support, environment variables HTTP_PROXY, HTTPS_PROXY, ALL_PROXY and NO_PROXY (and their lower-case variants) are supported. Usually, you want to set HTTP_PROXY *and* HTTPS_PROXY or set ALL_PROXY if both values are the same.\n\nFor updates and detailed usage instructions, visit https://github.com/samply/beam"; - -#[derive(Parser, Debug)] -#[clap( - name("🌈 Samply.Beam (shared library)"), - version, - arg_required_else_help(true), - after_help(crate::config_shared::CLAP_FOOTER) -)] -struct CliArgs { - /// Outgoing HTTP proxy: Directory with CA certificates to trust for TLS connections (e.g. /etc/samply/cacerts/) - #[clap(long, env, value_parser)] - tls_ca_certificates_dir: Option, - - /// samply.pki: Path to own secret key - #[clap(long, env, value_parser, default_value = "/run/secrets/privkey.pem")] - privkey_file: PathBuf, - - /// samply.pki: Path to CA Root certificate - #[clap(long, env, value_parser, default_value = "/run/secrets/root.crt.pem")] - rootcert_file: PathBuf, - - // TODO: The following arguments have been added for compatibility reasons with the proxy config. Find another way to merge configs. - /// (included for technical reasons) - #[clap(long, env, value_parser)] - broker_url: Url, - - /// (included for technical reasons) - #[clap(long, env, value_parser)] - proxy_id: Option, - - /// (included for technical reasons) - #[clap(action)] - examples: Option, - - /// (included for technical reasons) - #[clap(long, hide(true))] - test_threads: Option, -} - -#[allow(dead_code)] -pub struct Config { - pub(crate) tls_ca_certificates_dir: Option, - pub broker_domain: String, - pub root_cert: X509, - pub tls_ca_certificates: Vec, -} - -#[derive(Debug, Clone)] -pub struct ConfigCrypto { - pub privkey_rs256: RS256KeyPair, - pub privkey_rsa: RsaPrivateKey, -} - -impl crate::config::Config for Config { - fn load() -> Result { - let cli_args = CliArgs::parse(); - beam_lib::set_broker_id(cli_args.broker_url.host().unwrap().to_string()); - - let root_cert = crypto::load_certificates_from_file(cli_args.rootcert_file)?; - let broker_domain = cli_args.broker_url.host(); - if false { - todo!() // TODO Tobias: Check if matches certificate, and fail - } - let broker_domain = broker_domain.unwrap().to_string(); - let tls_ca_certificates_dir = cli_args.tls_ca_certificates_dir; - let tls_ca_certificates = crate::crypto::load_certificates_from_dir( - tls_ca_certificates_dir.clone(), - ) - .map_err(|e| { - SamplyBeamError::ConfigurationFailed(format!( - "Unable to read from TLS CA directory: {}", - e - )) - })?; - Ok(Config { - broker_domain, - tls_ca_certificates_dir, - root_cert, - tls_ca_certificates, - }) - } -} - -fn get_enrollment_msg(proxy_id: &Option) -> String { - let divider = "***************************************************************************\n - *** Beam Certificate Enrollment Warning ***\n - ***************************************************************************"; - format!( - "{}\nIf you are not yet enrolled in the central certificate store, please execute the beam-enroll companion tool (https://github.com/samply/beam-enroll) by executing:\n docker run --rm -it -v \"$(pwd)\":/data -e PROXY_ID={} samply/beam-enroll\nand follow the steps on the screen.\nAfter your certificate signing request (CSR) has been approved, please restart this Beam.Proxy and this message should disappear.", - divider, - proxy_id.as_deref().unwrap_or("") - ) -} - -pub async fn init_public_crypto_for_proxy( - mut config: ConfigCrypto, -) -> Result { - let cli_args = CliArgs::parse(); - let public_info = load_public_crypto_for_proxy(&cli_args, &mut config).await?; - - let cert_info = crypto::ProxyCertInfo::try_from(&public_info.cert)?; - if CONFIG_SHARED_CRYPTO.set(config).is_err() { - panic!("Tried to initialize crypto twice (init_public_crypto_for_proxy())"); - } - Ok(cert_info) -} - -pub fn load_private_crypto_for_proxy() -> Result { - let cli_args = CliArgs::parse(); - let privkey_pem = read_to_string(&cli_args.privkey_file) - .map_err(|e| { - SamplyBeamError::ConfigurationFailed(format!( - "Unable to load private key from file {}: {}\n{}", - cli_args.privkey_file.to_string_lossy(), - e, - get_enrollment_msg(&cli_args.proxy_id) - )) - })? - .trim() - .to_string(); - let privkey_rsa = RsaPrivateKey::from_pkcs1_pem(&privkey_pem) - .or_else(|_| RsaPrivateKey::from_pkcs8_pem(&privkey_pem)) - .map_err(|e| { - SamplyBeamError::ConfigurationFailed(format!( - "Unable to interpret private key PEM as PKCS#1 or PKCS#8: {}", - e - )) - })?; - let privkey_rs256 = RS256KeyPair::from_pem(&privkey_pem).map_err(|e| { - SamplyBeamError::ConfigurationFailed(format!( - "Unable to interpret private key PEM as PKCS#1 or PKCS#8: {}", - e - )) - })?; - Ok(ConfigCrypto { - privkey_rs256, - privkey_rsa, - }) -} - -async fn load_public_crypto_for_proxy( - cli_args: &CliArgs, - config: &mut ConfigCrypto, -) -> Result { - let proxy_id = cli_args.proxy_id.as_ref() - .expect("load_crypto() has been called without setting a Proxy ID (maybe in broker?). This should not happen."); - let proxy_id = ProxyId::new(proxy_id)?; - let publics: Vec = get_all_certs_and_clients_by_cname_as_pemstr(&proxy_id) - .await - .into_iter() - .filter_map(|r| { - r.map_err(|e| debug!("Unable to parse Certificate: {e}")) - .ok() - }) - .collect(); - let public = crypto::get_best_own_certificate(publics, &config.privkey_rsa).ok_or( - SamplyBeamError::SignEncryptError( - "Unable to choose valid, newest certificate for this proxy".into(), - ), - )?; - let serial = asn_str_to_vault_str(public.cert.serial_number())?; - config.privkey_rs256 = config.privkey_rs256.clone().with_key_id(&serial); - Ok(public) -} - -fn asn_str_to_vault_str(asn: &Asn1IntegerRef) -> Result { - let mut a = asn - .to_bn() - .map_err(|e| { - SamplyBeamError::SignEncryptError(format!("Unable to parse your certificate: {}", e)) - })? - .to_hex_str() - .map_err(|e| { - SamplyBeamError::SignEncryptError(format!("Unable to parse your certificate: {}", e)) - })? - .to_string() - .to_ascii_lowercase(); - - let mut i = 2; - while i < a.len() { - a.insert(i, ':'); - i += 3; - } - - Ok(a) -} - -#[cfg(test)] -mod test { - use openssl::{ - asn1::{Asn1Integer, Asn1String, Asn1StringRef}, - bn::BigNum, - }; - - use super::asn_str_to_vault_str; - - #[test] - fn hex_str() { - let bn = BigNum::from_hex_str("440E0D94F36966391117BC9F867D84F0C48CFCB7").unwrap(); - let input = Asn1Integer::from_bn(&bn).unwrap(); - let expected = "44:0e:0d:94:f3:69:66:39:11:17:bc:9f:86:7d:84:f0:c4:8c:fc:b7"; - assert_eq!(expected, asn_str_to_vault_str(&input).unwrap()); - } -} diff --git a/shared/src/crypto.rs b/shared/src/crypto.rs index 3ce3b487..2dca8de7 100644 --- a/shared/src/crypto.rs +++ b/shared/src/crypto.rs @@ -1,15 +1,14 @@ use async_trait::async_trait; use axum::{body::Body, http::Request, Json}; -use clap::error; use itertools::Itertools; use once_cell::sync::{Lazy, OnceCell}; use openssl::{ - asn1::{Asn1Time, Asn1TimeRef, Asn1Integer}, + asn1::{Asn1Integer, Asn1IntegerRef, Asn1Time, Asn1TimeRef}, error::ErrorStack, rand::rand_bytes, string::OpensslString, - x509::{X509, X509Crl, CrlStatus}, + x509::{CrlStatus, X509Crl, X509}, }; use rsa::{ pkcs1::DecodeRsaPublicKey, pkcs8::DecodePublicKey, RsaPrivateKey, RsaPublicKey, traits::PublicKeyParts, @@ -29,8 +28,6 @@ use tracing::{debug, error, info, warn}; use beam_lib::{AppOrProxyId, ProxyId}; use crate::{ - config, - config_shared::ConfigCrypto, crypto, errors::{CertificateInvalidReason, SamplyBeamError}, EncryptedMsgTaskRequest, MsgTaskRequest, @@ -285,9 +282,15 @@ impl CertificateCache { revoked_certs } - pub async fn update_certificates_mut(&mut self) -> Result { + pub async fn update_certificates_mut( + &mut self, + ) -> Result { debug!("Updating certificates via network ..."); - let certificate_list = CERT_GETTER.get().unwrap().certificate_list_via_network().await?; + let certificate_list = CERT_GETTER + .get() + .unwrap() + .certificate_list_via_network() + .await?; let certificate_revocation_list = CERT_GETTER.get().unwrap().get_crl().await?; // Check if any of the certs in the cache have been revoked let mut revoked_certs = certificate_revocation_list @@ -306,32 +309,28 @@ impl CertificateCache { ); let mut new_count = 0; - //TODO Check for validity - for serial in new_certificate_serials { + let cert_getter = CERT_GETTER.get().unwrap(); + let cert_pems = new_certificate_serials + .iter() + .map(|s| cert_getter.certificate_by_serial_as_pem(s)) + .collect::>() + .await; + for (serial, cert_pem) in new_certificate_serials.into_iter().zip(cert_pems) { debug!("Checking certificate with serial {serial}"); - let certificate = CERT_GETTER - .get() - .unwrap() - .certificate_by_serial_as_pem(serial) - .await; - if let Err(e) = certificate { - match e { - SamplyBeamError::CertificateError(err) => { - debug!("Will skip invalid certificate {serial} from now on."); - self.serial_to_x509 - .insert(serial.clone(), CertificateCacheEntry::Invalid(err)); - } - other_error => { - warn!( - "Could not retrieve certificate for serial {serial}: {}", - other_error - ); - } - }; - continue; - } - let certificate = certificate.unwrap(); + let certificate = match cert_pem { + Err(SamplyBeamError::CertificateError(err)) => { + debug!("Will skip invalid certificate {serial} from now on."); + self.serial_to_x509 + .insert(serial.clone(), CertificateCacheEntry::Invalid(err)); + continue; + }, + Err(other_error) => { + warn!("Could not retrieve certificate for serial {serial}: {other_error}"); + continue; + } + Ok(cert) => cert, + }; let opensslcert = match X509::from_pem(certificate.as_bytes()) { Ok(x) => x, Err(err) => { @@ -504,9 +503,9 @@ async fn get_all_certs_from_cache_by_cname(cname: &ProxyId) -> Vec Result<(), SamplyBeamError> { +pub async fn init_ca_chain(root_cert: &X509) -> Result<(), SamplyBeamError> { let mut cache = CERT_CACHE.write().await; - cache.set_root_cert(&config::CONFIG_SHARED.root_cert); + cache.set_root_cert(root_cert); cache.set_im_cert().await?; Ok(()) } @@ -707,13 +706,10 @@ pub(crate) fn hash(data: &[u8]) -> Result<[u8; 32], SamplyBeamError> { Ok(digest) } -pub fn get_own_crypto_material() -> &'static ConfigCrypto { - config::CONFIG_SHARED_CRYPTO.get().unwrap() -} /* Utility Functions */ /// Extracts the pem-encoded public key from a x509 certificate -fn x509_cert_to_x509_public_key(cert: &X509) -> Result, SamplyBeamError> { +pub fn x509_cert_to_x509_public_key(cert: &X509) -> Result, SamplyBeamError> { match cert.public_key() { Ok(key) => key.public_key_to_pem().or_else(|_| { Err(SamplyBeamError::SignEncryptError( @@ -755,6 +751,28 @@ pub fn asn1_time_to_system_time(time: &Asn1TimeRef) -> Result Result { + let mut a = asn + .to_bn() + .map_err(|e| { + SamplyBeamError::SignEncryptError(format!("Unable to parse your certificate: {}", e)) + })? + .to_hex_str() + .map_err(|e| { + SamplyBeamError::SignEncryptError(format!("Unable to parse your certificate: {}", e)) + })? + .to_string() + .to_ascii_lowercase(); + + let mut i = 2; + while i < a.len() { + a.insert(i, ':'); + i += 3; + } + + Ok(a) +} + /// Checks if SystemTime::now() is between the not_before and the not_after dates of a x509 certificate pub fn x509_date_valid(cert: &X509) -> Result { let expirydate = asn1_time_to_system_time(cert.not_after())?; @@ -807,10 +825,8 @@ pub fn load_certificates_from_dir(ca_dir: Option) -> Result Result { let cert_rsa = cert.public_key()?.rsa()?; - let cert_mod = cert_rsa.n(); - let key_mod = key.n(); - let key_mod_bignum = openssl::bn::BigNum::from_slice(&key_mod.to_bytes_be())?; - let is_equal = cert_mod.ucmp(&key_mod_bignum) == std::cmp::Ordering::Equal; + let cert_mod = rsa::BoxedUint::from_be_slice_vartime(&cert_rsa.n().to_vec()); + let is_equal = cert_mod.cmp(&key.n()) == std::cmp::Ordering::Equal; if !is_equal { match ProxyCertInfo::try_from(cert) { Ok(x) => { @@ -857,7 +873,7 @@ pub fn get_newest_cert(certs: &mut Vec) -> Option>, private_rsa: &RsaPrivateKey, ) -> Option { @@ -920,6 +936,14 @@ mod tests { use super::*; + #[test] + fn hex_str() { + let bn = BigNum::from_hex_str("440E0D94F36966391117BC9F867D84F0C48CFCB7").unwrap(); + let input = Asn1Integer::from_bn(&bn).unwrap(); + let expected = "44:0e:0d:94:f3:69:66:39:11:17:bc:9f:86:7d:84:f0:c4:8c:fc:b7"; + assert_eq!(expected, asn_str_to_vault_str(&input).unwrap()); + } + fn build_x509(ttl: Duration) -> X509 { let mut builder = X509::builder().unwrap(); let duration = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap() + ttl; diff --git a/shared/src/crypto_jwt.rs b/shared/src/crypto_jwt.rs index 8b60c93d..25e8b139 100644 --- a/shared/src/crypto_jwt.rs +++ b/shared/src/crypto_jwt.rs @@ -2,8 +2,6 @@ use std::net::{SocketAddr, IpAddr}; use beam_lib::{AppOrProxyId, ProxyId}; use crate::{ - config, - config_shared::ConfigCrypto, crypto::{self, CryptoPublicPortion}, errors::{CertificateInvalidReason, SamplyBeamError}, Msg, MsgEmpty, MsgId, MsgSigned, diff --git a/shared/src/examples.rs b/shared/src/examples.rs deleted file mode 100644 index 7bf84bb3..00000000 --- a/shared/src/examples.rs +++ /dev/null @@ -1,120 +0,0 @@ -use std::collections::HashMap; - -use rand::Rng; -use serde_json::json; - -use beam_lib::{AppId, ProxyId}; -use crate::{ - config, FailureStrategy, MsgId, MsgSigned, MsgTaskRequest, MsgTaskResult, -}; - -type BrokerId = String; - -#[cfg(debug_assertions)] -pub fn print_example_objects() -> bool { - if std::env::args().nth(1).unwrap_or_default() == "examples" { - let broker_id = std::env::args().nth(2); - let proxy_id = match std::env::args().nth(3) { - Some(id) => ProxyId::new(&id).ok(), - None => None, - }; - let (tasks, results) = generate_example_tasks(broker_id, proxy_id); - for (num, task) in tasks.iter().enumerate() { - println!( - "export TASK{}='{}'", - num, - serde_json::to_string(task).unwrap().replace('\'', "\'") - ); - } - for (num, result) in results.iter().enumerate() { - println!( - "export RESULT{}='{}'", - num, - serde_json::to_string(&result).unwrap().replace('\'', "\'") - ); - } - true - } else { - false - } -} - -pub fn generate_example_tasks( - broker: Option, - proxy: Option, -) -> (Vec, Vec) { - let broker = - broker.unwrap_or(config::CONFIG_SHARED.broker_domain.clone()); - let proxy = { - if let Some(id) = proxy { - if !id.can_be_signed_by(&broker) { - panic!( - "Submitted proxy_id ({id}) cannot be signed by submitted broker_id ({broker})" - ); - } - id - } else { - ProxyId::new(&format!("proxy{}.{}", 23, broker)).unwrap() - } - }; - let app1 = AppId::new(&format!("app1.{proxy}")).unwrap(); - let app2 = AppId::new(&format!("app2.{proxy}")).unwrap(); - - let task_for_apps_1_2 = MsgTaskRequest::new( - app1.clone().into(), - vec![app1.clone().into(), app2.clone().into()], - "My important task".into(), - FailureStrategy::Retry { - backoff_millisecs: 1000, - max_tries: 5, - }, - json!(["The", "Broker", "can", "read", "and", "filter", "this"]), - ); - - let response_by_app1 = MsgTaskResult { - from: app1.clone().into(), - to: vec![app1.clone().into()], - task: task_for_apps_1_2.id, - status: beam_lib::WorkStatus::Succeeded, - body: "All done!".into(), - metadata: json!("A normal string works, too!"), - }; - let response_by_app2 = MsgTaskResult { - from: app2.into(), - to: vec![app1.into()], - task: task_for_apps_1_2.id, - status: beam_lib::WorkStatus::PermFailed, - body: "Unable to complete".into(), - metadata: json!({ "I": { "like": [ "results", "cake" ] } }), - }; - let mut tasks = Vec::new(); - for task in [task_for_apps_1_2] { - tasks.push(task); - } - let mut results = Vec::new(); - for result in [response_by_app1, response_by_app2] { - results.push(result); - } - (tasks, results) -} - -// Random - - -fn random_str() -> String { - const LENGTH: u8 = 8; - const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789"; - let mut rng = rand::thread_rng(); - let random_str: String = (0..=LENGTH) - .map(|_| { - let idx = rng.gen_range(0..CHARSET.len()); - CHARSET[idx] as char - }) - .collect(); - random_str -} - -fn random_number() -> u8 { - let mut rng = rand::thread_rng(); - rng.gen_range(u8::MIN..u8::MAX) -} diff --git a/shared/src/http_client.rs b/shared/src/http_client.rs index 355eedab..d9f83e6c 100644 --- a/shared/src/http_client.rs +++ b/shared/src/http_client.rs @@ -7,15 +7,15 @@ use openssl::x509::X509; use reqwest::{Certificate, Client, ClientBuilder}; use tracing::{debug, info, warn}; -use crate::{config, errors::SamplyBeamError}; +use crate::{errors::SamplyBeamError}; pub type SamplyHttpClient = reqwest::Client; -pub fn build( +pub fn builder( ca_certificates: &Vec, timeout: Option, keepalive: Option, -) -> Result { +) -> ClientBuilder { let mut builder = Client::builder().tcp_keepalive(keepalive); if let Some(to) = timeout { builder = builder.connect_timeout(to); @@ -43,7 +43,7 @@ pub fn build( }; info!("Using {proxies} and {certs} for TLS termination."); - builder.build().map_err(|e| SamplyBeamError::ConfigurationFailed(e.to_string())) + builder } #[cfg(test)] @@ -60,13 +60,13 @@ mod test { #[tokio::test] async fn https() { - let client = http_client::build(&vec![], None, None).unwrap(); + let client = http_client::builder(&vec![], None, None).build().unwrap(); run(HTTPS.parse().unwrap(), client).await; } #[tokio::test] async fn http() { - let client = http_client::build(&vec![], None, None).unwrap(); + let client = http_client::builder(&vec![], None, None).build().unwrap(); run(HTTP.parse().unwrap(), client).await; } diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 283d1a96..8e1a1584 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -2,8 +2,7 @@ use beam_lib::{AppId, AppOrProxyId, ProxyId, FailureStrategy, WorkStatus}; use chacha20poly1305::{ - aead::{Aead, AeadCore, KeyInit, OsRng}, - XChaCha20Poly1305, XNonce, + aead, XChaCha20Poly1305, XNonce, aead::{Aead, AeadCore, Generate, KeyInit, array::typenum::Unsigned} }; use crypto_jwt::extract_jwt; use errors::SamplyBeamError; @@ -21,7 +20,7 @@ use std::{ time::{Duration, Instant, SystemTime}, net::SocketAddr, error::Error, }; -use rand::Rng; +use rand::{rng, Rng}; use serde::{ de::{DeserializeOwned, Visitor}, Deserialize, Serialize, @@ -32,6 +31,7 @@ use uuid::Uuid; use crate::{crypto_jwt::JWT_VERIFICATION_OPTIONS, serde_helpers::*}; // Reexport b64 implementation pub use jwt_simple::reexports::ct_codecs; +pub use jwt_simple; pub use reqwest; pub use async_trait::async_trait; @@ -48,12 +48,6 @@ mod traits; #[cfg(test)] mod serializing_compatibility_test; -pub mod config; -pub mod config_shared; -// #[cfg(feature = "config-for-broker")] -pub mod config_broker; -// #[cfg(feature = "config-for-proxy")] -pub mod config_proxy; #[cfg(feature = "expire_map")] pub mod expire_map; #[cfg(feature = "sockets")] @@ -65,13 +59,12 @@ pub mod graceful_shutdown; pub mod http_client; pub mod middleware; -pub mod examples; - pub mod sse_event; // Reexports pub use openssl; +pub const CLAP_FOOTER: &str = "For proxy support, environment variables HTTP_PROXY, HTTPS_PROXY, ALL_PROXY and NO_PROXY (and their lower-case variants) are supported. Usually, you want to set HTTP_PROXY *and* HTTPS_PROXY or set ALL_PROXY if both values are the same.\n\nFor updates and detailed usage instructions, visit https://github.com/samply/beam"; #[derive(Serialize, Deserialize, Debug, Clone, Copy)] pub struct HowLongToBlock { @@ -234,6 +227,7 @@ pub trait DecryptableMsg: Msg + Serialize + Sized { my_id: &AppOrProxyId, my_priv_key: &RsaPrivateKey, ) -> Result { + let Some(Encrypted { encrypted, encryption_keys, @@ -242,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| { @@ -255,25 +249,39 @@ 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("".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 - let cipher_engine = XChaCha20Poly1305::new_from_slice(&my_priv_key.decrypt( - Oaep::new::(), + let symmetric_key = my_priv_key.decrypt( + Oaep::::new(), &encrypted_decryption_key, - )?) - .map_err(|e| { - SamplyBeamError::SignEncryptError(format!( - "Decryption error: Cannot initialize stream cipher because {}", - e - )) - })?; - let nonce: XNonce = XNonce::clone_from_slice(&encrypted[0..24]); - let ciphertext = &encrypted[24..]; + )?; + let symmetric_key = aead::Key::::try_from(symmetric_key.as_slice()) + .map_err(|_| { + SamplyBeamError::SignEncryptError( + "Decryption error: Invalid symmetric key length".into(), + ) + })?; + let cipher_engine = XChaCha20Poly1305::new(&symmetric_key); + const NONCE_SIZE: usize = ::NonceSize::USIZE; + let nonce = encrypted + .get(..NONCE_SIZE) + .and_then(|nonce| XNonce::try_from(nonce).ok()) + .ok_or(SamplyBeamError::SignEncryptError( + "Decryption error: Missing or invalid nonce".into(), + ))?; + let ciphertext = encrypted.get(NONCE_SIZE..).ok_or(SamplyBeamError::SignEncryptError( + "Decryption error: Missing ciphertext".into(), + ))?; let plaintext = String::from_utf8( cipher_engine .decrypt(&nonce, ciphertext.as_ref()) @@ -308,9 +316,9 @@ pub trait EncryptableMsg: Msg + Serialize + Sized { receivers_public_keys: &Vec, ) -> Result { // Generate Symmetric Key and Nonce - let mut rng = rand::thread_rng(); - let symmetric_key = XChaCha20Poly1305::generate_key(&mut rng); - let nonce = XChaCha20Poly1305::generate_nonce(&mut rng); + let mut rng = rng(); + let symmetric_key = aead::Key::::generate_from_rng(&mut rng); + let nonce = aead::Nonce::::generate_from_rng(&mut rng); // Encrypt symmetric key with receivers' public keys let Ok(encrypted_keys) = receivers_public_keys @@ -318,7 +326,7 @@ pub trait EncryptableMsg: Msg + Serialize + Sized { .map(|key| { key.encrypt( &mut rng, - Oaep::new::(), + Oaep::::new(), symmetric_key.as_slice(), ) }) @@ -722,6 +730,20 @@ impl Msg for MsgPing { } } +pub fn format_to_without_broker(to: &[AppOrProxyId]) -> impl Display + use<'_> { + struct Helper<'a> { + to: &'a [AppOrProxyId], + } + + impl<'a> std::fmt::Display for Helper<'a> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_list().entries(self.to.iter().map(|to| to.hide_broker())).finish() + } + } + + Helper { to } +} + pub fn try_read(map: &HashMap, key: &str) -> Option where T: FromStr, @@ -759,7 +781,7 @@ mod tests { }; //Setup Keypairs - 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) .expect("Failed to generate private key for proxy 1"); @@ -805,7 +827,7 @@ mod tests { }; //Setup Keypairs - 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) .expect("Failed to generate private key for proxy 1"); @@ -833,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 body + let as_creator = msg_encr + .clone() + .decrypt(&p1_id, &p1_private) + .expect("Creator must receive the task with body"); + 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()); + } } diff --git a/shared/src/logger.rs b/shared/src/logger.rs index ec983d7a..435a3935 100644 --- a/shared/src/logger.rs +++ b/shared/src/logger.rs @@ -1,36 +1,44 @@ +use std::{io, path::PathBuf}; + use tracing::{debug, dispatcher::SetGlobalDefaultError, Level}; -use tracing_subscriber::fmt::format::{debug_fn, self}; +use tracing_appender::{non_blocking::WorkerGuard, rolling::{InitError, Rotation}}; +use tracing_subscriber::{fmt::{self, format::{self, debug_fn}}, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; + +#[derive(Debug, clap::Args)] +pub struct LogOptions { + #[clap(long, env)] + /// Directory to store log files in. + pub log_dir: Option, -#[allow(clippy::if_same_then_else)] // The redundant if-else serves documentation purposes -pub fn init_logger() -> Result<(), SetGlobalDefaultError> { - let subscriber = tracing_subscriber::FmtSubscriber::builder() - .fmt_fields(debug_fn(|w, f, v| match f.name() { - "from" | "message" => write!(w, "{v:?}"), - _ => write!(w, "{f}={v:?} "), - })) - .with_max_level(Level::DEBUG); + /// Number of days to retain log files. + #[clap(long, env, default_value_t = 30)] + pub log_retention: usize, +} - // TODO: Reduce code complexity. - let env_filter = match std::env::var("RUST_LOG") { - Ok(env) if !env.is_empty() => { - if env.contains("hyper=") { - env - } else { - format!("{env},hyper=info") - } - } - _ => { - if cfg!(debug_assertions) { - "info,hyper=info".to_string() - } else { - "info,hyper=info".to_string() - } - } +pub fn init_logger(log_opts: &LogOptions) -> Result, InitError> { + let env_filter = EnvFilter::builder() + .with_default_directive(Level::INFO.into()) + .from_env_lossy(); + let registry = tracing_subscriber::registry() + .with(env_filter); + let (registry, guard) = if let LogOptions { log_dir: Some(log_dir), log_retention } = log_opts { + let appender = tracing_appender::rolling::Builder::new() + .max_log_files(*log_retention) + .filename_suffix("log") + .rotation(Rotation::DAILY) + .build(log_dir)?; + let (appender, guard) = tracing_appender::non_blocking(appender); + let json_layer = fmt::layer() + .json() + .with_writer(appender); + (registry.with(Some(json_layer)), Some(guard)) + } else { + (registry.with(None), None) }; - let subscriber = subscriber.with_env_filter(env_filter.clone()).finish(); - tracing::subscriber::set_global_default(subscriber)?; + registry + .with(fmt::layer()) + .init(); - debug!("Logging initialized with env_filter {env_filter}."); - Ok(()) + Ok(guard) } diff --git a/shared/src/middleware.rs b/shared/src/middleware.rs index 70250118..79ad249c 100644 --- a/shared/src/middleware.rs +++ b/shared/src/middleware.rs @@ -12,7 +12,7 @@ pub async fn log( ) -> Response { let method = req.method().clone(); let uri = req.uri().clone(); - let span = info_span!("", from = field::Empty); + let span = info_span!("request", from = field::Empty); async move { let resp = next.run(req).instrument(Span::current()).await; diff --git a/tests/Cargo.toml b/tests/Cargo.toml index abaa9860..6fb3c1d5 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -11,11 +11,11 @@ beam-lib = { workspace = true, features = ["http-util"] } once_cell = "1" serde_json = "1" anyhow = "1" -rand = "0.8" +rand.workspace = true serde = { version = "1", features = ["derive"] } -reqwest = { version = "0.12", features = ["stream"], default-features = false } -futures = "0.3.28" -async-sse = "5.1.0" +reqwest = { workspace = true, features = ["stream"], default-features = false } +futures.workspace = true +sse-stream = "0.2.3" [features] sockets = ["beam-lib/sockets"] diff --git a/tests/src/socket_test.rs b/tests/src/socket_test.rs index e49723a6..1df50d25 100644 --- a/tests/src/socket_test.rs +++ b/tests/src/socket_test.rs @@ -1,7 +1,7 @@ use std::time::Duration; use beam_lib::{BlockingOptions, MsgId}; -use rand::RngCore; +use rand::Rng; use tokio::io::{AsyncWriteExt, AsyncReadExt, AsyncRead, AsyncWrite}; use anyhow::Result; use crate::*; @@ -10,7 +10,7 @@ async fn test_connection(mut a: T, mut b: T) const N: usize = 2_usize.pow(8); for _ in 0..10 { let test_data: &mut [u8; N] = &mut [0; N]; - rand::thread_rng().fill_bytes(test_data); + rand::rng().fill_bytes(test_data); let mut read_buf = [0; N]; a.write_all(test_data).await?; a.flush().await?; diff --git a/tests/src/task_test.rs b/tests/src/task_test.rs index 957da30b..316cc7e7 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, diff --git a/tests/src/test_sse.rs b/tests/src/test_sse.rs index 2c0fff5d..705c964c 100644 --- a/tests/src/test_sse.rs +++ b/tests/src/test_sse.rs @@ -1,10 +1,8 @@ -use std::io; - use anyhow::{Result, bail, anyhow}; -use async_sse::Event; use beam_lib::TaskResult; -use futures::{StreamExt, TryStreamExt}; +use futures::StreamExt; use reqwest::{header::{self, HeaderValue}, Method}; +use sse_stream::SseStream; use crate::{client1, task_test}; @@ -25,10 +23,7 @@ async fn test_sse() -> Result<()> { task_test::put_result(id, "foo", Some(beam_lib::WorkStatus::Claimed)).await?; task_test::put_result(id, "foo", Some(beam_lib::WorkStatus::Claimed)).await?; task_test::put_result(id, "bar", Some(beam_lib::WorkStatus::Succeeded)).await?; - let mut stream = async_sse::decode(res.bytes_stream() - .map_err(|e| io::Error::new(io::ErrorKind::Other, e)) - .into_async_read() - ); + let mut stream = SseStream::from_byte_stream(res.bytes_stream()); assert_body(stream.next().await, "foo")?; assert_body(stream.next().await, "foo")?; assert_body(stream.next().await, "bar")?; @@ -36,14 +31,13 @@ async fn test_sse() -> Result<()> { Ok(()) } -fn assert_body(event: Option>, expected_body: &str) -> Result<()> { +fn assert_body(event: Option>, expected_body: &str) -> Result<()> { let Ok(event) = event.ok_or(anyhow!("SSE stream ended early"))? else { bail!("Unexpected error parsing SSE") }; - let Event::Message(m) = event else { - bail!("Event is not a message"); - }; - let result = serde_json::from_slice::>(m.data())?; + let result = serde_json::from_str::>( + event.data.as_deref().ok_or(anyhow!("Event has no data"))?, + )?; assert_eq!(result.body, expected_body); Ok(()) }