wire-api: strip framing headers in generic streaming response renderers

changelog.d/5-internal/wire-api-streaming-framing-headers

@@ -0,0 +1 @@
+wire-api: the generic streaming response renderers (`MultiVerb`'s `IsWaiBody (SourceIO ByteString)` instance and `LowLevelStream`) now drop the `Content-Length`/`Transfer-Encoding` headers before handing the response to Warp. Warp re-frames a streamed body itself (chunked); forwarding a stale framing header would let Warp serve the body under a declared length that need not match the streamed bytes, desynchronising the caller's keep-alive connection (the same failure mode fixed in the federator's `streamingResponseToWai`). No endpoint currently attaches such a header to a streaming response, so this is defense-in-depth guarding against the bug class recurring.

libs/wire-api/default.nix

@@ -275,6 +275,7 @@ mkDerivation {
     saml2-web-sso
     schema-profunctor
     servant
+    servant-client-core
     servant-server
     string-conversions
     tasty
@@ -289,6 +290,7 @@ mkDerivation {
     uuid
     vector
     wai
+    wai-extra
     wire-message-proto-lens
   ];
   license = lib.meta.getLicenseFromSpdxId "AGPL-3.0-only";

libs/wire-api/src/Wire/API/Routes/LowLevelStream.hs

@@ -31,6 +31,7 @@ import GHC.TypeLits
 import Imports
 import Network.HTTP.Media qualified as HTTP
 import Network.HTTP.Types
+import Network.HTTP.Types.Header (hTransferEncoding)
 import Network.Wai
 import Servant.API
 import Servant.API.ContentTypes
@@ -45,6 +46,15 @@ import Wire.API.Routes.Version
 -- makes it possible to add finalisation logic to the streaming action.
 type LowLevelStreamingBody = Codensity IO StreamingBody
 
+-- | Drop the hop-by-hop framing headers ('Content-Length',
+-- 'Transfer-Encoding') from a response header list. The streamed body is
+-- re-framed (chunked) by Warp, so forwarding a type-level 'Content-Length'
+-- would let a length mismatch desync the caller's keep-alive connection. See
+-- also @Federator.Response.streamingResponseToWai@.
+stripFramingHeaders :: [(HeaderName, ByteString)] -> [(HeaderName, ByteString)]
+stripFramingHeaders =
+  Imports.filter (\(hName, _) -> hName /= hContentLength && hName /= hTransferEncoding)
+
 -- FUTUREWORK: make it possible to generate headers at runtime
 data LowLevelStream method status (headers :: [(Symbol, Symbol)]) desc ctype
 
@@ -94,7 +104,7 @@ instance
                 Left e -> respond $ FailFatal e
                 Right getStreamingBody -> lowerCodensity $ do
                   body <- getStreamingBody
-                  let resp = responseStream status (contentHeader : extraHeaders) body
+                  let resp = responseStream status (stripFramingHeaders (contentHeader : extraHeaders)) body
                   lift $ respond $ Route resp
             Fail e -> respond $ Fail e
             FailFatal e -> respond $ FailFatal e

libs/wire-api/src/Wire/API/Routes/MultiVerb.hs

@@ -40,6 +40,7 @@ module Wire.API.Routes.MultiVerb
     GenericAsUnion (..),
     ResponseType,
     IsResponse (..),
+    IsWaiBody (..),
     IsSwaggerResponse (..),
     IsSwaggerResponseList (..),
     simpleResponseSwagger,
@@ -74,6 +75,7 @@ import Generics.SOP as GSOP
 import Imports
 import Network.HTTP.Media qualified as M
 import Network.HTTP.Types (hContentType)
+import Network.HTTP.Types.Header (hTransferEncoding)
 import Network.HTTP.Types qualified as HTTP
 import Network.HTTP.Types.Status
 import Network.Wai qualified as Wai
@@ -767,6 +769,20 @@ instance
               . toList
       refResps = S.Inline . addMime <$> resps
 
+-- | Drop the hop-by-hop framing headers ('Content-Length',
+-- 'Transfer-Encoding') from a response header list.
+--
+-- Streaming responses ('Wai.responseStream') are re-framed (chunked) by Warp.
+-- Forwarding an upstream 'Content-Length' makes Warp serve the streamed body
+-- under that declared length instead; any mismatch between the declared length
+-- and the bytes actually streamed then desynchronises the caller's keep-alive
+-- connection (the client reads past the response boundary into the next
+-- response). Stripping the framing headers lets Warp frame exactly what is
+-- streamed. See also @Federator.Response.streamingResponseToWai@.
+stripFramingHeaders :: [HTTP.Header] -> [HTTP.Header]
+stripFramingHeaders =
+  filter (\(hName, _) -> hName /= HTTP.hContentLength && hName /= hTransferEncoding)
+
 class (Typeable a) => IsWaiBody a where
   responseToWai :: ResponseF a -> Wai.Response
 
@@ -788,7 +804,7 @@ instance IsWaiBody (SourceIO ByteString) where
   responseToWai r =
     Wai.responseStream
       (responseStatusCode r)
-      (toList (responseHeaders r))
+      (stripFramingHeaders (toList (responseHeaders r)))
       $ \output flush -> do
         foreach
           (const (pure ()))

libs/wire-api/test/unit/Test/Wire/API/Routes/Streaming.hs

@@ -0,0 +1,136 @@
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2025 Wire Swiss GmbH <opensource@wire.com>
+--
+-- This program is free software: you can redistribute it and/or modify it under
+-- the terms of the GNU Affero General Public License as published by the Free
+-- Software Foundation, either version 3 of the License, or (at your option) any
+-- later version.
+--
+-- This program is distributed in the hope that it will be useful, but WITHOUT
+-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+-- details.
+--
+-- You should have received a copy of the GNU Affero General Public License along
+-- with this program. If not, see <https://www.gnu.org/licenses/>.
+
+-- | Regression tests for the two generic streaming response renderers in
+-- @wire-api@: the 'MultiVerb' @SourceIO ByteString@ instance and the
+-- 'LowLevelStream' server.
+--
+-- Both re-frame a streamed body via Warp (chunked). If they forward an upstream
+-- (or type-level) @Content-Length@/@Transfer-Encoding@ header, Warp honours it
+-- verbatim, and any mismatch between the declared length and the streamed bytes
+-- desynchronises the caller's keep-alive HTTP/1.1 connection — the exact bug
+-- fixed in @Federator.Response.streamingResponseToWai@. These tests assert that
+-- neither renderer emits those framing headers, so the bug class cannot recur
+-- here.
+--
+-- We inspect the emitted 'Network.Wai.Response' via 'Network.Wai.Test' (which
+-- captures exactly the header list the response carries); the desync behaviour
+-- of Warp itself is exercised end-to-end in the federator test suite.
+module Test.Wire.API.Routes.Streaming (tests) where
+
+import Data.ByteString.Builder (byteString)
+import Data.ByteString.Lazy qualified as LBS
+import Data.Proxy (Proxy (..))
+import Data.Sequence qualified as Seq
+import Imports
+import Network.HTTP.Types qualified as HTTP
+import Network.HTTP.Types.Header (hTransferEncoding)
+import Network.HTTP.Types.Method (StdMethod (GET))
+import Network.Wai (Application, requestHeaders)
+import Network.Wai.Test
+import Servant.API (OctetStream, SourceIO, type (:>))
+import Servant.Client.Core (ResponseF (..))
+import Servant.Server (Server, serve)
+import Servant.Types.SourceT (source)
+import Test.Tasty
+import Test.Tasty.HUnit
+import Wire.API.Routes.LowLevelStream (LowLevelStream, LowLevelStreamingBody)
+import Wire.API.Routes.MultiVerb (responseToWai)
+
+tests :: TestTree
+tests =
+  testGroup
+    "Streaming response renderers drop framing headers"
+    [ testCase
+        "MultiVerb SourceIO renderer strips forwarded Content-Length/Transfer-Encoding"
+        multiVerbStreamingStripsFramingHeaders,
+      testCase
+        "LowLevelStream server strips type-level Content-Length/Transfer-Encoding"
+        lowLevelStreamStripsFramingHeaders
+    ]
+
+-- | Assert the common invariant on a served streaming response: framing headers
+-- gone, unrelated headers kept, body intact.
+assertNoFramingHeaders :: SResponse -> ByteString -> IO ()
+assertNoFramingHeaders sresp expectedBody = do
+  let hdrs = simpleHeaders sresp
+  assertEqual "status" HTTP.status200 (simpleStatus sresp)
+  assertBool
+    ("Content-Length must be stripped, but headers were: " <> show hdrs)
+    (isNothing (lookup HTTP.hContentLength hdrs))
+  assertBool
+    ("Transfer-Encoding must be stripped, but headers were: " <> show hdrs)
+    (isNothing (lookup hTransferEncoding hdrs))
+  assertEqual "Content-Type is preserved" (Just "application/octet-stream") (lookup HTTP.hContentType hdrs)
+  assertEqual "unrelated header is preserved" (Just "yes") (lookup "X-Keep" hdrs)
+  assertEqual "body is streamed intact" expectedBody (LBS.toStrict (simpleBody sresp))
+
+--------------------------------------------------------------------------------
+-- MultiVerb: the generic 'IsWaiBody (SourceIO ByteString)' renderer.
+
+-- | An upstream streaming response carrying framing headers (as the outward
+-- HTTP/2 federation response does). This is what would flow into the MultiVerb
+-- renderer if a streaming endpoint ever surfaced an upstream length.
+multiVerbUpstream :: ResponseF (SourceIO ByteString)
+multiVerbUpstream =
+  Response
+    { responseStatusCode = HTTP.status200,
+      responseHeaders =
+        Seq.fromList
+          [ (HTTP.hContentType, "application/octet-stream"),
+            (HTTP.hContentLength, "999"),
+            (hTransferEncoding, "chunked"),
+            ("X-Keep", "yes")
+          ],
+      responseHttpVersion = HTTP.http11,
+      responseBody = source ["ab", "cd"]
+    }
+
+multiVerbStreamingStripsFramingHeaders :: Assertion
+multiVerbStreamingStripsFramingHeaders = do
+  let app :: Application
+      app _req respond = respond (responseToWai multiVerbUpstream)
+  sresp <- runSession (request defaultRequest) app
+  assertNoFramingHeaders sresp "abcd"
+
+--------------------------------------------------------------------------------
+-- LowLevelStream: the servant streaming server.
+
+type StreamAPI =
+  "s"
+    :> LowLevelStream
+         'GET
+         200
+         '[ '("Content-Length", "999"),
+            '("Transfer-Encoding", "chunked"),
+            '("X-Keep", "yes")
+          ]
+         "test stream"
+         OctetStream
+
+streamServer :: Server StreamAPI
+streamServer = pure streamingBody
+  where
+    streamingBody :: LowLevelStreamingBody
+    streamingBody = pure (\write flush -> write (byteString "hello") *> flush)
+
+lowLevelStreamStripsFramingHeaders :: Assertion
+lowLevelStreamStripsFramingHeaders = do
+  let app = serve (Proxy @StreamAPI) streamServer
+      req = (setPath defaultRequest "/s") {requestHeaders = [(HTTP.hAccept, "*/*")]}
+  sresp <- runSession (request req) app
+  assertNoFramingHeaders sresp "hello"

libs/wire-api/test/unit/Test/Wire/API/Run.hs

@@ -34,6 +34,7 @@ import Test.Wire.API.Roundtrip.MLS qualified as Roundtrip.MLS
 import Test.Wire.API.Roundtrip.PostgresMarshall as PostgresMarshall
 import Test.Wire.API.Routes qualified as Routes
 import Test.Wire.API.Routes.Version qualified as Routes.Version
+import Test.Wire.API.Routes.Streaming qualified as Routes.Streaming
 import Test.Wire.API.Routes.Version.Wai qualified as Routes.Version.Wai
 import Test.Wire.API.Swagger qualified as Swagger
 import Test.Wire.API.Team.Export qualified as Team.Export
@@ -62,6 +63,7 @@ main =
         Swagger.tests,
         Roundtrip.CSV.tests,
         Routes.tests,
+        Routes.Streaming.tests,
         Conversation.tests,
         MLS.tests,
         Group.tests,

libs/wire-api/wire-api.cabal

@@ -724,6 +724,7 @@ test-suite wire-api-tests
     Test.Wire.API.Roundtrip.MLS
     Test.Wire.API.Roundtrip.PostgresMarshall
     Test.Wire.API.Routes
+    Test.Wire.API.Routes.Streaming
     Test.Wire.API.Routes.Version
     Test.Wire.API.Routes.Version.Wai
     Test.Wire.API.Run
@@ -764,6 +765,7 @@ test-suite wire-api-tests
     , random
     , schema-profunctor
     , servant
+    , servant-client-core
     , servant-server
     , string-conversions
     , tasty
@@ -777,6 +779,7 @@ test-suite wire-api-tests
     , uuid
     , vector
     , wai
+    , wai-extra
     , wire-api
 
   ghc-options: