photon_backend/
input.rs

1//! Boundary validation and safe endpoint formatting.
2
3use std::fmt;
4
5use serde_json::Value;
6
7use crate::{PhotonError, Result};
8
9/// Maximum accepted topic name length in bytes.
10pub const MAX_TOPIC_NAME_BYTES: usize = 256;
11
12/// Maximum serialized JSON payload size in bytes.
13pub const MAX_PAYLOAD_JSON_BYTES: usize = 1024 * 1024;
14
15/// Validate a Photon topic name before it reaches a storage adapter.
16///
17/// NATS wildcard tokens are rejected so a Photon topic always denotes one concrete topic.
18///
19/// # Errors
20///
21/// Returns [`PhotonError::InvalidTopicName`] for empty, wildcard-containing, or oversized names.
22pub fn validate_topic_name(topic_name: &str) -> Result<()> {
23    if topic_name.is_empty() {
24        return Err(PhotonError::InvalidTopicName(
25            "topic name must not be empty".into(),
26        ));
27    }
28    if topic_name.len() > MAX_TOPIC_NAME_BYTES {
29        return Err(PhotonError::InvalidTopicName(format!(
30            "topic name exceeds {MAX_TOPIC_NAME_BYTES} bytes"
31        )));
32    }
33    if topic_name.contains('*') || topic_name.contains('>') {
34        return Err(PhotonError::InvalidTopicName(
35            "topic name must not contain wildcard tokens ('*' or '>')".into(),
36        ));
37    }
38    Ok(())
39}
40
41/// Reject payloads whose serialized JSON representation exceeds the transport limit.
42///
43/// # Errors
44///
45/// Returns [`PhotonError::PayloadError`] when serialization fails or exceeds the limit.
46pub fn validate_payload_size(payload: &Value) -> Result<()> {
47    let size = serde_json::to_vec(payload)?.len();
48    if size > MAX_PAYLOAD_JSON_BYTES {
49        return Err(PhotonError::PayloadError(format!(
50            "serialized payload exceeds {MAX_PAYLOAD_JSON_BYTES} bytes"
51        )));
52    }
53    Ok(())
54}
55
56/// Remove URL userinfo before placing an endpoint in an error or log message.
57#[must_use]
58pub fn redact_endpoint(endpoint: &str) -> String {
59    endpoint
60        .split(',')
61        .map(redact_single_endpoint)
62        .collect::<Vec<_>>()
63        .join(",")
64}
65
66fn redact_single_endpoint(endpoint: &str) -> String {
67    let Some(scheme_end) = endpoint.find("://") else {
68        return endpoint.to_string();
69    };
70    let authority_start = scheme_end + 3;
71    let authority = &endpoint[authority_start..];
72    let Some(userinfo_end) = authority.find('@') else {
73        return endpoint.to_string();
74    };
75    let userinfo_end = authority_start + userinfo_end;
76    let host_start = userinfo_end + 1;
77    if endpoint[authority_start..userinfo_end].contains(['/', '?', '#']) {
78        return endpoint.to_string();
79    }
80    format!(
81        "{}***@{}",
82        &endpoint[..authority_start],
83        &endpoint[host_start..]
84    )
85}
86
87/// Length of a URL-like prefix starting at `s` (scheme through host/path until whitespace).
88fn consume_endpoint_prefix(s: &str) -> usize {
89    let Some(scheme_end) = s.find("://") else {
90        return s.len();
91    };
92    let after_scheme = scheme_end + 3;
93    let rest = &s[after_scheme..];
94    let end_rel = rest
95        .find(|c: char| c.is_whitespace() || c == '"' || c == '\'' || c == ')' || c == ']')
96        .unwrap_or(rest.len());
97    after_scheme + end_rel
98}
99
100/// Redact `scheme://userinfo@host` substrings embedded in free-form error text.
101#[must_use]
102pub fn redact_credentials_in_text(text: &str) -> String {
103    let mut out = String::with_capacity(text.len());
104    let mut i = 0;
105    while i < text.len() {
106        if let Some(rel) = text[i..].find("://") {
107            let abs = i + rel;
108            let scheme_start = text[..abs]
109                .rfind(|c: char| !(c.is_ascii_alphanumeric() || c == '+' || c == '.' || c == '-'))
110                .map_or(i, |j| j + 1);
111            out.push_str(&text[i..scheme_start]);
112            let consumed = consume_endpoint_prefix(&text[scheme_start..]);
113            let endpoint = &text[scheme_start..scheme_start + consumed];
114            out.push_str(&redact_endpoint(endpoint));
115            i = scheme_start + consumed;
116        } else {
117            out.push_str(&text[i..]);
118            break;
119        }
120    }
121    out
122}
123
124/// Broker connect failure labeled with a redacted endpoint and redacted source text.
125///
126/// `label` is a short prefix such as `"nats connect"` or `"nats credentials file"`.
127/// Use for all adapter connect paths so URL userinfo never lands in error labels or sources.
128#[must_use]
129pub fn map_broker_connect_err(label: &str, endpoint: &str, err: impl fmt::Display) -> PhotonError {
130    let detail = redact_credentials_in_text(&err.to_string());
131    PhotonError::caused(format!("{label} {}", redact_endpoint(endpoint)), detail)
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    #[test]
139    fn accepts_concrete_topic_name() {
140        assert!(validate_topic_name("orders.created").is_ok());
141    }
142
143    #[test]
144    fn rejects_nats_wildcard_topic_names() {
145        assert!(matches!(
146            validate_topic_name("foo.>"),
147            Err(PhotonError::InvalidTopicName(_))
148        ));
149        assert!(matches!(
150            validate_topic_name("*"),
151            Err(PhotonError::InvalidTopicName(_))
152        ));
153    }
154
155    #[test]
156    fn rejects_empty_and_oversized_topic_names() {
157        assert!(matches!(
158            validate_topic_name(""),
159            Err(PhotonError::InvalidTopicName(_))
160        ));
161        assert!(matches!(
162            validate_topic_name(&"x".repeat(MAX_TOPIC_NAME_BYTES + 1)),
163            Err(PhotonError::InvalidTopicName(_))
164        ));
165    }
166
167    #[test]
168    fn rejects_oversized_payload() {
169        let payload = Value::String("x".repeat(MAX_PAYLOAD_JSON_BYTES));
170        assert!(matches!(
171            validate_payload_size(&payload),
172            Err(PhotonError::PayloadError(_))
173        ));
174    }
175
176    #[test]
177    fn redacts_url_userinfo() {
178        let redacted = redact_endpoint("nats://user:secret@host:4222");
179        assert_eq!(redacted, "nats://***@host:4222");
180        assert!(!redacted.contains("secret"));
181    }
182
183    #[test]
184    fn leaves_urls_without_userinfo() {
185        assert_eq!(
186            redact_endpoint("nats://127.0.0.1:4222"),
187            "nats://127.0.0.1:4222"
188        );
189    }
190
191    #[test]
192    fn redacts_embedded_url_in_error_text() {
193        let raw = "connection failed: nats://u:p@localhost:4222 refused";
194        let redacted = redact_credentials_in_text(raw);
195        assert!(redacted.contains("nats://***@localhost:4222"));
196        assert!(!redacted.contains("u:p@"));
197    }
198
199    #[test]
200    fn map_broker_connect_err_redacts_label_and_source() {
201        let err = map_broker_connect_err(
202            "nats connect",
203            "nats://user:secret@host:4222",
204            "dial nats://user:secret@host:4222 timed out",
205        );
206        let msg = err.to_string();
207        assert!(
208            msg.contains("nats connect nats://***@host:4222"),
209            "msg: {msg}"
210        );
211        assert!(!msg.contains("secret"), "msg: {msg}");
212        let source = std::error::Error::source(&err)
213            .expect("caused keeps source")
214            .to_string();
215        assert!(!source.contains("secret"), "source: {source}");
216        assert!(source.contains("nats://***@host:4222"), "source: {source}");
217    }
218
219    #[test]
220    fn map_broker_connect_err_sad_path_still_surfaces_failure() {
221        let err = map_broker_connect_err("kafka connect", "plain-host:9092", "broker down");
222        assert!(err.to_string().contains("kafka connect plain-host:9092"));
223        let source = std::error::Error::source(&err)
224            .expect("caused keeps source")
225            .to_string();
226        assert!(source.contains("broker down"));
227    }
228}
photon_backend/input.rs

photon_backend/
input.rs
