Add streaming signature for PutObject (#123)

Use streaming signature to avoid reading the body twice in PutObject
requests, where the body can be upto 5GIB.

Note that the body is signed only used when the connection is not
using TLS.

minio-hs.cabal

@@ -116,15 +116,18 @@ test-suite minio-hs-live-server-test
                      , TypeFamilies
   other-modules:       Lib.Prelude
                      , Network.Minio
-                     , Network.Minio.AdminAPI
                      , Network.Minio.API
+                     , Network.Minio.API.Test
                      , Network.Minio.APICommon
+                     , Network.Minio.AdminAPI
                      , Network.Minio.CopyObject
                      , Network.Minio.Data
                      , Network.Minio.Data.ByteString
                      , Network.Minio.Data.Crypto
                      , Network.Minio.Data.Time
                      , Network.Minio.Errors
+                     , Network.Minio.JsonParser
+                     , Network.Minio.JsonParser.Test
                      , Network.Minio.ListOps
                      , Network.Minio.PresignedOperations
                      , Network.Minio.PutObject
@@ -134,13 +137,10 @@ test-suite minio-hs-live-server-test
                      , Network.Minio.TestHelpers
                      , Network.Minio.Utils
                      , Network.Minio.Utils.Test
-                     , Network.Minio.API.Test
                      , Network.Minio.XmlGenerator
                      , Network.Minio.XmlGenerator.Test
                      , Network.Minio.XmlParser
                      , Network.Minio.XmlParser.Test
-                     , Network.Minio.JsonParser
-                     , Network.Minio.JsonParser.Test
   build-depends:       base >= 4.7 && < 5
                      , minio-hs
                      , protolude >= 0.1.6
@@ -243,15 +243,18 @@ test-suite minio-hs-test
                      , TypeFamilies
   other-modules:       Lib.Prelude
                      , Network.Minio
-                     , Network.Minio.AdminAPI
                      , Network.Minio.API
+                     , Network.Minio.API.Test
                      , Network.Minio.APICommon
+                     , Network.Minio.AdminAPI
+                     , Network.Minio.CopyObject
                      , Network.Minio.Data
                      , Network.Minio.Data.ByteString
                      , Network.Minio.Data.Crypto
                      , Network.Minio.Data.Time
-                     , Network.Minio.CopyObject
                      , Network.Minio.Errors
+                     , Network.Minio.JsonParser
+                     , Network.Minio.JsonParser.Test
                      , Network.Minio.ListOps
                      , Network.Minio.PresignedOperations
                      , Network.Minio.PutObject
@@ -261,13 +264,10 @@ test-suite minio-hs-test
                      , Network.Minio.TestHelpers
                      , Network.Minio.Utils
                      , Network.Minio.Utils.Test
-                     , Network.Minio.API.Test
                      , Network.Minio.XmlGenerator
                      , Network.Minio.XmlGenerator.Test
                      , Network.Minio.XmlParser
                      , Network.Minio.XmlParser.Test
-                     , Network.Minio.JsonParser
-                     , Network.Minio.JsonParser.Test
 
 source-repository head
   type:     git

src/Network/Minio/API.hs

@@ -37,7 +37,6 @@ import qualified Data.Conduit              as C
 import qualified Data.Map                  as Map
 import qualified Data.Text                 as T
 import qualified Data.Time.Clock           as Time
-
 import           Network.HTTP.Conduit      (Response)
 import qualified Network.HTTP.Conduit      as NC
 import qualified Network.HTTP.Types        as HT
@@ -75,73 +74,97 @@ discoverRegion ri = runMaybeT $ do
             return l
         ) return regionMay
 
+getRegion :: S3ReqInfo -> Minio (Maybe Region)
+getRegion ri = do
+    ci <- asks mcConnInfo
+
+         -- getService/makeBucket/getLocation -- don't need location
+    if | not $ riNeedsLocation ri ->
+             return $ Just $ connectRegion ci
+
+         -- if autodiscovery of location is disabled by user
+       | not $ connectAutoDiscoverRegion ci ->
+                     return $ Just $ connectRegion ci
+
+         -- discover the region for the request
+       | otherwise -> discoverRegion ri
+
+getRegionHost :: Region -> Minio Text
+getRegionHost r = do
+    ci <- asks mcConnInfo
+
+    if "amazonaws.com" `T.isSuffixOf` connectHost ci
+    then maybe (throwIO $ MErrVRegionNotSupported r)
+         return (Map.lookup r awsRegionMap)
+    else return $ connectHost ci
 
 buildRequest :: S3ReqInfo -> Minio NC.Request
 buildRequest ri = do
