saluki_io/net/
addr.rs

1use std::{
2    fmt,
3    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4},
4    path::{Path, PathBuf},
5};
6
7use axum::extract::connect_info::Connected;
8use serde::Deserialize;
9use url::Url;
10
11use super::Connection;
12
13/// A listen address.
14///
15/// Listen addresses are used to bind listeners to specific local addresses and ports, and multiple address families and
16/// protocols are supported. In textual form, listen addresses are represented as URLs, with the scheme indicating the
17/// protocol and the authority/path representing the address to listen on.
18///
19/// # Examples
20///
21/// - `tcp://127.0.0.1:6789` (listen on IPv4 loopback, TCP port 6789)
22/// - `udp://[::1]:53` (listen on IPv6 loopback, UDP port 53)
23/// - `unixgram:///tmp/app.socket` (listen on a Unix datagram socket at `/tmp/app.socket`)
24/// - `unix:///tmp/app.socket` (listen on a Unix stream socket at `/tmp/app.socket`)
25#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
26#[serde(try_from = "String")]
27pub enum ListenAddress {
28    /// A TCP listen address.
29    Tcp(SocketAddr),
30
31    /// A UDP listen address.
32    Udp(SocketAddr),
33
34    /// A Unix datagram listen address.
35    Unixgram(PathBuf),
36
37    /// A Unix stream listen address.
38    Unix(PathBuf),
39
40    /// A Windows named pipe listen address.
41    NamedPipe {
42        /// Named pipe name without the `\\.\pipe\` prefix.
43        name: String,
44
45        /// Security descriptor string applied when creating the pipe.
46        security_descriptor: String,
47
48        /// Input buffer size to request from Windows when creating the pipe.
49        input_buffer_size: Option<u32>,
50    },
51}
52
53impl ListenAddress {
54    /// Creates a TCP address for the given port that listens on all IPv4 interfaces.
55    pub const fn tcp_any(port: u16) -> Self {
56        Self::Tcp(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port)))
57    }
58
59    /// Creates a TCP address for the given port that listens on the IPv4 loopback interface.
60    pub const fn tcp_loopback(port: u16) -> Self {
61        Self::Tcp(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), port)))
62    }
63
64    /// Creates a UDP address for the given port that listens on all IPv4 interfaces.
65    pub const fn udp_any(port: u16) -> Self {
66        Self::Udp(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port)))
67    }
68
69    /// Creates a UDP address for the given port that listens on the IPv4 loopback interface.
70    pub const fn udp_loopback(port: u16) -> Self {
71        Self::Udp(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), port)))
72    }
73
74    /// Creates a Windows named pipe listen address.
75    pub fn named_pipe(name: impl Into<String>, security_descriptor: impl Into<String>) -> Self {
76        Self::named_pipe_with_input_buffer_size(name, security_descriptor, None)
77    }
78
79    /// Creates a Windows named pipe listen address with a requested input buffer size.
80    pub fn named_pipe_with_input_buffer_size(
81        name: impl Into<String>, security_descriptor: impl Into<String>, input_buffer_size: impl Into<Option<u32>>,
82    ) -> Self {
83        Self::NamedPipe {
84            name: normalize_windows_named_pipe_name(name.into()),
85            security_descriptor: security_descriptor.into(),
86            input_buffer_size: input_buffer_size.into(),
87        }
88    }
89
90    /// Returns the socket type of the listen address.
91    pub const fn listener_type(&self) -> &'static str {
92        match self {
93            Self::Tcp(_) => "tcp",
94            Self::Udp(_) => "udp",
95            Self::Unixgram(_) => "unixgram",
96            Self::Unix(_) => "unix",
97            Self::NamedPipe { .. } => "named_pipe",
98        }
99    }
100
101    /// Returns a socket address that can be used to connect to the configured listen address with a bias for local
102    /// clients.
103    ///
104    /// When the listen address is a TCP or UDP address, this method returns a socket address that can be used to
105    /// connect to the listener bound to this listen address, such that if the listen address is unspecified
106    /// (`0.0.0.0`), the client will connect locally using `localhost`. When the listen address isn't unspecified or
107    /// already uses `localhost`, this method returns the listen address as-is.
108    ///
109    /// If the address is a Unix domain socket, this method returns `None`.
110    pub fn as_local_connect_addr(&self) -> Option<SocketAddr> {
111        match self {
112            Self::Tcp(addr) | Self::Udp(addr) => {
113                let mut connect_addr = *addr;
114                if connect_addr.ip().is_unspecified() {
115                    let localhost_ip = match connect_addr.is_ipv4() {
116                        true => IpAddr::V4(Ipv4Addr::LOCALHOST),
117                        false => IpAddr::V6(Ipv6Addr::LOCALHOST),
118                    };
119
120                    connect_addr.set_ip(localhost_ip);
121                }
122
123                Some(connect_addr)
124            }
125            // TODO: why did i do this? it's totally possible to connect to a unix domain socket locally...
126            // in fact, it's kind of the only way to connect to a unix domain socket :thonk:
127            Self::Unixgram(_) => None,
128            Self::Unix(_) => None,
129            Self::NamedPipe { .. } => None,
130        }
131    }
132
133    /// Returns the fully qualified Windows named pipe path, if this is a named pipe address.
134    pub fn as_windows_named_pipe_path(&self) -> Option<String> {
135        match self {
136            Self::NamedPipe { name, .. } => Some(format!(r"\\.\pipe\{name}")),
137            _ => None,
138        }
139    }
140
141    /// Returns the Windows named pipe security descriptor, if this is a named pipe address.
142    pub fn as_windows_named_pipe_security_descriptor(&self) -> Option<&str> {
143        match self {
144            Self::NamedPipe {
145                security_descriptor, ..
146            } => Some(security_descriptor.as_str()),
147            _ => None,
148        }
149    }
150
151    /// Returns the Unix domain socket path if the address is a Unix domain socket in SOCK_STREAM mode.
152    ///
153    /// Returns `None` otherwise.
154    pub fn as_unix_stream_path(&self) -> Option<&Path> {
155        match self {
156            Self::Unix(path) => Some(path),
157            _ => None,
158        }
159    }
160}
161
162fn normalize_windows_named_pipe_name(name: String) -> String {
163    name.strip_prefix(r"\\.\pipe\")
164        .or_else(|| name.strip_prefix(r"//./pipe/"))
165        .or_else(|| name.strip_prefix("pipe/"))
166        .unwrap_or(&name)
167        .to_string()
168}
169
170impl fmt::Display for ListenAddress {
171    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
172        match self {
173            Self::Tcp(addr) => write!(f, "tcp://{}", addr),
174            Self::Udp(addr) => write!(f, "udp://{}", addr),
175            Self::Unixgram(path) => write!(f, "unixgram://{}", path.display()),
176            Self::Unix(path) => write!(f, "unix://{}", path.display()),
177            Self::NamedPipe { name, .. } => write!(f, "npipe://{}", name),
178        }
179    }
180}
181
182impl TryFrom<String> for ListenAddress {
183    type Error = String;
184
185    fn try_from(value: String) -> Result<Self, Self::Error> {
186        Self::try_from(value.as_str())
187    }
188}
189
190impl<'a> TryFrom<&'a str> for ListenAddress {
191    type Error = String;
192
193    fn try_from(value: &'a str) -> Result<Self, Self::Error> {
194        let url = match Url::parse(value) {
195            Ok(url) => url,
196            Err(e) => match e {
197                url::ParseError::RelativeUrlWithoutBase => {
198                    Url::parse(&format!("unixgram://{}", value)).map_err(|e| e.to_string())?
199                }
200                _ => return Err(e.to_string()),
201            },
202        };
203
204        match url.scheme() {
205            "tcp" => {
206                let mut socket_addresses = url.socket_addrs(|| None).map_err(|e| e.to_string())?;
207                if socket_addresses.is_empty() {
208                    Err("listen address must resolve to at least one valid IP address/port pair".to_string())
209                } else {
210                    Ok(Self::Tcp(socket_addresses.swap_remove(0)))
211                }
212            }
213            "udp" => {
214                let mut socket_addresses = url.socket_addrs(|| None).map_err(|e| e.to_string())?;
215                if socket_addresses.is_empty() {
216                    Err("listen address must resolve to at least one valid IP address/port pair".to_string())
217                } else {
218                    Ok(Self::Udp(socket_addresses.swap_remove(0)))
219                }
220            }
221            "unixgram" => {
222                let path = url.path();
223                if path.is_empty() {
224                    return Err("socket path cannot be empty".to_string());
225                }
226
227                let path_buf = PathBuf::from(path);
228                if !path_buf.is_absolute() {
229                    return Err("socket path must be absolute".to_string());
230                }
231
232                Ok(Self::Unixgram(path_buf))
233            }
234            "unix" => {
235                let path = url.path();
236                if path.is_empty() {
237                    return Err("socket path cannot be empty".to_string());
238                }
239
240                let path_buf = PathBuf::from(path);
241                if !path_buf.is_absolute() {
242                    return Err("socket path must be absolute".to_string());
243                }
244
245                Ok(Self::Unix(path_buf))
246            }
247            "npipe" => {
248                let name = url.host_str().unwrap_or_else(|| url.path().trim_start_matches('/'));
249                if name.is_empty() {
250                    return Err("named pipe name cannot be empty".to_string());
251                }
252
253                Ok(Self::named_pipe(name, String::new()))
254            }
255            scheme => Err(format!("unknown/unsupported address scheme '{}'", scheme)),
256        }
257    }
258}
259
260/// A bound listen address.
261///
262/// This represents the "local" address of a listener (whether [`Listener`][crate::net::listener::Listener] or
263/// [`ConnectionOrientedListener`][crate::net::listener::ConnectionOrientedListener]), and is reciprocal to
264/// [`ListenAddress`], containing the same exact variants. This type is meant to be published as a dataspace assertion
265/// when a server binds to a particular [`ListenAddress`], providing the resolved address, such as the exact port used
266/// for a TCP/UDP socket when an ephemeral port was originally specified, and so on.
267#[derive(Clone, Debug, PartialEq, Eq)]
268pub enum BoundListenAddress {
269    /// A TCP listen address.
270    Tcp(SocketAddr),
271
272    /// A UDP listen address.
273    Udp(SocketAddr),
274
275    /// A Unix datagram listen address.
276    Unixgram(PathBuf),
277
278    /// A Unix stream listen address.
279    Unix(PathBuf),
280
281    /// A Windows named pipe listen address.
282    NamedPipe(String),
283}
284
285impl fmt::Display for BoundListenAddress {
286    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
287        match self {
288            Self::Tcp(addr) => write!(f, "tcp://{}", addr),
289            Self::Udp(addr) => write!(f, "udp://{}", addr),
290            Self::Unixgram(path) => write!(f, "unixgram://{}", path.display()),
291            Self::Unix(path) => write!(f, "unix://{}", path.display()),
292            Self::NamedPipe(name) => write!(f, "npipe://{}", name),
293        }
294    }
295}
296
297/// Process credentials for a Unix domain socket connection.
298///
299/// When dealing with Unix domain sockets, they can be configured such that the "process credentials" of the remote peer
300/// are sent as part of each received message. These "credentials" are the process ID of the remote peer, and the user
301/// ID and group ID that the process is running as.
302///
303/// In some cases, this information can be useful for identifying the remote peer and enriching the received data in an
304/// automatic way.
305#[derive(Clone)]
306pub struct ProcessCredentials {
307    /// Process ID of the remote peer.
308    pub pid: i32,
309
310    /// User ID of the remote peer process.
311    pub uid: u32,
312
313    /// Group ID of the remote peer process.
314    pub gid: u32,
315}
316
317/// Reason UDS process credential detection failed.
318#[derive(Clone, Copy)]
319pub enum ProcessCredentialsError {
320    /// Ancillary data was present but didn't contain usable process credentials.
321    InvalidCredentials,
322
323    /// Process credentials were present, but the PID was zero.
324    ZeroPid,
325
326    /// UDS process credential detection isn't supported on this platform.
327    UnsupportedPlatform,
328}
329
330impl ProcessCredentialsError {
331    /// Returns a concise identifier for the failure reason.
332    pub const fn identifier(&self) -> &'static str {
333        match self {
334            Self::InvalidCredentials => "invalid-credentials",
335            Self::ZeroPid => "zero-pid",
336            Self::UnsupportedPlatform => "unsupported-platform",
337        }
338    }
339}
340
341impl fmt::Display for ProcessCredentialsError {
342    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
343        match self {
344            Self::InvalidCredentials => write!(f, "invalid process credentials"),
345            Self::ZeroPid => write!(f, "process credential PID is zero"),
346            Self::UnsupportedPlatform => write!(f, "process credentials are unsupported on this platform"),
347        }
348    }
349}
350
351/// Process identity associated with a Unix domain socket peer.
352#[derive(Clone)]
353pub enum ProcessIdentity {
354    /// Process credentials were detected.
355    Credentials(ProcessCredentials),
356
357    /// Process credential detection failed.
358    Error(ProcessCredentialsError),
359
360    /// Process identity isn't available for this peer.
361    Unavailable,
362}
363
364impl ProcessIdentity {
365    /// Returns process credentials, if they were detected.
366    pub fn credentials(&self) -> Option<&ProcessCredentials> {
367        match self {
368            Self::Credentials(creds) => Some(creds),
369            Self::Error(_) | Self::Unavailable => None,
370        }
371    }
372
373    /// Returns `true` if process credential detection failed.
374    pub const fn is_error(&self) -> bool {
375        matches!(self, Self::Error(_))
376    }
377
378    /// Returns `true` if process credential detection failed for a per-message reason.
379    pub const fn is_telemetry_error(&self) -> bool {
380        matches!(
381            self,
382            Self::Error(ProcessCredentialsError::InvalidCredentials | ProcessCredentialsError::ZeroPid)
383        )
384    }
385}
386
387/// Connection address.
388///
389/// A generic representation of the address of a remote peer. This can either be a typical socket address (used for
390/// IPv4/IPv6), or potentially the process credentials of a Unix domain socket connection.
391#[derive(Clone)]
392pub enum ConnectionAddress {
393    /// A socket-like address.
394    SocketLike(SocketAddr),
395
396    /// A process-like address.
397    ProcessLike(ProcessIdentity),
398}
399
400impl fmt::Display for ConnectionAddress {
401    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
402        match self {
403            Self::SocketLike(addr) => write!(f, "{}", addr),
404            Self::ProcessLike(identity) => match identity {
405                ProcessIdentity::Credentials(creds) => {
406                    write!(f, "<pid={} uid={} gid={}>", creds.pid, creds.uid, creds.gid)
407                }
408                ProcessIdentity::Error(error) => write!(f, "<origin-detection-error: {}>", error.identifier()),
409                ProcessIdentity::Unavailable => write!(f, "<no-origin>"),
410            },
411        }
412    }
413}
414
415impl ConnectionAddress {
416    /// Returns process credentials for a Unix domain socket peer, if available.
417    pub fn process_credentials(&self) -> Option<&ProcessCredentials> {
418        match self {
419            Self::ProcessLike(identity) => identity.credentials(),
420            Self::SocketLike(_) => None,
421        }
422    }
423
424    /// Returns `true` if Unix domain socket process credential detection failed.
425    pub const fn has_process_credential_error(&self) -> bool {
426        match self {
427            Self::ProcessLike(identity) => identity.is_error(),
428            Self::SocketLike(_) => false,
429        }
430    }
431
432    /// Returns `true` if Unix domain socket process credential detection failed for a per-message reason.
433    pub const fn has_process_credential_telemetry_error(&self) -> bool {
434        match self {
435            Self::ProcessLike(identity) => identity.is_telemetry_error(),
436            Self::SocketLike(_) => false,
437        }
438    }
439}
440
441impl From<SocketAddr> for ConnectionAddress {
442    fn from(value: SocketAddr) -> Self {
443        Self::SocketLike(value)
444    }
445}
446
447impl From<ProcessCredentials> for ConnectionAddress {
448    fn from(creds: ProcessCredentials) -> Self {
449        Self::ProcessLike(ProcessIdentity::Credentials(creds))
450    }
451}
452
453impl<'a> Connected<&'a Connection> for ConnectionAddress {
454    fn connect_info(target: &'a Connection) -> Self {
455        target.remote_addr()
456    }
457}
458
459/// A gRPC target address.
460///
461/// This represents the address of a gRPC server that can be connected to. `GrpcTargetAddress` exposes a `Display`
462/// implementation that emits the target address following the rules of the [gRPC Name
463/// Resolution][grpc_name_resolution_docs] documentation.
464///
465/// Only connection-oriented transports are supported: TCP and Unix domain sockets in SOCK_STREAM mode.
466///
467/// [grpc_name_resolution_docs]: https://github.com/grpc/grpc/blob/master/doc/naming.md
468pub enum GrpcTargetAddress {
469    Tcp(SocketAddr),
470    Unix(PathBuf),
471}
472
473impl GrpcTargetAddress {
474    /// Creates a new `GrpcTargetAddress` from the given `ListenAddress`.
475    ///
476    /// For TCP addresses, this method converts unspecified addresses (`0.0.0.0` or `::`) to localhost
477    /// (`127.0.0.1` or `::1`) to ensure the advertised address matches TLS certificates.
478    ///
479    /// Returns `None` if the listen address isn't a connection-oriented transport.
480    pub fn try_from_listen_addr(listen_address: &ListenAddress) -> Option<Self> {
481        match listen_address {
482            ListenAddress::Tcp(_) => {
483                // For TCP, convert 0.0.0.0 to 127.0.0.1 to match TLS certificate
484                listen_address.as_local_connect_addr().map(GrpcTargetAddress::Tcp)
485            }
486            ListenAddress::Unix(path) => Some(GrpcTargetAddress::Unix(path.clone())),
487            _ => None,
488        }
489    }
490}
491
492impl fmt::Display for GrpcTargetAddress {
493    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
494        match self {
495            GrpcTargetAddress::Tcp(addr) => write!(f, "{}", addr),
496            GrpcTargetAddress::Unix(path) => write!(f, "unix://{}", path.display()),
497        }
498    }
499}
500
501#[cfg(test)]
502mod tests {
503    use super::*;
504
505    #[test]
506    fn named_pipe_listen_address_formats_with_windows_pipe_prefix() {
507        let address = ListenAddress::named_pipe("datadog-dogstatsd", "D:AI(A;;GA;;;WD)");
508
509        assert_eq!(address.listener_type(), "named_pipe");
510        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
511        assert_eq!(
512            address.as_windows_named_pipe_path().as_deref(),
513            Some(r"\\.\pipe\datadog-dogstatsd")
514        );
515    }
516
517    #[test]
518    fn named_pipe_listen_address_accepts_full_windows_pipe_path() {
519        let address = ListenAddress::named_pipe(r"\\.\pipe\datadog-dogstatsd", "D:AI(A;;GA;;;WD)");
520
521        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
522        assert_eq!(
523            address.as_windows_named_pipe_path().as_deref(),
524            Some(r"\\.\pipe\datadog-dogstatsd")
525        );
526    }
527
528    #[test]
529    fn npipe_url_parses_full_windows_pipe_path() {
530        let address = ListenAddress::try_from("npipe:////./pipe/datadog-dogstatsd").unwrap();
531
532        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
533        assert_eq!(
534            address.as_windows_named_pipe_path().as_deref(),
535            Some(r"\\.\pipe\datadog-dogstatsd")
536        );
537    }
538
539    #[test]
540    fn unsupported_platform_process_identity_is_not_a_telemetry_error() {
541        let peer_addr =
542            ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::UnsupportedPlatform));
543
544        assert!(peer_addr.has_process_credential_error());
545        assert!(!peer_addr.has_process_credential_telemetry_error());
546    }
547
548    #[test]
549    fn invalid_process_credentials_are_telemetry_errors() {
550        let peer_addr =
551            ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::InvalidCredentials));
552
553        assert!(peer_addr.has_process_credential_error());
554        assert!(peer_addr.has_process_credential_telemetry_error());
555    }
556
557    #[test]
558    fn zero_pid_process_credentials_are_telemetry_errors() {
559        let peer_addr = ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::ZeroPid));
560
561        assert!(peer_addr.has_process_credential_error());
562        assert!(peer_addr.has_process_credential_telemetry_error());
563    }
564
565    #[test]
566    fn as_local_connect_addr() {
567        let tcp_any_addr = ListenAddress::try_from("tcp://0.0.0.0:1234").unwrap();
568        assert_eq!(
569            tcp_any_addr.as_local_connect_addr(),
570            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1234)))
571        );
572
573        let tcp_localhost_addr = ListenAddress::try_from("tcp://127.0.0.1:2345").unwrap();
574        assert_eq!(
575            tcp_localhost_addr.as_local_connect_addr(),
576            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 2345)))
577        );
578
579        let tcp_private_addr = ListenAddress::try_from("tcp://192.168.10.2:3456").unwrap();
580        assert_eq!(
581            tcp_private_addr.as_local_connect_addr(),
582            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(192, 168, 10, 2), 3456)))
583        );
584
585        let udp_any_addr = ListenAddress::try_from("udp://0.0.0.0:4567").unwrap();
586        assert_eq!(
587            udp_any_addr.as_local_connect_addr(),
588            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 4567)))
589        );
590
591        let udp_localhost_addr = ListenAddress::try_from("udp://127.0.0.1:5678").unwrap();
592        assert_eq!(
593            udp_localhost_addr.as_local_connect_addr(),
594            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5678)))
595        );
596
597        let udp_private_addr = ListenAddress::try_from("udp://192.168.10.2:6789").unwrap();
598        assert_eq!(
599            udp_private_addr.as_local_connect_addr(),
600            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(192, 168, 10, 2), 6789)))
601        );
602    }
603
604    #[cfg(unix)]
605    #[test]
606    fn unix_scheme_parses_as_unix() {
607        let address = ListenAddress::try_from("unix:///tmp/otlp.sock").unwrap();
608        assert_eq!(address.listener_type(), "unix");
609        assert_eq!(
610            address.as_unix_stream_path(),
611            Some(std::path::Path::new("/tmp/otlp.sock"))
612        );
613    }
614}
saluki_io/net/addr.rs

saluki_io/net/
addr.rs
