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

saluki_io/net/
addr.rs
