libsignal_service/push_service/
mod.rs

1use std::{sync::LazyLock, time::Duration};
2
3use crate::{
4    configuration::{Endpoint, ServiceCredentials, SignalServers},
5    prelude::ServiceConfiguration,
6    utils::serde_device_id_vec,
7    websocket::{SignalWebSocket, WebSocketType},
8};
9
10use libsignal_core::DeviceId;
11use protobuf::ProtobufResponseExt;
12use reqwest::{Method, RequestBuilder};
13use reqwest_websocket::Upgrade;
14use serde::{Deserialize, Serialize};
15use tracing::{debug_span, Instrument};
16
17pub const KEEPALIVE_TIMEOUT_SECONDS: Duration = Duration::from_secs(55);
18pub static DEFAULT_DEVICE_ID: LazyLock<libsignal_core::DeviceId> =
19    LazyLock::new(|| libsignal_core::DeviceId::try_from(1).unwrap());
20
21mod account;
22mod cdn;
23mod error;
24pub mod linking;
25pub(crate) mod response;
26
27pub use account::*;
28pub use cdn::*;
29pub use error::*;
30pub(crate) use response::{ReqwestExt, SignalServiceResponse};
31
32#[derive(Debug, Serialize, Deserialize)]
33pub struct ProofRequired {
34    pub token: String,
35    pub options: Vec<String>,
36}
37
38#[derive(derive_more::Debug, Clone, Serialize, Deserialize)]
39pub struct HttpAuth {
40    pub username: String,
41    #[debug(ignore)]
42    pub password: String,
43}
44
45#[derive(Debug, Clone)]
46pub enum HttpAuthOverride {
47    NoOverride,
48    Unidentified,
49    Identified(HttpAuth),
50}
51
52#[derive(Debug, Clone, Eq, PartialEq)]
53pub enum AvatarWrite<C> {
54    NewAvatar(C),
55    RetainAvatar,
56    NoAvatar,
57}
58
59#[derive(Debug, Deserialize)]
60#[serde(rename_all = "camelCase")]
61pub struct MismatchedDevices {
62    #[serde(with = "serde_device_id_vec")]
63    pub missing_devices: Vec<DeviceId>,
64    #[serde(with = "serde_device_id_vec")]
65    pub extra_devices: Vec<DeviceId>,
66}
67
68#[derive(Debug, Deserialize)]
69#[serde(rename_all = "camelCase")]
70pub struct StaleDevices {
71    #[serde(with = "serde_device_id_vec")]
72    pub stale_devices: Vec<DeviceId>,
73}
74
75#[derive(Clone)]
76pub struct PushService {
77    pub(crate) servers: SignalServers,
78    cfg: ServiceConfiguration,
79    credentials: Option<HttpAuth>,
80    client: reqwest::Client,
81}
82
83impl PushService {
84    pub fn new(
85        env: SignalServers,
86        credentials: Option<ServiceCredentials>,
87        user_agent: impl AsRef<str>,
88    ) -> Self {
89        let cfg: ServiceConfiguration = env.into();
90
91        // Use the ring provider except if the application already installed one.
92        if rustls::crypto::CryptoProvider::get_default().is_none() {
93            let _ = rustls::crypto::ring::default_provider().install_default();
94        }
95
96        let client = reqwest::ClientBuilder::new()
97            .tls_certs_only([reqwest::Certificate::from_pem(
98                cfg.certificate_authority.as_bytes(),
99            )
100            .unwrap()])
101            .connect_timeout(Duration::from_secs(10))
102            .timeout(Duration::from_secs(65))
103            .user_agent(user_agent.as_ref())
104            .http1_only()
105            .build()
106            .unwrap();
107
108        Self {
109            servers: env,
110            cfg,
111            credentials: credentials.and_then(|c| c.authorization()),
112            client,
113        }
114    }
115
116    #[tracing::instrument(skip(self), fields(endpoint = %endpoint))]
117    pub fn request(
118        &self,
119        method: Method,
120        endpoint: Endpoint,
121        auth_override: HttpAuthOverride,
122    ) -> Result<RequestBuilder, ServiceError> {
123        let url = endpoint.into_url(&self.cfg)?;
124        let mut builder = self.client.request(method, url);
125
126        builder = match auth_override {
127            HttpAuthOverride::NoOverride => {
128                if let Some(HttpAuth { username, password }) =
129                    self.credentials.as_ref()
130                {
131                    builder.basic_auth(username, Some(password))
132                } else {
133                    builder
134                }
135            },
136            HttpAuthOverride::Identified(HttpAuth { username, password }) => {
137                builder.basic_auth(username, Some(password))
138            },
139            HttpAuthOverride::Unidentified => builder,
140        };
141
142        Ok(builder)
143    }
144
145    pub async fn ws<C: WebSocketType>(
146        &mut self,
147        path: &str,
148        keepalive_path: &str,
149        additional_headers: &[(&'static str, &str)],
150        credentials: Option<ServiceCredentials>,
151    ) -> Result<SignalWebSocket<C>, ServiceError> {
152        let span = debug_span!("websocket");
153
154        let mut url = Endpoint::service(path).into_url(&self.cfg)?;
155        url.set_scheme("wss").expect("valid https base url");
156
157        let mut builder = self.client.get(url);
158        for (key, value) in additional_headers {
159            builder = builder.header(*key, *value);
160        }
161
162        if let Some(credentials) = credentials {
163            builder =
164                builder.basic_auth(credentials.login(), credentials.password);
165        }
166
167        let ws = builder
168            .upgrade()
169            .send()
170            .await?
171            .into_websocket()
172            .instrument(span.clone())
173            .await?;
174
175        let unidentified_push_service = PushService {
176            servers: self.servers,
177            cfg: self.cfg.clone(),
178            credentials: None,
179            client: self.client.clone(),
180        };
181        let (ws, task) = SignalWebSocket::new(
182            ws,
183            keepalive_path.to_owned(),
184            unidentified_push_service,
185        );
186        let task = task.instrument(span);
187        tokio::task::spawn(task);
188        Ok(ws)
189    }
190
191    pub(crate) async fn get_group(
192        &mut self,
193        credentials: HttpAuth,
194    ) -> Result<crate::proto::Group, ServiceError> {
195        self.request(
196            Method::GET,
197            Endpoint::storage("/v1/groups/"),
198            HttpAuthOverride::Identified(credentials),
199        )?
200        .send()
201        .await?
202        .service_error_for_status()
203        .await?
204        .protobuf()
205        .await
206    }
207}
208
209pub(crate) mod protobuf {
210    use async_trait::async_trait;
211    use prost::{EncodeError, Message};
212    use reqwest::{header, RequestBuilder, Response};
213
214    use super::ServiceError;
215
216    pub(crate) trait ProtobufRequestBuilderExt
217    where
218        Self: Sized,
219    {
220        /// Set the request payload encoded as protobuf.
221        /// Sets the `Content-Type` header to `application/x-protobuf`
222        #[allow(dead_code)]
223        fn protobuf<T: Message + Default>(
224            self,
225            value: T,
226        ) -> Result<Self, EncodeError>;
227    }
228
229    #[async_trait::async_trait]
230    pub(crate) trait ProtobufResponseExt {
231        /// Get the response body decoded from Protobuf
232        async fn protobuf<T>(self) -> Result<T, ServiceError>
233        where
234            T: prost::Message + Default;
235    }
236
237    impl ProtobufRequestBuilderExt for RequestBuilder {
238        fn protobuf<T: Message + Default>(
239            self,
240            value: T,
241        ) -> Result<Self, EncodeError> {
242            let mut buf = Vec::new();
243            value.encode(&mut buf)?;
244            let this =
245                self.header(header::CONTENT_TYPE, "application/x-protobuf");
246            Ok(this.body(buf))
247        }
248    }
249
250    #[async_trait]
251    impl ProtobufResponseExt for Response {
252        async fn protobuf<T>(self) -> Result<T, ServiceError>
253        where
254            T: Message + Default,
255        {
256            let body = self.bytes().await?;
257            let decoded = T::decode(body)?;
258            Ok(decoded)
259        }
260    }
261}
262
263#[cfg(test)]
264mod tests {
265    use crate::configuration::SignalServers;
266    use bytes::{Buf, Bytes};
267
268    #[test]
269    fn create_clients() {
270        let environments = &[SignalServers::Staging, SignalServers::Production];
271
272        for env in environments {
273            let _ =
274                super::PushService::new(*env, None, "libsignal-service test");
275        }
276    }
277
278    #[test]
279    fn serde_json_from_empty_reader() {
280        // This fails, so we have handle empty response body separately in HyperPushService::json()
281        let bytes: Bytes = "".into();
282        assert!(
283            serde_json::from_reader::<bytes::buf::Reader<Bytes>, String>(
284                bytes.reader()
285            )
286            .is_err()
287        );
288    }
289
290    #[test]
291    fn serde_json_form_empty_vec() {
292        // If we're trying to send and empty payload, serde_json must be able to make a Vec out of it
293        assert!(serde_json::to_vec(b"").is_ok());
294    }
295}
libsignal_service/push_service/mod.rs

libsignal_service/push_service/
mod.rs