-  maybe (return ()) checkBucketNameValidity $ riBucket ri
+    maybe (return ()) checkBucketNameValidity $ riBucket ri
-  maybe (return ()) checkObjectNameValidity $ riObject ri
+    maybe (return ()) checkObjectNameValidity $ riObject ri
 
-  ci <- asks mcConnInfo
+    ci <- asks mcConnInfo
 
-               -- getService/makeBucket/getLocation -- don't need
+    regionMay <- getRegion ri
-               -- location
-  region <- if | not $ riNeedsLocation ri ->
-                   return $ Just $ connectRegion ci
 
-               -- if autodiscovery of location is disabled by user
+    regionHost <- maybe (return $ connectHost ci) getRegionHost regionMay
-               | not $ connectAutoDiscoverRegion ci ->
-                   return $ Just $ connectRegion ci
 
-               -- discover the region for the request
+    let ri' = ri { riHeaders = hostHeader : riHeaders ri
-               | otherwise -> discoverRegion ri
+                 , riRegion = regionMay
-
-  regionHost <- case region of
-    Nothing ->  return $ connectHost ci
-    Just r -> if "amazonaws.com" `T.isSuffixOf` connectHost ci
-              then maybe
-                   (throwIO $ MErrVRegionNotSupported r)
-                   return
-                   (Map.lookup r awsRegionMap)
-              else return $ connectHost ci
-
-  sha256Hash <- if | connectIsSecure ci ->
-                       -- if secure connection
-                       return "UNSIGNED-PAYLOAD"
-
-                   -- otherwise compute sha256
-                   | otherwise -> getPayloadSHA256Hash (riPayload ri)
-
-  timeStamp <- liftIO Time.getCurrentTime
-
-  let hostHeader = (hHost, getHostAddr ci)
-      newRi = ri { riPayloadHash = Just sha256Hash
-                 , riHeaders = hostHeader
-                             : sha256Header sha256Hash
-                             : riHeaders ri
-                 , riRegion = region
                  }
