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    /// The peer process credentials of a connected socket couldn't be looked up.
330    PeerLookupFailed,
331}
332
333impl ProcessCredentialsError {
334    /// Returns a concise identifier for the failure reason.
335    pub const fn identifier(&self) -> &'static str {
336        match self {
337            Self::InvalidCredentials => "invalid-credentials",
338            Self::ZeroPid => "zero-pid",
339            Self::UnsupportedPlatform => "unsupported-platform",
340            Self::PeerLookupFailed => "peer-lookup-failed",
341        }
342    }
343}
344
345impl fmt::Display for ProcessCredentialsError {
346    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
347        match self {
348            Self::InvalidCredentials => write!(f, "invalid process credentials"),
349            Self::ZeroPid => write!(f, "process credential PID is zero"),
350            Self::UnsupportedPlatform => write!(f, "process credentials are unsupported on this platform"),
351            Self::PeerLookupFailed => write!(f, "failed to look up peer process credentials"),
352        }
353    }
354}
355
356/// Process identity associated with a Unix domain socket peer.
357#[derive(Clone)]
358pub enum ProcessIdentity {
359    /// Process credentials were detected.
360    Credentials(ProcessCredentials),
361
362    /// Process credential detection failed.
363    Error(ProcessCredentialsError),
364
365    /// Process identity isn't available for this peer.
366    Unavailable,
367}
368
369impl ProcessIdentity {
370    /// Builds the identity of a peer process from its credentials, rejecting a zero PID.
371    #[cfg(unix)]
372    pub(crate) fn new(pid: i32, uid: u32, gid: u32) -> Self {
373        if pid == 0 {
374            Self::Error(ProcessCredentialsError::ZeroPid)
375        } else {
376            Self::Credentials(ProcessCredentials { pid, uid, gid })
377        }
378    }
379
380    /// Returns process credentials, if they were detected.
381    pub fn credentials(&self) -> Option<&ProcessCredentials> {
382        match self {
383            Self::Credentials(creds) => Some(creds),
384            Self::Error(_) | Self::Unavailable => None,
385        }
386    }
387
388    /// Returns `true` if process credential detection failed.
389    pub const fn is_error(&self) -> bool {
390        matches!(self, Self::Error(_))
391    }
392
393    /// Returns `true` if process credential detection failed for a per-message reason.
394    pub const fn is_telemetry_error(&self) -> bool {
395        matches!(
396            self,
397            Self::Error(ProcessCredentialsError::InvalidCredentials | ProcessCredentialsError::ZeroPid)
398        )
399    }
400}
401
402/// Connection address.
403///
404/// A generic representation of the address of a remote peer. This can either be a typical socket address (used for
405/// IPv4/IPv6), or potentially the process credentials of a Unix domain socket connection.
406#[derive(Clone)]
407pub enum ConnectionAddress {
408    /// A socket-like address.
409    SocketLike(SocketAddr),
410
411    /// A process-like address.
412    ProcessLike(ProcessIdentity),
413}
414
415impl fmt::Display for ConnectionAddress {
416    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
417        match self {
418            Self::SocketLike(addr) => write!(f, "{}", addr),
419            Self::ProcessLike(identity) => match identity {
420                ProcessIdentity::Credentials(creds) => {
421                    write!(f, "<pid={} uid={} gid={}>", creds.pid, creds.uid, creds.gid)
422                }
423                ProcessIdentity::Error(error) => write!(f, "<origin-detection-error: {}>", error.identifier()),
424                ProcessIdentity::Unavailable => write!(f, "<no-origin>"),
425            },
426        }
427    }
428}
429
430impl ConnectionAddress {
431    /// Returns process credentials for a Unix domain socket peer, if available.
432    pub fn process_credentials(&self) -> Option<&ProcessCredentials> {
433        match self {
434            Self::ProcessLike(identity) => identity.credentials(),
435            Self::SocketLike(_) => None,
436        }
437    }
438
439    /// Returns `true` if Unix domain socket process credential detection failed.
440    pub const fn has_process_credential_error(&self) -> bool {
441        match self {
442            Self::ProcessLike(identity) => identity.is_error(),
443            Self::SocketLike(_) => false,
444        }
445    }
446
447    /// Returns `true` if Unix domain socket process credential detection failed for a per-message reason.
448    pub const fn has_process_credential_telemetry_error(&self) -> bool {
449        match self {
450            Self::ProcessLike(identity) => identity.is_telemetry_error(),
451            Self::SocketLike(_) => false,
452        }
453    }
454}
455
456impl From<SocketAddr> for ConnectionAddress {
457    fn from(value: SocketAddr) -> Self {
458        Self::SocketLike(value)
459    }
460}
461
462impl From<ProcessCredentials> for ConnectionAddress {
463    fn from(creds: ProcessCredentials) -> Self {
464        Self::ProcessLike(ProcessIdentity::Credentials(creds))
465    }
466}
467
468impl<'a> Connected<&'a Connection> for ConnectionAddress {
469    fn connect_info(target: &'a Connection) -> Self {
470        target.remote_addr()
471    }
472}
473
474/// A gRPC target address.
475///
476/// This represents the address of a gRPC server that can be connected to. `GrpcTargetAddress` exposes a `Display`
477/// implementation that emits the target address following the rules of the [gRPC Name
478/// Resolution][grpc_name_resolution_docs] documentation.
479///
480/// Only connection-oriented transports are supported: TCP and Unix domain sockets in SOCK_STREAM mode.
481///
482/// [grpc_name_resolution_docs]: https://github.com/grpc/grpc/blob/master/doc/naming.md
483pub enum GrpcTargetAddress {
484    Tcp(SocketAddr),
485    Unix(PathBuf),
486}
487
488impl GrpcTargetAddress {
489    /// Creates a new `GrpcTargetAddress` from the given `ListenAddress`.
490    ///
491    /// For TCP addresses, this method converts unspecified addresses (`0.0.0.0` or `::`) to localhost
492    /// (`127.0.0.1` or `::1`) to ensure the advertised address matches TLS certificates.
493    ///
494    /// Returns `None` if the listen address isn't a connection-oriented transport.
495    pub fn try_from_listen_addr(listen_address: &ListenAddress) -> Option<Self> {
496        match listen_address {
497            ListenAddress::Tcp(_) => {
498                // For TCP, convert 0.0.0.0 to 127.0.0.1 to match TLS certificate
499                listen_address.as_local_connect_addr().map(GrpcTargetAddress::Tcp)
500            }
501            ListenAddress::Unix(path) => Some(GrpcTargetAddress::Unix(path.clone())),
502            _ => None,
503        }
504    }
505}
506
507impl fmt::Display for GrpcTargetAddress {
508    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
509        match self {
510            GrpcTargetAddress::Tcp(addr) => write!(f, "{}", addr),
511            GrpcTargetAddress::Unix(path) => write!(f, "unix://{}", path.display()),
512        }
513    }
514}
515
516#[cfg(test)]
517mod tests {
518    use super::*;
519
520    #[test]
521    fn named_pipe_listen_address_formats_with_windows_pipe_prefix() {
522        let address = ListenAddress::named_pipe("datadog-dogstatsd", "D:AI(A;;GA;;;WD)");
523
524        assert_eq!(address.listener_type(), "named_pipe");
525        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
526        assert_eq!(
527            address.as_windows_named_pipe_path().as_deref(),
528            Some(r"\\.\pipe\datadog-dogstatsd")
529        );
530    }
531
532    #[test]
533    fn named_pipe_listen_address_accepts_full_windows_pipe_path() {
534        let address = ListenAddress::named_pipe(r"\\.\pipe\datadog-dogstatsd", "D:AI(A;;GA;;;WD)");
535
536        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
537        assert_eq!(
538            address.as_windows_named_pipe_path().as_deref(),
539            Some(r"\\.\pipe\datadog-dogstatsd")
540        );
541    }
542
543    #[test]
544    fn npipe_url_parses_full_windows_pipe_path() {
545        let address = ListenAddress::try_from("npipe:////./pipe/datadog-dogstatsd").unwrap();
546
547        assert_eq!(address.to_string(), r"npipe://datadog-dogstatsd");
548        assert_eq!(
549            address.as_windows_named_pipe_path().as_deref(),
550            Some(r"\\.\pipe\datadog-dogstatsd")
551        );
552    }
553
554    #[test]
555    fn unsupported_platform_process_identity_is_not_a_telemetry_error() {
556        let peer_addr =
557            ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::UnsupportedPlatform));
558
559        assert!(peer_addr.has_process_credential_error());
560        assert!(!peer_addr.has_process_credential_telemetry_error());
561    }
562
563    #[test]
564    fn invalid_process_credentials_are_telemetry_errors() {
565        let peer_addr =
566            ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::InvalidCredentials));
567
568        assert!(peer_addr.has_process_credential_error());
569        assert!(peer_addr.has_process_credential_telemetry_error());
570    }
571
572    #[test]
573    fn zero_pid_process_credentials_are_telemetry_errors() {
574        let peer_addr = ConnectionAddress::ProcessLike(ProcessIdentity::Error(ProcessCredentialsError::ZeroPid));
575
576        assert!(peer_addr.has_process_credential_error());
577        assert!(peer_addr.has_process_credential_telemetry_error());
578    }
579
580    #[test]
581    fn as_local_connect_addr() {
582        let tcp_any_addr = ListenAddress::try_from("tcp://0.0.0.0:1234").unwrap();
583        assert_eq!(
584            tcp_any_addr.as_local_connect_addr(),
585            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1234)))
586        );
587
588        let tcp_localhost_addr = ListenAddress::try_from("tcp://127.0.0.1:2345").unwrap();
589        assert_eq!(
590            tcp_localhost_addr.as_local_connect_addr(),
591            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 2345)))
592        );
593
594        let tcp_private_addr = ListenAddress::try_from("tcp://192.168.10.2:3456").unwrap();
595        assert_eq!(
596            tcp_private_addr.as_local_connect_addr(),
597            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(192, 168, 10, 2), 3456)))
598        );
599
600        let udp_any_addr = ListenAddress::try_from("udp://0.0.0.0:4567").unwrap();
601        assert_eq!(
602            udp_any_addr.as_local_connect_addr(),
603            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 4567)))
604        );
605
606        let udp_localhost_addr = ListenAddress::try_from("udp://127.0.0.1:5678").unwrap();
607        assert_eq!(
608            udp_localhost_addr.as_local_connect_addr(),
609            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5678)))
610        );
611
612        let udp_private_addr = ListenAddress::try_from("udp://192.168.10.2:6789").unwrap();
613        assert_eq!(
614            udp_private_addr.as_local_connect_addr(),
615            Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(192, 168, 10, 2), 6789)))
616        );
617    }
618
619    #[cfg(unix)]
620    #[test]
621    fn unix_scheme_parses_as_unix() {
622        let address = ListenAddress::try_from("unix:///tmp/otlp.sock").unwrap();
623        assert_eq!(address.listener_type(), "unix");
624        assert_eq!(
625            address.as_unix_stream_path(),
626            Some(std::path::Path::new("/tmp/otlp.sock"))
627        );
628    }
629}
saluki_io/net/addr.rs

saluki_io/net/
addr.rs
