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::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            // Not a total timeout: attachments take as long as they take to transfer.
103            .read_timeout(Duration::from_secs(65))
104            .user_agent(user_agent.as_ref())
105            .http1_only()
106            .build()
107            .unwrap();
108
109        Self {
110            servers: env,
111            cfg,
112            credentials: credentials.and_then(|c| c.authorization()),
113            client,
114        }
115    }
116
117    #[tracing::instrument(skip(self), fields(endpoint = %endpoint))]
118    pub fn request(
119        &self,
120        method: Method,
121        endpoint: Endpoint,
122        auth_override: HttpAuthOverride,
123    ) -> Result<RequestBuilder, ServiceError> {
124        let url = endpoint.into_url(&self.cfg)?;
125        let mut builder = self.client.request(method, url);
126
127        builder = match auth_override {
128            HttpAuthOverride::NoOverride => {
129                if let Some(HttpAuth { username, password }) =
130                    self.credentials.as_ref()
131                {
132                    builder.basic_auth(username, Some(password))
133                } else {
134                    builder
135                }
136            },
137            HttpAuthOverride::Identified(HttpAuth { username, password }) => {
138                builder.basic_auth(username, Some(password))
139            },
140            HttpAuthOverride::Unidentified => builder,
141        };
142
143        Ok(builder)
144    }
145
146    pub async fn ws<C: WebSocketType>(
147        &mut self,
148        path: &str,
149        keepalive_path: &str,
150        additional_headers: &[(&'static str, &str)],
151        credentials: Option<ServiceCredentials>,
152    ) -> Result<SignalWebSocket<C>, ServiceError> {
153        let span = debug_span!("websocket");
154
155        let mut url = Endpoint::service(path).into_url(&self.cfg)?;
156        url.set_scheme("wss").expect("valid https base url");
157
158        let mut builder = self.client.get(url);
159        for (key, value) in additional_headers {
160            builder = builder.header(*key, *value);
161        }
162
163        if let Some(credentials) = credentials {
164            builder =
165                builder.basic_auth(credentials.login(), credentials.password);
166        }
167
168        let ws = builder
169            .upgrade()
170            .send()
171            .await?
172            .into_websocket()
173            .instrument(span.clone())
174            .await?;
175
176        let unidentified_push_service = PushService {
177            servers: self.servers,
178            cfg: self.cfg.clone(),
179            credentials: None,
180            client: self.client.clone(),
181        };
182        let (ws, task) = SignalWebSocket::new(
183            ws,
184            keepalive_path.to_owned(),
185            unidentified_push_service,
186        );
187        let task = task.instrument(span);
188        tokio::task::spawn(task);
189        Ok(ws)
190    }
191
192    pub(crate) async fn get_group(
193        &mut self,
194        credentials: HttpAuth,
195    ) -> Result<crate::proto::Group, ServiceError> {
196        self.request(
197            Method::GET,
198            Endpoint::storage("/v1/groups/"),
199            HttpAuthOverride::Identified(credentials),
200        )?
201        .send()
202        .await?
203        .service_error_for_status()
204        .await?
205        .protobuf()
206        .await
207    }
208}
209
210pub(crate) mod protobuf {
211    use async_trait::async_trait;
212    use prost::{EncodeError, Message};
213    use reqwest::{header, RequestBuilder, Response};
214
215    use super::ServiceError;
216
217    pub(crate) trait ProtobufRequestBuilderExt
218    where
219        Self: Sized,
220    {
221        /// Set the request payload encoded as protobuf.
222        /// Sets the `Content-Type` header to `application/x-protobuf`
223        #[allow(dead_code)]
224        fn protobuf<T: Message + Default>(
225            self,
226            value: T,
227        ) -> Result<Self, EncodeError>;
228    }
229
230    #[async_trait::async_trait]
231    pub(crate) trait ProtobufResponseExt {
232        /// Get the response body decoded from Protobuf
233        async fn protobuf<T>(self) -> Result<T, ServiceError>
234        where
235            T: prost::Message + Default;
236    }
237
238    impl ProtobufRequestBuilderExt for RequestBuilder {
239        fn protobuf<T: Message + Default>(
240            self,
241            value: T,
242        ) -> Result<Self, EncodeError> {
243            let mut buf = Vec::new();
244            value.encode(&mut buf)?;
245            let this =
246                self.header(header::CONTENT_TYPE, "application/x-protobuf");
247            Ok(this.body(buf))
248        }
249    }
250
251    #[async_trait]
252    impl ProtobufResponseExt for Response {
253        async fn protobuf<T>(self) -> Result<T, ServiceError>
254        where
255            T: Message + Default,
256        {
257            let body = self.bytes().await?;
258            let decoded = T::decode(body)?;
259            Ok(decoded)
260        }
261    }
262}
263
264#[cfg(test)]
265mod tests {
266    use crate::configuration::SignalServers;
267    use bytes::{Buf, Bytes};
268
269    #[test]
270    fn create_clients() {
271        let environments = &[SignalServers::Staging, SignalServers::Production];
272
273        for env in environments {
274            let _ =
275                super::PushService::new(*env, None, "libsignal-service test");
276        }
277    }
278
279    #[test]
280    fn serde_json_from_empty_reader() {
281        // This fails, so we have handle empty response body separately in HyperPushService::json()
282        let bytes: Bytes = "".into();
283        assert!(
284            serde_json::from_reader::<bytes::buf::Reader<Bytes>, String>(
285                bytes.reader()
286            )
287            .is_err()
288        );
289    }
290
291    #[test]
292    fn serde_json_form_empty_vec() {
293        // If we're trying to send and empty payload, serde_json must be able to make a Vec out of it
294        assert!(serde_json::to_vec(b"").is_ok());
295    }
296}
libsignal_service/push_service/mod.rs

libsignal_service/push_service/
mod.rs