-      newCi = ci { connectHost = regionHost }
+        ci' = ci { connectHost = regionHost }
-      signReq = toRequest newCi newRi
+        hostHeader = (hHost, getHostAddr ci')
-      sp = SignParams (connectAccessKey ci) (connectSecretKey ci)
-           timeStamp (riRegion newRi) Nothing (riPayloadHash newRi)
-  let signHeaders = signV4 sp signReq
 
-  -- Update signReq with Authorization header containing v4 signature
+        -- Does not contain body and auth info.
-  return signReq {
+        baseRequest = NC.defaultRequest
-      NC.requestHeaders = riHeaders newRi ++ mkHeaderFromPairs signHeaders
+                      { NC.method = riMethod ri'
-      }
+                      , NC.secure = connectIsSecure ci'
-  where
+                      , NC.host = encodeUtf8 $ connectHost ci'
-    toRequest :: ConnectInfo -> S3ReqInfo -> NC.Request
+                      , NC.port = connectPort ci'
-    toRequest ci s3Req = NC.defaultRequest {
+                      , NC.path = getS3Path (riBucket ri') (riObject ri')
-          NC.method = riMethod s3Req
+                      , NC.requestHeaders = riHeaders ri'
-        , NC.secure = connectIsSecure ci
+                      , NC.queryString = HT.renderQuery False $ riQueryParams ri'
-        , NC.host = encodeUtf8 $ connectHost ci
+                      }
-        , NC.port = connectPort ci
+
-        , NC.path = getS3Path (riBucket s3Req) (riObject s3Req)
+    timeStamp <- liftIO Time.getCurrentTime
-        , NC.requestHeaders = riHeaders s3Req
+
-        , NC.queryString = HT.renderQuery False $ riQueryParams s3Req
+    let sp = SignParams (connectAccessKey ci') (connectSecretKey ci')
-        , NC.requestBody = getRequestBody (riPayload s3Req)
+             timeStamp (riRegion ri') Nothing Nothing
-        }
+
+    -- Cases to handle:
+    --
+    -- 1. Connection is secure: use unsigned payload
+    --
+    -- 2. Insecure connection, streaming signature is enabled via use of
+    --    conduit payload: use streaming signature for request.
+    --
+    -- 3. Insecure connection, non-conduit payload: compute payload
+    -- sha256hash, buffer request in memory and perform request.
+
+         -- case 2 from above.
+    if | isStreamingPayload (riPayload ri') &&
+         (not $ connectIsSecure ci') -> do
+             (pLen, pSrc) <- case riPayload ri of
+                               PayloadC l src -> return (l, src)
+                               _              -> throwIO MErrVUnexpectedPayload
+             let reqFn = signV4Stream pLen sp baseRequest
+             return $ reqFn pSrc
+
+       | otherwise -> do
+                         -- case 1 described above.
+             sp' <- if | connectIsSecure ci' -> return sp
+                         -- case 3 described above.
+                       | otherwise -> do
+                             pHash <- getPayloadSHA256Hash $ riPayload ri'
+                             return $ sp { spPayloadHash = Just pHash }
+
+             let signHeaders = signV4 sp' baseRequest
+             return $ baseRequest
+                      { NC.requestHeaders =
+                            NC.requestHeaders baseRequest ++
+                            mkHeaderFromPairs signHeaders
+                      , NC.requestBody = getRequestBody (riPayload ri')
+                      }
 
 
 retryAPIRequest :: Minio a -> Minio a

src/Network/Minio/APICommon.hs

@@ -16,6 +16,9 @@
 
 module Network.Minio.APICommon where
 
+import qualified Conduit                   as C
+import qualified Data.ByteString           as BS
+import qualified Data.ByteString.Lazy      as LB
 import           Data.Conduit.Binary       (sourceHandleRange)
 import qualified Network.HTTP.Conduit      as NC
 import qualified Network.HTTP.Types        as HT
@@ -24,16 +27,20 @@ import           Lib.Prelude
 
 import           Network.Minio.Data
 import           Network.Minio.Data.Crypto
+import           Network.Minio.Errors
 
 sha256Header :: ByteString -> HT.Header
 sha256Header = ("x-amz-content-sha256", )
 
-getPayloadSHA256Hash :: (MonadIO m) => Payload -> m ByteString
+-- | This function throws an error if the payload is a conduit (as it
+-- will not be possible to re-read the conduit after it is consumed).
+getPayloadSHA256Hash :: Payload -> Minio ByteString
 getPayloadSHA256Hash (PayloadBS bs) = return $ hashSHA256 bs
 getPayloadSHA256Hash (PayloadH h off size) = hashSHA256FromSource $
   sourceHandleRange h
     (return . fromIntegral $ off)
     (return . fromIntegral $ size)
+getPayloadSHA256Hash (PayloadC _ _) = throwIO MErrVUnexpectedPayload
 
 getRequestBody :: Payload -> NC.RequestBody
 getRequestBody (PayloadBS bs) = NC.RequestBodyBS bs
@@ -42,3 +49,20 @@ getRequestBody (PayloadH h off size) =
     sourceHandleRange h
       (return . fromIntegral $ off)
       (return . fromIntegral $ size)
+getRequestBody (PayloadC n src) = NC.requestBodySource n src
+
+mkStreamingPayload :: Payload -> Payload
+mkStreamingPayload payload =
+  case payload of
+    PayloadBS bs ->
+        PayloadC (fromIntegral $ BS.length bs)
+        (C.sourceLazy $ LB.fromStrict bs)
+    PayloadH h off len ->
+        PayloadC len $ sourceHandleRange h
+        (return . fromIntegral $ off)
+        (return . fromIntegral $ len)
+    _ -> payload
+
+isStreamingPayload :: Payload -> Bool
+isStreamingPayload (PayloadC _ _) = True
+isStreamingPayload _              = False

src/Network/Minio/Data.hs

@@ -20,6 +20,7 @@
 {-# LANGUAGE TypeFamilies               #-}
 module Network.Minio.Data where
 
+import qualified Conduit                      as C
 import           Control.Concurrent.MVar      (MVar)
 import qualified Control.Concurrent.MVar      as M
 import           Control.Monad.IO.Unlift      (MonadUnliftIO, UnliftIO (..),
@@ -883,10 +884,10 @@ type Stats = Progress
 
 -- | Represents different kinds of payload that are used with S3 API
 -- requests.
-data Payload = PayloadBS ByteString
+data Payload
-             | PayloadH Handle
+    = PayloadBS ByteString
-                        Int64 -- offset
+    | PayloadH Handle Int64 Int64 -- file handle, offset and length
-                        Int64 -- size
+    | PayloadC Int64 (C.ConduitT () ByteString (ResourceT IO) ()) -- length and byte source
 
 defaultPayload :: Payload
 defaultPayload = PayloadBS ""

src/Network/Minio/Errors.hs

@@ -1,5 +1,5 @@
 --
--- MinIO Haskell SDK, (C) 2017, 2018 MinIO, Inc.
+-- MinIO Haskell SDK, (C) 2017-2019 MinIO, Inc.
 --
 -- Licensed under the Apache License, Version 2.0 (the "License");
 -- you may not use this file except in compliance with the License.
@@ -42,6 +42,8 @@ data MErrV = MErrVSinglePUTSizeExceeded Int64
            | MErrVInvalidHealPath
            | MErrVMissingCredentials
            | MErrVInvalidEncryptionKeyLength
+           | MErrVStreamingBodyUnexpectedEOF
+           | MErrVUnexpectedPayload
   deriving (Show, Eq)
 
 instance Exception MErrV

src/Network/Minio/PutObject.hs

@@ -1,5 +1,5 @@
 --
--- MinIO Haskell SDK, (C) 2017 MinIO, Inc.
+-- MinIO Haskell SDK, (C) 2017-2019 MinIO, Inc.
 --
 -- Licensed under the Apache License, Version 2.0 (the "License");
 -- you may not use this file except in compliance with the License.
@@ -23,13 +23,14 @@ module Network.Minio.PutObject
 
 
 import           Conduit                  (takeC)
+import qualified Conduit                  as C
 import qualified Data.ByteString.Lazy     as LBS
-import qualified Data.Conduit             as C
 import qualified Data.Conduit.Binary      as CB
 import qualified Data.Conduit.Combinators as CC
 import qualified Data.Conduit.List        as CL
 import qualified Data.List                as List
 
+
 import           Lib.Prelude
 
 import           Network.Minio.Data
@@ -63,8 +64,8 @@ putObjectInternal :: Bucket -> Object -> PutObjectOptions
                   -> ObjectData Minio -> Minio ETag
 putObjectInternal b o opts (ODStream src sizeMay) = do
   case sizeMay of
-    -- unable to get size, so assume non-seekable file and max-object size
+    -- unable to get size, so assume non-seekable file
-    Nothing -> sequentialMultipartUpload b o opts (Just maxObjectSize) src
+    Nothing -> sequentialMultipartUpload b o opts Nothing src
 
     -- got file size, so check for single/multipart upload
     Just size ->
@@ -85,9 +86,8 @@ putObjectInternal b o opts (ODFile fp sizeMay) = do
   let finalSizeMay = listToMaybe $ catMaybes [sizeMay, handleSizeMay]
 
   case finalSizeMay of
-    -- unable to get size, so assume non-seekable file and max-object size
+    -- unable to get size, so assume non-seekable file
-    Nothing -> sequentialMultipartUpload b o opts (Just maxObjectSize) $
+    Nothing -> sequentialMultipartUpload b o opts Nothing $ CB.sourceFile fp
-               CB.sourceFile fp
 
     -- got file size, so check for single/multipart upload
     Just size ->
@@ -138,7 +138,7 @@ sequentialMultipartUpload b o opts sizeMay src = do
       (pnums, _, sizes) = List.unzip3 partSizes
   uploadedParts <- C.runConduit
                  $ src
-              C..| chunkBSConduit sizes
+              C..| chunkBSConduit (map fromIntegral sizes)
               C..| CL.map PayloadBS
               C..| uploadPart' uploadId pnums
               C..| CC.sinkList

src/Network/Minio/S3API.hs

@@ -42,6 +42,7 @@ module Network.Minio.S3API
   ---------------------------------
   , putBucket
   , ETag
+  , maxSinglePutObjectSizeBytes
   , putObjectSingle'
   , putObjectSingle
   , copyObjectSingle
@@ -90,8 +91,8 @@ module Network.Minio.S3API
   , removeAllBucketNotification
   ) where
 
+import qualified Conduit                           as C
 import qualified Data.ByteString                   as BS
-import qualified Data.Conduit                      as C
 import qualified Data.Text                         as T
 import qualified Network.HTTP.Conduit              as NC
 import qualified Network.HTTP.Types                as HT
@@ -101,6 +102,7 @@ import           UnliftIO                          (Handler (Handler))
 import           Lib.Prelude
 
 import           Network.Minio.API
+import           Network.Minio.APICommon
 import           Network.Minio.Data
 import           Network.Minio.Errors
 import           Network.Minio.PresignedOperations
@@ -156,13 +158,13 @@ putObjectSingle' bucket object headers bs = do
   when (size > maxSinglePutObjectSizeBytes) $
     throwIO $ MErrVSinglePUTSizeExceeded size
 
-  -- content-length header is automatically set by library.
+  let payload = mkStreamingPayload $ PayloadBS bs
   resp <- executeRequest $
           defaultS3ReqInfo { riMethod = HT.methodPut
               , riBucket = Just bucket
               , riObject = Just object
               , riHeaders = headers
-              , riPayload = PayloadBS bs
+              , riPayload = payload
               }
 
   let rheaders = NC.responseHeaders resp
@@ -181,13 +183,14 @@ putObjectSingle bucket object headers h offset size = do
     throwIO $ MErrVSinglePUTSizeExceeded size
 
   -- content-length header is automatically set by library.
+  let payload = mkStreamingPayload $ PayloadH h offset size
   resp <- executeRequest $
           defaultS3ReqInfo { riMethod = HT.methodPut
-              , riBucket = Just bucket
+                           , riBucket = Just bucket
-              , riObject = Just object
+                           , riObject = Just object
-              , riHeaders = headers
+                           , riHeaders = headers
-              , riPayload = PayloadH h offset size
+                           , riPayload = payload
-              }
+                           }
 
   let rheaders = NC.responseHeaders resp
       etag = getETagHeader rheaders
@@ -264,14 +267,16 @@ newMultipartUpload bucket object headers = do
 putObjectPart :: Bucket -> Object -> UploadId -> PartNumber -> [HT.Header]
               -> Payload -> Minio PartTuple
 putObjectPart bucket object uploadId partNumber headers payload = do
+  -- transform payload to conduit to enable streaming signature
+  let payload' = mkStreamingPayload payload
   resp <- executeRequest $
           defaultS3ReqInfo { riMethod = HT.methodPut
-              , riBucket = Just bucket
+                           , riBucket = Just bucket
-              , riObject = Just object
+                           , riObject = Just object
-              , riQueryParams = mkOptionalParams params
+                           , riQueryParams = mkOptionalParams params
-              , riHeaders = headers
+                           , riHeaders = headers
-              , riPayload = payload
+                           , riPayload = payload'
-              }
+                           }
   let rheaders = NC.responseHeaders resp
       etag = getETagHeader rheaders
   maybe

src/Network/Minio/Sign/V4.hs

@@ -1,5 +1,5 @@
 --
--- MinIO Haskell SDK, (C) 2017 MinIO, Inc.
+-- MinIO Haskell SDK, (C) 2017-2019 MinIO, Inc.
 --
 -- Licensed under the Apache License, Version 2.0 (the "License");
 -- you may not use this file except in compliance with the License.
@@ -14,21 +14,9 @@
 -- limitations under the License.
 --
 
-module Network.Minio.Sign.V4
+module Network.Minio.Sign.V4 where
-  (
-    signV4
-  , signV4PostPolicy
-  , mkScope
-  , getHeadersToSign
-  , mkCanonicalRequest
-  , mkStringToSign
-  , mkSigningKey
-  , computeSignature
-  , SignV4Data(..)
-  , SignParams(..)
-  , debugPrintSignV4Data
-  ) where
 
+import qualified Conduit                       as C
 import qualified Data.ByteString               as B
 import qualified Data.ByteString.Base64        as Base64
 import qualified Data.ByteString.Char8         as B8
@@ -39,13 +27,15 @@ import qualified Data.Set                      as Set
 import qualified Data.Time                     as Time
 import qualified Network.HTTP.Conduit          as NC
 import           Network.HTTP.Types            (Header, parseQuery)
-import qualified Network.HTTP.Types.Header     as H
+import qualified Network.HTTP.Types            as H
+import           Text.Printf                   (printf)
 
 import           Lib.Prelude
 
 import           Network.Minio.Data.ByteString
 import           Network.Minio.Data.Crypto
 import           Network.Minio.Data.Time
+import           Network.Minio.Errors
 
 -- these headers are not included in the string to sign when signing a
 -- request
@@ -53,7 +43,6 @@ ignoredHeaders :: Set ByteString
 ignoredHeaders = Set.fromList $ map CI.foldedCase
                  [ H.hAuthorization
                  , H.hContentType
-                 , H.hContentLength
                  , H.hUserAgent
                  ]
 
@@ -93,6 +82,20 @@ debugPrintSignV4Data (SignV4Data t s cr h2s o sts sk) = do
       mapM_ (\x -> B.putStr $ B.concat [show x,  " "]) $ B.unpack b
       B8.putStrLn ""
 
+mkAuthHeader :: Text -> ByteString -> ByteString -> ByteString -> H.Header
+mkAuthHeader accessKey scope signedHeaderKeys sign =
+    let authValue = B.concat
+                    [ "AWS4-HMAC-SHA256 Credential="
+                    , toS accessKey
+                    , "/"
+                    , scope
+                    , ", SignedHeaders="
+                    , signedHeaderKeys
+                    , ", Signature="
+                    , sign
+                    ]
+    in (H.hAuthorization, authValue)
+
 -- | Given SignParams and request details, including request method,
 -- request path, headers, query params and payload hash, generates an
 -- updated set of headers, including the x-amz-date header and the
@@ -105,7 +108,6 @@ debugPrintSignV4Data (SignV4Data t s cr h2s o sts sk) = do
 -- is being created. The expiry is interpreted as an integer number of
 -- seconds. The output will be the list of query-parameters to add to
 -- the request.
-
 signV4 :: SignParams -> NC.Request -> [(ByteString, ByteString)]
 signV4 !sp !req =
   let
@@ -139,7 +141,8 @@ signV4 !sp !req =
               else []
 
     -- 1. compute canonical request
-    canonicalRequest = mkCanonicalRequest sp (NC.setQueryString finalQP req)
+    canonicalRequest = mkCanonicalRequest False sp
+                       (NC.setQueryString finalQP req)
                        headersToSign
 
     -- 2. compute string to sign
@@ -152,23 +155,15 @@ signV4 !sp !req =
     signature = computeSignature stringToSign signingKey
 
     -- 4. compute auth header
-    authValue = B.concat
+    authHeader = mkAuthHeader (spAccessKey sp) scope signedHeaderKeys signature
-      [ "AWS4-HMAC-SHA256 Credential="
-      , accessKey
-      , "/"
-      , scope
-      , ", SignedHeaders="
-      , signedHeaderKeys
-      , ", Signature="
-      , signature
-      ]
-    authHeader = (H.hAuthorization, authValue)
 
     -- finally compute output pairs
+    sha256Hdr = ("x-amz-content-sha256",
+                 fromMaybe "UNSIGNED-PAYLOAD" $ spPayloadHash sp)
     output = if isJust expiry
              then ("X-Amz-Signature", signature) : authQP
              else [(\(x, y) -> (CI.foldedCase x, y)) authHeader,
-                   datePair]
+                   datePair, sha256Hdr]
 
   in output
 
@@ -186,9 +181,9 @@ getHeadersToSign !h =
   filter (flip Set.notMember ignoredHeaders . fst) $
   map (\(x, y) -> (CI.foldedCase x, stripBS y)) h
 
-mkCanonicalRequest :: SignParams -> NC.Request -> [(ByteString, ByteString)]
+mkCanonicalRequest :: Bool -> SignParams -> NC.Request -> [(ByteString, ByteString)]
-                    -> ByteString
+                   -> ByteString
-mkCanonicalRequest !sp !req !headersForSign =
+mkCanonicalRequest !isStreaming !sp !req !headersForSign =
   let
     canonicalQueryString = B.intercalate "&" $
       map (\(x, y) -> B.concat [x, "=", y]) $
@@ -203,6 +198,10 @@ mkCanonicalRequest !sp !req !headersForSign =
 
     signedHeaders = B.intercalate ";" $ map fst sortedHeaders
 
+    payloadHashStr =
+        if isStreaming
+        then "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
+        else fromMaybe "UNSIGNED-PAYLOAD" $ spPayloadHash sp
   in
     B.intercalate "\n"
     [ NC.method req
@@ -210,7 +209,7 @@ mkCanonicalRequest !sp !req !headersForSign =
     , canonicalQueryString
     , canonicalHeaders
     , signedHeaders
-    , maybe "UNSIGNED-PAYLOAD" identity $ spPayloadHash sp
+    , payloadHashStr
     ]
 
 mkStringToSign :: UTCTime -> ByteString -> ByteString -> ByteString
@@ -246,3 +245,147 @@ signV4PostPolicy !postPolicyJSON !sp =
     Map.fromList [ ("x-amz-signature", signature)
                  , ("policy", stringToSign)
                  ]
+
+chunkSizeConstant :: Int
+chunkSizeConstant = 64 * 1024
+
+-- base16Len computes the number of bytes required to represent @n (> 0)@ in
+-- hexadecimal.
+base16Len :: Integral a => a -> Int
+base16Len n | n == 0 = 0
+            | otherwise = 1 + base16Len (n `div` 16)
+
+signedStreamLength :: Int64 -> Int64
+signedStreamLength dataLen =
+  let
+    chunkSzInt = fromIntegral chunkSizeConstant
+    (numChunks, lastChunkLen) = quotRem dataLen chunkSzInt
+
+
+    -- Structure of a chunk:
+    --   string(IntHexBase(chunk-size)) + ";chunk-signature=" + signature + \r\n + chunk-data + \r\n
+    encodedChunkLen csz = fromIntegral (base16Len csz) + 17 + 64 + 2 + csz + 2
+    fullChunkSize = encodedChunkLen chunkSzInt
+    lastChunkSize = bool 0 (encodedChunkLen lastChunkLen) $ lastChunkLen > 0
+    finalChunkSize = 1 + 17 + 64 + 2 + 2
+  in
+    numChunks * fullChunkSize + lastChunkSize + finalChunkSize
+
+signV4Stream :: Int64 -> SignParams -> NC.Request
+             -> (C.ConduitT () ByteString (C.ResourceT IO) () -> NC.Request)
+             -- -> ([Header], C.ConduitT () ByteString (C.ResourceT IO) () -> NC.RequestBody)
+signV4Stream !payloadLength !sp !req =
+  let
+    ts = spTimeStamp sp
+
+    addContentEncoding hs =
+        let ceMay = headMay $ filter (\(x, _) -> x == "content-encoding") hs
+        in case ceMay of
+             Nothing      -> ("content-encoding", "aws-chunked") : hs
+             Just (_, ce) -> ("content-encoding", ce <> ",aws-chunked") :
+                             filter (\(x, _) -> x /= "content-encoding") hs
+
+    -- headers to be added to the request
+    datePair = ("X-Amz-Date", awsTimeFormatBS ts)
+    computedHeaders = addContentEncoding $
+                      datePair : NC.requestHeaders req
+
+    -- headers specific to streaming signature
+    signedContentLength = signedStreamLength payloadLength
+    streamingHeaders :: [Header]
+    streamingHeaders =
+        [ ("x-amz-decoded-content-length", show payloadLength)
+        , ("content-length", show signedContentLength )
+        , ("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD")
+        ]
+    headersToSign = getHeadersToSign $ computedHeaders ++ streamingHeaders
+    signedHeaderKeys = B.intercalate ";" $ sort $ map fst headersToSign
+    finalQP = parseQuery (NC.queryString req)
+
+    -- 1. Compute Seed Signature
+    -- 1.1 Canonical Request
+    canonicalReq = mkCanonicalRequest True sp
+                   (NC.setQueryString finalQP req)
+                   headersToSign
+
+    region = fromMaybe "" $ spRegion sp
+    scope = mkScope ts region
+    accessKey = spAccessKey sp
+    secretKey = spSecretKey sp
+
+    -- 1.2 String toSign
+    stringToSign = mkStringToSign ts scope canonicalReq
+
+    -- 1.3 Compute signature
+    -- 1.3.1 compute signing key
+    signingKey = mkSigningKey ts region $ toS secretKey
+
+    -- 1.3.2 Compute signature
+    seedSignature = computeSignature stringToSign signingKey
+
+    -- 1.3.3 Compute Auth Header
+    authHeader = mkAuthHeader accessKey scope signedHeaderKeys seedSignature
+
+    -- 1.4 Updated headers for the request
+    finalReqHeaders = authHeader : (computedHeaders ++ streamingHeaders)
+    -- headersToAdd = authHeader : datePair : streamingHeaders
+
+    toHexStr n = B8.pack $ printf "%x" n
+
+    (numParts, lastPSize) = payloadLength `quotRem` fromIntegral chunkSizeConstant
+
+    -- Function to compute string to sign for each chunk.
+    chunkStrToSign prevSign currChunkHash =
+        B.intercalate "\n"
+        [ "AWS4-HMAC-SHA256-PAYLOAD"
+        , awsTimeFormatBS ts
+        , scope
+        , prevSign
+        , hashSHA256 ""
+        , currChunkHash
+        ]
+
+    -- Read n byte from upstream and return a strict bytestring.
+    mustTakeN n = do
+        bs <- toS <$> (C.takeCE n C..| C.sinkLazy)
+        when (B.length bs /= n) $
+            throwIO MErrVStreamingBodyUnexpectedEOF
+        return bs
+
+    signerConduit n lps prevSign =
+         -- First case encodes a full chunk of length
+         -- 'chunkSizeConstant'.
+      if | n > 0 -> do
+               bs <- mustTakeN chunkSizeConstant
+               let strToSign = chunkStrToSign prevSign (hashSHA256 bs)
+                   nextSign = computeSignature strToSign signingKey
+                   chunkBS = toHexStr chunkSizeConstant
+                          <> ";chunk-signature="
+                          <> nextSign <> "\r\n" <> bs <> "\r\n"
+               C.yield chunkBS
+               signerConduit (n-1) lps nextSign
+
+         -- Second case encodes the last chunk which is smaller than
+         -- 'chunkSizeConstant'
+         | lps > 0 -> do
+               bs <- mustTakeN $ fromIntegral lps
+               let strToSign = chunkStrToSign prevSign (hashSHA256 bs)
+                   nextSign = computeSignature strToSign signingKey
+                   chunkBS = toHexStr lps <> ";chunk-signature="
+                          <> nextSign <> "\r\n" <> bs <> "\r\n"
+               C.yield chunkBS
+               signerConduit 0 0 nextSign
+
+         -- Last case encodes the final signature chunk that has no
+         -- data.
+         | otherwise -> do
+               let strToSign = chunkStrToSign prevSign (hashSHA256 "")
+                   nextSign = computeSignature strToSign signingKey
+                   lastChunkBS = "0;chunk-signature=" <> nextSign <> "\r\n\r\n"
+               C.yield lastChunkBS
+  in
+    \src -> req { NC.requestHeaders = finalReqHeaders
+                , NC.requestBody =
+                  NC.requestBodySource signedContentLength $
+                  src C..| signerConduit numParts lastPSize seedSignature
+                }

src/Network/Minio/Utils.hs

@@ -1,5 +1,5 @@
 --
--- MinIO Haskell SDK, (C) 2017 MinIO, Inc.
+-- MinIO Haskell SDK, (C) 2017-2019 MinIO, Inc.
 --
 -- Licensed under the Apache License, Version 2.0 (the "License");
 -- you may not use this file except in compliance with the License.
@@ -16,12 +16,12 @@
 
 module Network.Minio.Utils where
 
+import qualified Conduit                       as C
 import           Control.Monad.IO.Unlift       (MonadUnliftIO)
 import qualified Control.Monad.Trans.Resource  as R
 import qualified Data.ByteString               as B
 import qualified Data.ByteString.Lazy          as LB
 import           Data.CaseInsensitive          (mk, original)
-import qualified Data.Conduit                  as C
 import qualified Data.Conduit.Binary           as CB
 import qualified Data.List                     as List
 import qualified Data.Map                      as Map
@@ -212,22 +212,17 @@ mkQuery k mv = (k,) <$> mv
 mkOptionalParams :: [(Text, Maybe Text)] -> HT.Query
 mkOptionalParams params = HT.toQuery $ uncurry  mkQuery <$> params
 
-chunkBSConduit :: (Monad m, Integral a)
+-- | Conduit that rechunks bytestrings into the given chunk
-               => [a] -> C.ConduitM ByteString ByteString m ()
+-- lengths. Stops after given chunk lengths are yielded. Stops if
-chunkBSConduit s = loop 0 [] s
+-- there are no more chunks to yield or if a shorter chunk is
-  where
+-- received. Does not throw any errors.
-    loop _ _ [] = return ()
+chunkBSConduit :: (Monad m) => [Int] -> C.ConduitM ByteString ByteString m ()
-    loop n readChunks (size:sizes) = do
+chunkBSConduit [] = return ()
-      bsMay <- C.await
+chunkBSConduit (s:ss) = do
-      case bsMay of
+    bs <- fmap LB.toStrict $ C.takeCE s C..| C.sinkLazy
-        Nothing -> when (n > 0) $ C.yield $ B.concat readChunks
+    if | B.length bs == s -> C.yield bs >> chunkBSConduit ss
-        Just bs -> if n + fromIntegral (B.length bs) >= size
+       | B.length bs > 0  -> C.yield bs
-                   then do let (a, b) = B.splitAt (fromIntegral $ size - n) bs
+       | otherwise        -> return ()
-                               chunkBS = B.concat $ readChunks ++ [a]
-                           C.yield chunkBS
-                           loop (fromIntegral $ B.length b) [b] sizes
-                   else loop (n + fromIntegral (B.length bs))
-                        (readChunks ++ [bs]) (size:sizes)
 
 -- | Select part sizes - the logic is that the minimum part-size will
 -- be 64MiB.