lzc_sdk/
device_transport.rs

1use std::fmt;
2use std::future::Future;
3use std::pin::Pin;
4use std::task::{Context, Poll};
5
6use http::header::{HeaderName, HeaderValue};
7use http::{Request, Response};
8use tonic::body::Body;
9use tonic::transport::{Channel, Endpoint};
10use tower::{Service, ServiceExt as _};
11use url::{Host, Url};
12
13use crate::connection::compatibility_server_verifier;
14use crate::{ClientCredentials, Error, TokenProvider};
15
16const AUTH_TOKEN_HEADER: HeaderName = HeaderName::from_static("lzc_dapi_auth_token");
17
18pub(crate) async fn connect_device_channel(
19    api_url: &str,
20    credentials: &ClientCredentials,
21) -> Result<Channel, Error> {
22    let device = device_endpoint(api_url)?;
23    let endpoint = Endpoint::from_shared(device.uri)?.tls_config_with_verifier(
24        credentials.tls_config(&device.tls_server_name),
25        compatibility_server_verifier(),
26    )?;
27    Ok(endpoint.connect().await?)
28}
29
30struct DeviceEndpoint {
31    uri: String,
32    tls_server_name: String,
33}
34
35fn device_endpoint(api_url: &str) -> Result<DeviceEndpoint, Error> {
36    let url = Url::parse(api_url).map_err(|_| Error::InvalidDeviceUrl)?;
37    if !matches!(url.scheme(), "http" | "https")
38        || !url.username().is_empty()
39        || url.password().is_some()
40    {
41        return Err(Error::InvalidDeviceUrl);
42    }
43    let host = url.host().ok_or(Error::InvalidDeviceUrl)?;
44    let port = url.port_or_known_default().ok_or(Error::InvalidDeviceUrl)?;
45    let (authority_host, tls_server_name) = match host {
46        Host::Ipv6(address) => (format!("[{address}]"), address.to_string()),
47        Host::Ipv4(address) => {
48            let address = address.to_string();
49            (address.clone(), address)
50        }
51        Host::Domain(address) => {
52            let address = address.to_owned();
53            (address.clone(), address)
54        }
55    };
56    Ok(DeviceEndpoint {
57        uri: format!("https://{authority_host}:{port}"),
58        tls_server_name,
59    })
60}
61
62/// Tower service that injects a cached device auth token into every gRPC call.
63#[derive(Clone)]
64pub struct AuthenticatedService<S> {
65    inner: S,
66    provider: TokenProvider,
67}
68
69impl<S> AuthenticatedService<S> {
70    /// Wrap an inner gRPC service with a token provider.
71    #[must_use]
72    pub fn new(inner: S, provider: TokenProvider) -> Self {
73        Self { inner, provider }
74    }
75}
76
77impl<S> fmt::Debug for AuthenticatedService<S> {
78    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
79        formatter
80            .debug_struct("AuthenticatedService")
81            .field("provider", &self.provider)
82            .finish_non_exhaustive()
83    }
84}
85
86impl Service<Request<Body>> for AuthenticatedService<Channel> {
87    type Response = Response<Body>;
88    type Error = Error;
89    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
90
91    fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
92        self.inner.poll_ready(context).map_err(Error::Transport)
93    }
94
95    fn call(&mut self, mut request: Request<Body>) -> Self::Future {
96        let provider = self.provider.clone();
97        let inner = self.inner.clone();
98        Box::pin(async move {
99            let token = provider.token().await?;
100            let value = HeaderValue::try_from(token.expose_secret())
101                .map_err(|_| Error::InvalidAuthTokenMetadata)?;
102            request.headers_mut().insert(AUTH_TOKEN_HEADER, value);
103            inner.oneshot(request).await.map_err(Error::Transport)
104        })
105    }
106}
107
108#[cfg(test)]
109mod tests {
110    use super::device_endpoint;
111
112    #[test]
113    fn separates_ipv6_authority_from_tls_server_name() {
114        let endpoint = device_endpoint("https://[2001:db8::1]:8443/ignored?query=yes")
115            .expect("IPv6 device endpoint");
116        assert_eq!(endpoint.uri, "https://[2001:db8::1]:8443");
117        assert_eq!(endpoint.tls_server_name, "2001:db8::1");
118    }
119}
lzc_sdk/device_transport.rs

lzc_sdk/
device_transport.rs
