compression: handle compression flag but no header (#763)

Co-authored-by: Quentin Perez <qperez42@gmail.com>
Co-authored-by: Lucio Franco <luciofranco14@gmail.com>

tests/compression/src/compressing_request.rs

@@ -87,3 +87,44 @@ async fn client_enabled_server_disabled() {
         "identity"
     );
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_mark_compressed_without_header_server_enabled() {
+    let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
+
+    let svc = test_server::TestServer::new(Svc::default()).accept_gzip();
+
+    tokio::spawn({
+        async move {
+            Server::builder()
+                .add_service(svc)
+                .serve_with_incoming(futures::stream::iter(vec![Ok::<_, std::io::Error>(
+                    MockStream(server),
+                )]))
+                .await
+                .unwrap();
+        }
+    });
+
+    let mut client = test_client::TestClient::with_interceptor(
+        mock_io_channel(client).await,
+        move |mut req: Request<()>| {
+            req.metadata_mut().remove("grpc-encoding");
+            Ok(req)
+        },
+    )
+    .send_gzip();
+
+    let status = client
+        .compress_input_unary(SomeData {
+            data: [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+        })
+        .await
+        .unwrap_err();
+
+    assert_eq!(status.code(), tonic::Code::Internal);
+    assert_eq!(
+        status.message(),
+        "protocol error: received message with compressed-flag but no grpc-encoding was specified"
+    );
+}

tonic/src/codec/decode.rs

@@ -205,9 +205,20 @@ impl<T> Streaming<T> {
             let is_compressed = match self.buf.get_u8() {
                 0 => false,
                 1 => {
-                    if cfg!(feature = "compression") {
-                        true
-                    } else {
+                    #[cfg(feature = "compression")]
+                    {
+                        if self.encoding.is_some() {
+                            true
+                        } else {
+                            // https://grpc.github.io/grpc/core/md_doc_compression.html
+                            // An ill-constructed message with its Compressed-Flag bit set but lacking a grpc-encoding
+                            // entry different from identity in its metadata MUST fail with INTERNAL status,
+                            // its associated description indicating the invalid Compressed-Flag condition.
+                            return Err(Status::new(Code::Internal, "protocol error: received message with compressed-flag but no grpc-encoding was specified"));
+                        }
+                    }
+                    #[cfg(not(feature = "compression"))]
+                    {
                         return Err(Status::new(
                             Code::Unimplemented,
                             "Message compressed, compression support not enabled.".to_string(),
@@ -250,6 +261,7 @@ impl<T> Streaming<T> {
 
                     if let Err(err) = decompress(
                         self.encoding.unwrap_or_else(|| {
+                            // SAFETY: The check while in State::ReadHeader would already have returned Code::Internal
                             unreachable!("message was compressed but `Streaming.encoding` was `None`. This is a bug in Tonic. Please file an issue")
                         }),
                         &mut self.buf,