camel_bridge/
process.rs

1use std::fmt;
2use std::ops::Deref;
3use std::path::PathBuf;
4use thiserror::Error;
5use tokio::task::JoinHandle;
6use tokio_util::sync::CancellationToken;
7
8use crate::spec::{BridgeSpec, CXF_BRIDGE, JMS_BRIDGE, XML_BRIDGE};
9
10// ---------------------------------------------------------------------------
11// Redacted<T> — wrapper that never leaks inner value via Debug/Display
12// ---------------------------------------------------------------------------
13
14/// A newtype that redacts its inner value in `Debug` and `Display` output.
15/// Used for password/credential fields to prevent accidental logging.
16#[derive(Clone)]
17pub struct Redacted<T>(T);
18
19impl<T> Redacted<T> {
20    pub fn new(value: T) -> Self {
21        Self(value)
22    }
23
24    pub fn into_inner(self) -> T {
25        self.0
26    }
27}
28
29impl<T> fmt::Debug for Redacted<T> {
30    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
31        write!(f, "[REDACTED]")
32    }
33}
34
35impl<T> fmt::Display for Redacted<T> {
36    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
37        write!(f, "[REDACTED]")
38    }
39}
40
41impl<T> Deref for Redacted<T> {
42    type Target = T;
43
44    fn deref(&self) -> &Self::Target {
45        &self.0
46    }
47}
48
49#[derive(Debug, Error)]
50pub enum BridgeError {
51    #[error("IO error: {0}")]
52    Io(#[from] std::io::Error),
53    #[error("Bridge timed out: {0}")]
54    Timeout(String),
55    #[error("Bridge stdout closed before ready message")]
56    StdoutClosed,
57    #[error("Bridge ready message malformed: {0}")]
58    BadReadyMessage(String),
59    #[error("Download failed: {0}")]
60    Download(String),
61    #[error("Checksum mismatch: expected {expected}, got {actual}")]
62    ChecksumMismatch { expected: String, actual: String },
63    #[error("URL not allowed: {0}")]
64    UrlNotAllowed(String),
65    #[error("Transport error: {0}")]
66    Transport(String),
67    #[error("Config error: {0}")]
68    Config(String),
69}
70
71#[derive(Debug, Clone, PartialEq, serde::Deserialize)]
72#[serde(rename_all = "lowercase")]
73pub enum BrokerType {
74    #[serde(alias = "active_mq")]
75    ActiveMq,
76    Artemis,
77    Generic,
78}
79
80impl BrokerType {
81    pub fn as_env_str(&self) -> &'static str {
82        match self {
83            BrokerType::ActiveMq => "activemq",
84            BrokerType::Artemis => "artemis",
85            BrokerType::Generic => "generic",
86        }
87    }
88}
89
90impl std::str::FromStr for BrokerType {
91    type Err = BridgeError;
92
93    fn from_str(s: &str) -> Result<Self, Self::Err> {
94        match s.to_lowercase().as_str() {
95            "activemq" => Ok(BrokerType::ActiveMq),
96            "artemis" => Ok(BrokerType::Artemis),
97            "generic" => Ok(BrokerType::Generic),
98            other => Err(BridgeError::Config(format!("unknown broker type: {other}"))), // allow-secret
99        }
100    }
101}
102
103/// Environment variables for a single CXF profile, used by the bridge Java side.
104/// Password fields use [`Redacted`] to prevent accidental credential leakage in logs.
105#[derive(Debug)]
106pub struct CxfProfileEnvVars {
107    pub name: String,
108    pub wsdl_path: String,
109    pub service_name: String,
110    pub port_name: String,
111    pub address: Option<String>,
112    pub keystore_path: Option<String>,
113    pub keystore_password: Option<Redacted<String>>,
114    pub truststore_path: Option<String>,
115    pub truststore_password: Option<Redacted<String>>,
116    pub sig_username: Option<String>,
117    pub sig_password: Option<Redacted<String>>,
118    pub enc_username: Option<String>,
119    pub security_actions_out: Option<String>,
120    pub security_actions_in: Option<String>,
121    pub signature_algorithm: Option<String>,
122    pub signature_digest_algorithm: Option<String>,
123    pub signature_c14n_algorithm: Option<String>,
124    pub signature_parts: Option<String>,
125}
126
127impl CxfProfileEnvVars {
128    pub fn to_env_vars(&self) -> Vec<(String, String)> {
129        let prefix = format!("CXF_PROFILE_{}_", self.name.to_uppercase());
130        let mut vars = vec![
131            (format!("{}WSDL_PATH", prefix), self.wsdl_path.clone()),
132            (format!("{}SERVICE_NAME", prefix), self.service_name.clone()),
133            (format!("{}PORT_NAME", prefix), self.port_name.clone()),
134        ];
135
136        if let Some(ref v) = self.address {
137            vars.push((format!("{}ADDRESS", prefix), v.clone()));
138        }
139        if let Some(ref v) = self.keystore_path {
140            vars.push((format!("{}KEYSTORE_PATH", prefix), v.clone()));
141        }
142        if let Some(ref v) = self.keystore_password {
143            vars.push((format!("{}KEYSTORE_PASSWORD", prefix), (**v).clone()));
144        }
145        if let Some(ref v) = self.truststore_path {
146            vars.push((format!("{}TRUSTSTORE_PATH", prefix), v.clone()));
147        }
148        if let Some(ref v) = self.truststore_password {
149            vars.push((format!("{}TRUSTSTORE_PASSWORD", prefix), (**v).clone()));
150        }
151        if let Some(ref v) = self.sig_username {
152            vars.push((format!("{}SIG_USERNAME", prefix), v.clone()));
153        }
154        if let Some(ref v) = self.sig_password {
155            vars.push((format!("{}SIG_PASSWORD", prefix), (**v).clone()));
156        }
157        if let Some(ref v) = self.enc_username {
158            vars.push((format!("{}ENC_USERNAME", prefix), v.clone()));
159        }
160        if let Some(ref v) = self.security_actions_out {
161            vars.push((format!("{}SECURITY_ACTIONS_OUT", prefix), v.clone()));
162        }
163        if let Some(ref v) = self.security_actions_in {
164            vars.push((format!("{}SECURITY_ACTIONS_IN", prefix), v.clone()));
165        }
166        if let Some(ref v) = self.signature_algorithm {
167            vars.push((format!("{}SIGNATURE_ALGORITHM", prefix), v.clone()));
168        }
169        if let Some(ref v) = self.signature_digest_algorithm {
170            vars.push((format!("{}SIGNATURE_DIGEST_ALGORITHM", prefix), v.clone()));
171        }
172        if let Some(ref v) = self.signature_c14n_algorithm {
173            vars.push((format!("{}SIGNATURE_C14N_ALGORITHM", prefix), v.clone()));
174        }
175        if let Some(ref v) = self.signature_parts {
176            vars.push((format!("{}SIGNATURE_PARTS", prefix), v.clone()));
177        }
178
179        vars
180    }
181}
182
183/// Configuration for spawning a bridge subprocess.
184/// Password fields use [`Redacted`] to prevent accidental credential leakage in logs.
185#[derive(Debug)]
186pub struct BridgeProcessConfig {
187    pub spec: &'static BridgeSpec,
188    pub binary_path: PathBuf,
189    pub broker_url: String,
190    pub broker_type: BrokerType,
191    pub username: Option<String>,
192    pub password: Option<Redacted<String>>,
193    pub start_timeout_ms: u64,
194    pub env_vars: Vec<(String, String)>,
195}
196
197impl BridgeProcessConfig {
198    /// Constructor for the JMS bridge.
199    pub fn jms(
200        binary_path: PathBuf,
201        broker_url: String,
202        broker_type: BrokerType,
203        username: Option<String>,
204        password: Option<Redacted<String>>,
205        start_timeout_ms: u64,
206    ) -> Self {
207        let mut env_vars = vec![
208            ("BRIDGE_BROKER_URL".to_string(), broker_url.clone()),
209            (
210                "BRIDGE_BROKER_TYPE".to_string(),
211                broker_type.as_env_str().to_string(),
212            ),
213        ];
214        if let Some(u) = &username {
215            env_vars.push(("BRIDGE_USERNAME".to_string(), u.clone()));
216        }
217        if let Some(p) = &password {
218            env_vars.push(("BRIDGE_PASSWORD".to_string(), (**p).clone()));
219        }
220        Self {
221            spec: &JMS_BRIDGE,
222            binary_path,
223            broker_url,
224            broker_type,
225            username,
226            password,
227            start_timeout_ms,
228            env_vars,
229        }
230    }
231
232    /// Constructor for the XML bridge.
233    pub fn xml(binary_path: PathBuf, start_timeout_ms: u64) -> Self {
234        Self {
235            spec: &XML_BRIDGE,
236            binary_path,
237            broker_url: String::new(),
238            broker_type: BrokerType::Generic,
239            username: None,
240            password: None,
241            start_timeout_ms,
242            env_vars: vec![],
243        }
244    }
245
246    /// Constructor for the CXF bridge with multi-profile support.
247    /// Generates `CXF_PROFILES=list` env var plus per-profile env vars.
248    pub fn cxf_profiles(
249        binary_path: PathBuf,
250        profiles: &[CxfProfileEnvVars],
251        start_timeout_ms: u64,
252    ) -> Self {
253        let profile_names: Vec<String> = profiles.iter().map(|p| p.name.clone()).collect();
254        let mut env_vars = vec![("CXF_PROFILES".to_string(), profile_names.join(","))];
255
256        for profile in profiles {
257            env_vars.extend(profile.to_env_vars());
258        }
259
260        Self {
261            spec: &CXF_BRIDGE,
262            binary_path,
263            broker_url: String::new(),
264            broker_type: BrokerType::Generic,
265            username: None,
266            password: None,
267            start_timeout_ms,
268            env_vars,
269        }
270    }
271
272    pub fn validate(&self) -> Result<(), String> {
273        if self.start_timeout_ms == 0 {
274            return Err("start_timeout_ms must be > 0".to_string());
275        }
276        Ok(())
277    }
278}
279
280pub struct BridgeProcess {
281    child: tokio::process::Child,
282    grpc_port: u16,
283    tls: crate::tls::BridgeTlsMaterial,
284    token: CancellationToken,
285    handle: Option<JoinHandle<()>>,
286}
287
288impl BridgeProcess {
289    pub fn grpc_port(&self) -> u16 {
290        self.grpc_port
291    }
292
293    /// Connect a mTLS tonic channel to this bridge process.
294    pub async fn connect(&self) -> Result<tonic::transport::Channel, BridgeError> {
295        crate::channel::connect_channel(self.grpc_port, &self.tls).await
296    }
297
298    /// Start the bridge process and connect a mTLS channel in one step.
299    pub async fn start_and_connect(
300        config: &BridgeProcessConfig,
301    ) -> Result<(Self, tonic::transport::Channel), BridgeError> {
302        let process = Self::start(config).await?;
303        let channel = process.connect().await?;
304        Ok((process, channel))
305    }
306
307    /// Spawn the bridge process. Reads the SSL port from stdout JSON line:
308    ///   {"status":"ready","port":PORT}
309    ///
310    /// Picks a free OS port and passes it to the bridge via `QUARKUS_HTTP_SSL_PORT`
311    /// so Quarkus binds exactly to that port and PortAnnouncer can echo it back.
312    /// Build-time TLS props are in application.yml; only runtime cert paths
313    /// and the SSL port are passed via env vars.
314    pub async fn start(config: &BridgeProcessConfig) -> Result<Self, BridgeError> {
315        use tokio::io::AsyncBufReadExt;
316        use tokio::process::Command;
317        use tokio::time::{Duration, timeout};
318
319        config.validate().map_err(BridgeError::Config)?;
320
321        let tls = crate::tls::BridgeTlsMaterial::generate()?;
322
323        // Bind :0 to let the OS pick a free port, then release so the bridge can use it.
324        let free_port = {
325            let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
326            listener.local_addr()?.port()
327        };
328
329        // If CAMEL_BRIDGE_LOG_STDERR is set, redirect stderr to a file for debugging.
330        let stderr_stdio: std::process::Stdio =
331            if let Ok(log_dir) = std::env::var("CAMEL_BRIDGE_LOG_STDERR") {
332                let log_filename = config
333                    .spec
334                    .log_file_template
335                    .replace("{pid}", &std::process::id().to_string());
336                let log_path = if log_dir.is_empty() {
337                    format!("/tmp/{log_filename}")
338                } else {
339                    format!("{log_dir}/{log_filename}")
340                };
341                match std::fs::File::create(&log_path) {
342                    Ok(f) => {
343                        eprintln!("[camel-bridge] stderr → {}", log_path);
344                        f.into()
345                    }
346                    Err(e) => {
347                        eprintln!(
348                            "[camel-bridge] failed to create log file {}: {}",
349                            log_path, e
350                        );
351                        std::process::Stdio::inherit()
352                    }
353                }
354            } else {
355                std::process::Stdio::inherit()
356            };
357
358        let mut command = Command::new(&config.binary_path);
359        command
360            .env("QUARKUS_HTTP_SSL_PORT", free_port.to_string())
361            .env(
362                "QUARKUS_TLS_BRIDGE_KEY_STORE_PEM_0_CERT",
363                &tls.server_pem_path,
364            )
365            .env(
366                "QUARKUS_TLS_BRIDGE_KEY_STORE_PEM_0_KEY",
367                &tls.server_key_path,
368            )
369            .env("QUARKUS_TLS_BRIDGE_TRUST_STORE_PEM_CERTS", &tls.ca_pem_path)
370            .stdout(std::process::Stdio::piped())
371            .stderr(stderr_stdio);
372
373        // Inject bridge-specific env vars (e.g. JMS broker URL/credentials via ::jms()).
374        for (key, value) in &config.env_vars {
375            command.env(key, value);
376        }
377
378        let mut child = command.spawn()?;
379
380        let stdout = child.stdout.take().ok_or(BridgeError::StdoutClosed)?;
381        let mut reader = tokio::io::BufReader::new(stdout).lines();
382
383        let port = timeout(Duration::from_millis(config.start_timeout_ms), async {
384            while let Some(line) = reader.next_line().await? {
385                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line)
386                    && v.get("status").and_then(|s| s.as_str()) == Some("ready")
387                {
388                    if let Some(p) = v.get("port").and_then(|p| p.as_u64()) {
389                        return Ok(p as u16);
390                    }
391                    // log-policy: system-broken
392                    tracing::error!("bridge ready message malformed: {line}");
393                    return Err(BridgeError::BadReadyMessage(line));
394                }
395            }
396            // log-policy: system-broken
397            tracing::error!("bridge stdout closed before ready message");
398            Err(BridgeError::StdoutClosed)
399        })
400        .await
401        .map_err(|_| {
402            let msg = format!(
403                "{} failed to start: health check timeout after {}ms",
404                config.spec.name, config.start_timeout_ms
405            );
406            // log-policy: system-broken
407            tracing::error!("{msg}");
408            BridgeError::Timeout(msg)
409        })??;
410
411        // Keep draining stdout in background; allow cooperative cancellation.
412        let token = CancellationToken::new();
413        let child_token = token.clone();
414        let handle = tokio::spawn(async move {
415            loop {
416                tokio::select! {
417                    _ = child_token.cancelled() => break,
418                    line = reader.next_line() => {
419                        match line {
420                            Ok(Some(line)) => tracing::debug!(target: "camel_bridge::child", "{}", line),
421                            Ok(None) | Err(_) => break,
422                        }
423                    }
424                }
425            }
426        });
427
428        Ok(BridgeProcess {
429            child,
430            grpc_port: port,
431            tls,
432            token,
433            handle: Some(handle),
434        })
435    }
436
437    /// Gracefully stop: SIGTERM + wait for exit.
438    pub async fn stop(mut self) -> Result<(), BridgeError> {
439        use tokio::time::{Duration, sleep};
440
441        self.token.cancel();
442
443        if let Some(handle) = self.handle.take() {
444            let join_result = tokio::time::timeout(Duration::from_secs(5), handle).await;
445            if join_result.is_err() {
446                tracing::warn!("bridge stdout drain task did not exit after cancellation");
447            }
448        }
449
450        // Send SIGTERM first (graceful shutdown)
451        #[cfg(unix)]
452        {
453            let pid = self.child.id().unwrap_or(0);
454            if pid > 0 {
455                // SAFETY: libc::kill is called with the child process PID obtained from tokio.
456                unsafe {
457                    libc::kill(pid as i32, libc::SIGTERM);
458                }
459            }
460        }
461
462        // On non-Unix (Windows), fall through to kill immediately
463        #[cfg(not(unix))]
464        let _ = self.child.start_kill();
465
466        // Wait up to 5 seconds for graceful exit, then SIGKILL
467        tokio::select! {
468            result = self.child.wait() => {
469                result?;
470            }
471            _ = sleep(Duration::from_secs(5)) => {
472                let _ = self.child.start_kill();
473                self.child.wait().await?;
474            }
475        }
476        Ok(())
477    }
478}
479
480impl Drop for BridgeProcess {
481    fn drop(&mut self) {
482        self.token.cancel();
483        // Best-effort only. Does NOT wait — cannot block in Drop.
484        let _ = self.child.start_kill();
485    }
486}
487
488#[cfg(test)]
489mod tests {
490    use super::*;
491
492    #[test]
493    fn broker_type_from_str_activemq() {
494        assert_eq!(
495            "activemq".parse::<BrokerType>().unwrap(),
496            BrokerType::ActiveMq
497        );
498        assert_eq!(
499            "ACTIVEMQ".parse::<BrokerType>().unwrap(),
500            BrokerType::ActiveMq
501        );
502    }
503
504    #[test]
505    fn broker_type_from_str_artemis() {
506        assert_eq!(
507            "artemis".parse::<BrokerType>().unwrap(),
508            BrokerType::Artemis
509        );
510    }
511
512    #[test]
513    fn broker_type_from_str_generic() {
514        assert_eq!(
515            "generic".parse::<BrokerType>().unwrap(),
516            BrokerType::Generic
517        );
518    }
519
520    #[test]
521    fn broker_type_from_str_unknown_returns_err() {
522        assert!("ibmmq".parse::<BrokerType>().is_err());
523        assert!("UnknownBroker".parse::<BrokerType>().is_err());
524    }
525
526    #[test]
527    fn broker_type_env_str() {
528        assert_eq!(BrokerType::ActiveMq.as_env_str(), "activemq");
529        assert_eq!(BrokerType::Artemis.as_env_str(), "artemis");
530        assert_eq!(BrokerType::Generic.as_env_str(), "generic");
531    }
532
533    #[test]
534    fn jms_constructor_uses_jms_spec() {
535        let cfg = BridgeProcessConfig::jms(
536            PathBuf::from("/tmp/jms-bridge"),
537            "tcp://localhost:61616".to_string(),
538            BrokerType::ActiveMq,
539            Some("user".to_string()),
540            Some(Redacted::new("pass".to_string())),
541            1000,
542        );
543        assert_eq!(cfg.spec.name, "jms-bridge");
544    }
545
546    #[test]
547    fn xml_constructor_uses_xml_spec() {
548        let cfg = BridgeProcessConfig::xml(PathBuf::from("/tmp/xml-bridge"), 1000);
549        assert_eq!(cfg.spec.name, "xml-bridge");
550    }
551
552    #[test]
553    fn cxf_profiles_generates_cxf_profiles_env_var() {
554        let profiles = vec![
555            CxfProfileEnvVars {
556                name: "baleares".to_string(),
557                wsdl_path: "/a.wsdl".to_string(),
558                service_name: "Svc".to_string(),
559                port_name: "Port".to_string(),
560                address: None,
561                keystore_path: None,
562                keystore_password: None,
563                truststore_path: None,
564                truststore_password: None,
565                sig_username: None,
566                sig_password: None,
567                enc_username: None,
568                security_actions_out: None,
569                security_actions_in: None,
570                signature_algorithm: None,
571                signature_digest_algorithm: None,
572                signature_c14n_algorithm: None,
573                signature_parts: None,
574            },
575            CxfProfileEnvVars {
576                name: "extremadura".to_string(),
577                wsdl_path: "/b.wsdl".to_string(),
578                service_name: "Svc2".to_string(),
579                port_name: "Port2".to_string(),
580                address: Some("http://host:9090/ws".to_string()),
581                keystore_path: Some("/b.jks".to_string()),
582                keystore_password: Some(Redacted::new("pass".to_string())),
583                truststore_path: None,
584                truststore_password: None,
585                sig_username: Some("cert".to_string()),
586                sig_password: Some(Redacted::new("sig_pass".to_string())),
587                enc_username: None,
588                security_actions_out: Some("Timestamp Signature".to_string()),
589                security_actions_in: Some("Timestamp Signature".to_string()),
590                signature_algorithm: None,
591                signature_digest_algorithm: None,
592                signature_c14n_algorithm: None,
593                signature_parts: None,
594            },
595        ];
596
597        let cfg =
598            BridgeProcessConfig::cxf_profiles(PathBuf::from("/tmp/cxf-bridge"), &profiles, 15_000);
599
600        assert_eq!(cfg.spec.name, "cxf-bridge");
601        assert!(cfg.broker_url.is_empty());
602        assert_eq!(cfg.broker_type, BrokerType::Generic);
603        assert!(cfg.username.is_none());
604        assert!(cfg.password.is_none());
605
606        // Find CXF_PROFILES env var
607        let profiles_var = cfg
608            .env_vars
609            .iter()
610            .find(|(k, _)| k == "CXF_PROFILES")
611            .expect("CXF_PROFILES env var must exist");
612        assert_eq!(profiles_var.1, "baleares,extremadura");
613
614        // Check baleares profile vars (no security)
615        assert!(
616            cfg.env_vars
617                .iter()
618                .any(|(k, v)| k == "CXF_PROFILE_BALEARES_WSDL_PATH" && v == "/a.wsdl")
619        );
620        assert!(
621            cfg.env_vars
622                .iter()
623                .any(|(k, v)| k == "CXF_PROFILE_BALEARES_SERVICE_NAME" && v == "Svc")
624        );
625        assert!(
626            cfg.env_vars
627                .iter()
628                .any(|(k, v)| k == "CXF_PROFILE_BALEARES_PORT_NAME" && v == "Port")
629        );
630        assert!(
631            !cfg.env_vars
632                .iter()
633                .any(|(k, _)| k == "CXF_PROFILE_BALEARES_ADDRESS")
634        );
635
636        // Check extremadura profile vars (with security)
637        assert!(
638            cfg.env_vars
639                .iter()
640                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_WSDL_PATH" && v == "/b.wsdl")
641        );
642        assert!(
643            cfg.env_vars
644                .iter()
645                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_ADDRESS" && v == "http://host:9090/ws")
646        );
647        assert!(
648            cfg.env_vars
649                .iter()
650                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_KEYSTORE_PATH" && v == "/b.jks")
651        );
652        assert!(
653            cfg.env_vars
654                .iter()
655                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_KEYSTORE_PASSWORD" && v == "pass")
656        );
657        assert!(
658            cfg.env_vars
659                .iter()
660                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_SIG_USERNAME" && v == "cert")
661        );
662        assert!(
663            cfg.env_vars
664                .iter()
665                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_SIG_PASSWORD" && v == "sig_pass")
666        );
667        assert!(
668            cfg.env_vars
669                .iter()
670                .any(|(k, v)| k == "CXF_PROFILE_EXTREMADURA_SECURITY_ACTIONS_OUT"
671                    && v == "Timestamp Signature")
672        );
673    }
674
675    #[test]
676    fn cxf_profiles_single_profile_no_security() {
677        let profiles = vec![CxfProfileEnvVars {
678            name: "test".to_string(),
679            wsdl_path: "service.wsdl".to_string(),
680            service_name: "{http://example.com}Service".to_string(),
681            port_name: "{http://example.com}Port".to_string(),
682            address: None,
683            keystore_path: None,
684            keystore_password: None,
685            truststore_path: None,
686            truststore_password: None,
687            sig_username: None,
688            sig_password: None,
689            enc_username: None,
690            security_actions_out: None,
691            security_actions_in: None,
692            signature_algorithm: None,
693            signature_digest_algorithm: None,
694            signature_c14n_algorithm: None,
695            signature_parts: None,
696        }];
697
698        let cfg =
699            BridgeProcessConfig::cxf_profiles(PathBuf::from("/tmp/cxf-bridge"), &profiles, 15_000);
700
701        assert_eq!(cfg.spec.name, "cxf-bridge");
702        // CXF_PROFILES + 3 required vars (WSDL_PATH, SERVICE_NAME, PORT_NAME)
703        assert_eq!(cfg.env_vars.len(), 4);
704        assert_eq!(cfg.env_vars[0].0, "CXF_PROFILES");
705        assert_eq!(cfg.env_vars[0].1, "test");
706        assert_eq!(cfg.env_vars[1].0, "CXF_PROFILE_TEST_WSDL_PATH");
707        assert_eq!(cfg.env_vars[1].1, "service.wsdl");
708        assert_eq!(cfg.env_vars[2].0, "CXF_PROFILE_TEST_SERVICE_NAME");
709        assert_eq!(cfg.env_vars[2].1, "{http://example.com}Service");
710        assert_eq!(cfg.env_vars[3].0, "CXF_PROFILE_TEST_PORT_NAME");
711        assert_eq!(cfg.env_vars[3].1, "{http://example.com}Port");
712    }
713
714    #[test]
715    fn profile_env_vars_to_env_vars_includes_all_fields() {
716        let vars = CxfProfileEnvVars {
717            name: "full".to_string(),
718            wsdl_path: "/wsdl".to_string(),
719            service_name: "Svc".to_string(),
720            port_name: "Port".to_string(),
721            address: Some("http://host:8080".to_string()),
722            keystore_path: Some("/ks.jks".to_string()),
723            keystore_password: Some(Redacted::new("ks_pass".to_string())),
724            truststore_path: Some("/ts.jks".to_string()),
725            truststore_password: Some(Redacted::new("ts_pass".to_string())),
726            sig_username: Some("user".to_string()),
727            sig_password: Some(Redacted::new("sig_pass".to_string())),
728            enc_username: None,
729            security_actions_out: Some("Timestamp Signature".to_string()),
730            security_actions_in: Some("Timestamp".to_string()),
731            signature_algorithm: None,
732            signature_digest_algorithm: None,
733            signature_c14n_algorithm: None,
734            signature_parts: None,
735        };
736
737        let env = vars.to_env_vars();
738        // 3 required + 1 address + 8 security = 12
739        assert_eq!(env.len(), 12);
740
741        let keys: Vec<&str> = env.iter().map(|(k, _)| k.as_str()).collect();
742        assert!(keys.contains(&"CXF_PROFILE_FULL_WSDL_PATH"));
743        assert!(keys.contains(&"CXF_PROFILE_FULL_SERVICE_NAME"));
744        assert!(keys.contains(&"CXF_PROFILE_FULL_PORT_NAME"));
745        assert!(keys.contains(&"CXF_PROFILE_FULL_ADDRESS"));
746        assert!(keys.contains(&"CXF_PROFILE_FULL_KEYSTORE_PATH"));
747        assert!(keys.contains(&"CXF_PROFILE_FULL_KEYSTORE_PASSWORD"));
748        assert!(keys.contains(&"CXF_PROFILE_FULL_TRUSTSTORE_PATH"));
749        assert!(keys.contains(&"CXF_PROFILE_FULL_TRUSTSTORE_PASSWORD"));
750        assert!(keys.contains(&"CXF_PROFILE_FULL_SIG_USERNAME"));
751        assert!(keys.contains(&"CXF_PROFILE_FULL_SIG_PASSWORD"));
752        assert!(keys.contains(&"CXF_PROFILE_FULL_SECURITY_ACTIONS_OUT"));
753        assert!(keys.contains(&"CXF_PROFILE_FULL_SECURITY_ACTIONS_IN"));
754    }
755
756    #[test]
757    fn test_start_timeout_zero_rejected() {
758        let config = BridgeProcessConfig::jms(
759            PathBuf::from("/usr/bin/echo"),
760            "tcp://localhost:61616".to_string(),
761            BrokerType::ActiveMq,
762            None,
763            None,
764            0,
765        );
766        let result = config.validate();
767        assert!(result.is_err());
768    }
769
770    #[test]
771    fn test_bridge_rejects_zero_start_timeout() {
772        let config = BridgeProcessConfig::jms(
773            PathBuf::from("/usr/bin/echo"),
774            "tcp://localhost:61616".to_string(),
775            BrokerType::ActiveMq,
776            None,
777            None,
778            0,
779        );
780        assert!(config.validate().is_err());
781    }
782
783    #[tokio::test]
784    async fn test_bridge_stop_completes() {
785        use tokio::process::Command;
786        use tokio::time::{Duration, timeout};
787
788        let child = Command::new("sh")
789            .arg("-c")
790            .arg("trap '' TERM; while true; do echo tick; sleep 1; done")
791            .stdout(std::process::Stdio::null())
792            .spawn()
793            .expect("must spawn test child process");
794
795        let bridge = BridgeProcess {
796            child,
797            grpc_port: 0,
798            tls: crate::tls::BridgeTlsMaterial::generate().expect("test tls"),
799            token: CancellationToken::new(),
800            handle: None,
801        };
802
803        let result = timeout(Duration::from_secs(10), bridge.stop()).await;
804        assert!(result.is_ok(), "stop() must complete within 10s");
805    }
806
807    // --- BRG-004: Redacted<T> tests ---
808
809    #[test]
810    fn redacted_debug_displays_redacted() {
811        let r = Redacted::new("secret_password".to_string());
812        assert_eq!(format!("{r:?}"), "[REDACTED]");
813    }
814
815    #[test]
816    fn redacted_display_displays_redacted() {
817        let r = Redacted::new("secret_password".to_string());
818        assert_eq!(format!("{r}"), "[REDACTED]");
819    }
820
821    #[test]
822    fn redacted_deref_gives_inner_value() {
823        let r = Redacted::new("secret".to_string());
824        assert_eq!(&*r, "secret");
825    }
826
827    #[test]
828    fn redacted_into_inner_returns_value() {
829        let r = Redacted::new("secret".to_string());
830        assert_eq!(r.into_inner(), "secret");
831    }
832
833    #[test]
834    fn redacted_clone_works() {
835        let r = Redacted::new("secret".to_string());
836        let c = r.clone();
837        assert_eq!(&*c, "secret");
838        assert_eq!(format!("{c:?}"), "[REDACTED]");
839    }
840
841    #[test]
842    fn bridge_process_config_debug_redacts_password() {
843        let cfg = BridgeProcessConfig::jms(
844            PathBuf::from("/tmp/jms-bridge"),
845            "tcp://localhost:61616".to_string(),
846            BrokerType::ActiveMq,
847            Some("user".to_string()),
848            Some(Redacted::new("super_secret".to_string())),
849            1000,
850        );
851        // The Redacted<T> password field must show [REDACTED] in debug output.
852        // env_vars is a separate Vec<(String, String)> used for process injection
853        // and legitimately contains the raw value — that is not a Redacted leak.
854        let password_debug = format!("{:?}", cfg.password); // allow-secret
855        assert!(
856            !password_debug.contains("super_secret"),
857            "Password field must not leak in Debug: {password_debug}"
858        );
859        assert_eq!(
860            password_debug, "Some([REDACTED])",
861            "Password field must show [REDACTED]: {password_debug}"
862        );
863    }
864
865    #[test]
866    fn cxf_profile_debug_redacts_passwords() {
867        let profile = CxfProfileEnvVars {
868            name: "test".to_string(),
869            wsdl_path: "/a.wsdl".to_string(),
870            service_name: "Svc".to_string(),
871            port_name: "Port".to_string(),
872            address: None,
873            keystore_path: None,
874            keystore_password: Some(Redacted::new("ks_secret_val".to_string())),
875            truststore_path: None,
876            truststore_password: Some(Redacted::new("ts_secret_val".to_string())),
877            sig_username: None,
878            sig_password: Some(Redacted::new("sig_secret_val".to_string())),
879            enc_username: None,
880            security_actions_out: None,
881            security_actions_in: None,
882            signature_algorithm: None,
883            signature_digest_algorithm: None,
884            signature_c14n_algorithm: None,
885            signature_parts: None,
886        };
887        let debug_output = format!("{profile:?}");
888        assert!(
889            !debug_output.contains("ks_secret_val")
890                && !debug_output.contains("ts_secret_val")
891                && !debug_output.contains("sig_secret_val"),
892            "Debug must not contain passwords: {debug_output}"
893        );
894    }
895}
camel_bridge/process.rs

camel_bridge/
process.rs
