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

saluki_io/net/
addr.rs
