fix: Sign payloads before encoding

relay-server/src/actors/processor.rs

@@ -1319,7 +1319,7 @@ impl EnvelopeProcessorService {
 
     fn handle_encode_envelope(&self, message: EncodeEnvelope) {
         let mut request = message.request;
-        match encode_payload(request.envelope_body, request.http_encoding) {
+        match encode_payload(&request.envelope_body, request.http_encoding) {
             Err(e) => {
                 request
                     .response_sender
@@ -1534,11 +1534,9 @@ impl EnvelopeProcessorService {
             return;
         }
 
-        let (payload, quantities) = partition.take_parts();
-
-        // TODO: Benchmark streamed encoding.
+        let (unencoded, quantities) = partition.take_parts();
         let http_encoding = self.inner.config.http_encoding();
-        let payload = match encode_payload(payload, http_encoding) {
+        let encoded = match encode_payload(&unencoded, http_encoding) {
             Ok(payload) => payload,
             Err(error) => {
                 let error = &error as &dyn std::error::Error;
@@ -1549,10 +1547,11 @@ impl EnvelopeProcessorService {
 
         let request = SendMetricsRequest {
             partition_key: key.map(|k| k.to_string()),
-            payload,
+            unencoded,
+            encoded,
+            http_encoding,
             quantities,
             outcome_aggregator: self.inner.outcome_aggregator.clone(),
-            http_encoding,
         };
 
         self.inner.upstream_relay.send(SendRequest(request));
@@ -1733,9 +1732,9 @@ impl Service for EnvelopeProcessorService {
     }
 }
 
-fn encode_payload(body: Bytes, http_encoding: HttpEncoding) -> Result<Bytes, std::io::Error> {
+fn encode_payload(body: &Bytes, http_encoding: HttpEncoding) -> Result<Bytes, std::io::Error> {
     let envelope_body: Vec<u8> = match http_encoding {
-        HttpEncoding::Identity => return Ok(body),
+        HttpEncoding::Identity => return Ok(body.clone()),
         HttpEncoding::Deflate => {
             let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
             encoder.write_all(body.as_ref())?;
@@ -1849,8 +1848,10 @@ impl<'a> Partition<'a> {
 struct SendMetricsRequest {
     /// If the partition key is set, the request is marked with `X-Sentry-Relay-Shard`.
     partition_key: Option<String>,
+    /// Serialized metric buckets without encoding applied, used for signing.
+    unencoded: Bytes,
     /// Serialized metric buckets with the stated HTTP encoding applied.
-    payload: Bytes,
+    encoded: Bytes,
     /// Encoding (compression) of the payload.
     http_encoding: HttpEncoding,
     /// Information about the metric quantities in the payload for outcomes.
@@ -1864,8 +1865,8 @@ impl UpstreamRequest for SendMetricsRequest {
         true
     }
 
-    fn sign(&self) -> bool {
-        true
+    fn sign(&mut self) -> Option<Bytes> {
+        Some(self.unencoded.clone())
     }
 
     fn method(&self) -> reqwest::Method {
@@ -1885,7 +1886,7 @@ impl UpstreamRequest for SendMetricsRequest {
             .content_encoding(self.http_encoding)
             .header_opt("X-Sentry-Relay-Shard", self.partition_key.as_ref())
             .header(header::CONTENT_TYPE, b"application/json")
-            .body(self.payload.clone());
+            .body(self.encoded.clone());
 
         Ok(())
     }

relay-server/src/actors/upstream.rs

@@ -299,12 +299,16 @@ pub trait UpstreamRequest: Send + Sync + fmt::Debug {
 
     /// Add the `X-Sentry-Relay-Signature` header to the outgoing request.
     ///
+    /// When no signature should be added, this method should return `None`. Otherwise, this method
+    /// should return the payload to sign. For requests with content encoding, this must be the
+    /// **uncompressed** payload.
+    ///
     /// This requires configuration of the Relay's credentials. If the credentials are not
     /// configured, the request will fail with [`UpstreamRequestError::NoCredentials`].
     ///
-    /// Defaults to `false`.
-    fn sign(&self) -> bool {
-        false
+    /// Defaults to `None`.
+    fn sign(&mut self) -> Option<Bytes> {
+        None
     }
 
     /// Returns the name of the logical route.
@@ -420,6 +424,16 @@ where
             sender,
         }
     }
+
+    /// Memoize the serialized body for retries and signing.
+    fn body(&mut self) -> Result<Bytes, HttpError> {
+        let body = match self.body {
+            Some(ref body) => body,
+            None => self.body.insert(serde_json::to_vec(&self.query)?.into()),
+        };
+
+        Ok(body.clone())
+    }
 }
 
 impl<T> UpstreamRequest for UpstreamQueryRequest<T>
@@ -442,8 +456,11 @@ where
         true
     }
 
-    fn sign(&self) -> bool {
-        true
+    fn sign(&mut self) -> Option<Bytes> {
+        // Computing the body is practically infallible since we're serializing standard structures
+        // into a string. Even if it fails, `sign` is called after `build` and the error will be
+        // reported there.
+        self.body().ok()
     }
 
     fn method(&self) -> Method {
@@ -465,11 +482,7 @@ where
     }
 
     fn build(&mut self, builder: &mut RequestBuilder) -> Result<(), HttpError> {
-        // Memoize the serialized body for retries.
-        let body = match self.body {
-            Some(ref body) => body,
-            None => self.body.insert(serde_json::to_vec(&self.query)?.into()),
-        };
+        let body = self.body()?;
 
         relay_statsd::metric!(
             histogram(RelayHistograms::UpstreamQueryBodySize) = body.len() as u64
@@ -761,22 +774,21 @@ impl SharedClient {
             let mut builder = RequestBuilder::reqwest(self.reqwest.request(request.method(), url));
             builder.header("Host", host_header.as_bytes());
 
-            if request.set_relay_id() || request.sign() {
+            if request.set_relay_id() {
                 if let Some(credentials) = self.config.credentials() {
                     builder.header("X-Sentry-Relay-Id", credentials.id.to_string());
                 }
             }
 
             request.build(&mut builder)?;
 
-            if request.sign() {
+            if let Some(payload) = request.sign() {
                 let credentials = self
                     .config
                     .credentials()
                     .ok_or(UpstreamRequestError::NoCredentials)?;
 
-                let body = builder.get_body().unwrap_or_default();
-                let signature = credentials.secret_key.sign(body);
+                let signature = credentials.secret_key.sign(&payload);
                 builder.header("X-Sentry-Relay-Signature", signature.as_bytes());
             }
 

relay-server/src/http.rs

@@ -46,28 +46,22 @@ pub struct Request(pub reqwest::Request);
 
 pub struct RequestBuilder {
     builder: Option<reqwest::RequestBuilder>,
-    body: Option<Bytes>,
 }
 
 impl RequestBuilder {
     pub fn reqwest(builder: reqwest::RequestBuilder) -> Self {
         RequestBuilder {
             builder: Some(builder),
-            body: None,
         }
     }
 
     pub fn finish(self) -> Result<Request, HttpError> {
-        let mut builder = self.builder.unwrap();
-        if let Some(body) = self.body {
-            builder = builder.body(body);
-        }
-        Ok(Request(builder.build()?))
+        Ok(Request(self.builder.unwrap().build()?))
     }
 
     fn build<F>(&mut self, f: F) -> &mut Self
     where
-        F: FnMut(reqwest::RequestBuilder) -> reqwest::RequestBuilder,
+        F: FnOnce(reqwest::RequestBuilder) -> reqwest::RequestBuilder,
     {
         self.builder = self.builder.take().map(f);
         self
@@ -98,12 +92,7 @@ impl RequestBuilder {
     }
 
     pub fn body(&mut self, body: Bytes) -> &mut Self {
-        self.body = Some(body);
-        self
-    }
-
-    pub fn get_body(&self) -> Option<&[u8]> {
-        self.body.as_deref()
+        self.build(|builder| builder.body(body))
     }
 }