Cargohold: remote assets

changelog.d/1-api-changes/remote-assets

@@ -0,0 +1,4 @@
+Enable downloading assets from a remote (federated) cargohold instance via the v4 API. 
+The content of remote assets is returned as stream with content type
+`application/octet-stream`.
+Please refer to the Swagger API documentation for more details.

charts/brig/templates/tests/configmap.yaml

@@ -65,6 +65,10 @@ data:
         host: galley.{{ .Release.Namespace }}-fed2.svc.cluster.local
         port: 8080
 
+      cargohold:
+        host: cargohold.{{ .Release.Namespace }}-fed2.svc.cluster.local
+        port: 8080
+
       # TODO remove this
       federator:
         host: federator.{{ .Release.Namespace }}-fed2.svc.cluster.local

charts/brig/values.yaml

@@ -29,7 +29,7 @@ config:
   # -- If set to false,  'dynamoDBEndpoint' _must_ be set.
   randomPrekeys: true
   useSES: true
-  enableFederator: false # keep enableFederator default in sync with galley chart's config.enableFederator as well as wire-server chart's tag.federator
+  enableFederator: false # keep enableFederator default in sync with galley and cargohold chart's config.enableFederator as well as wire-server chart's tag.federator
   emailSMS:
     general:
       templateBranding:

charts/cargohold/templates/configmap.yaml

@@ -11,6 +11,12 @@ data:
       host: 0.0.0.0
       port: {{ .Values.service.internalPort }}
 
+    {{- if .Values.config.enableFederator }}
+    federator:
+      host: federator
+      port: 8080
+    {{- end }}
+
     aws:
       {{- with .Values.config.aws }}
       s3Bucket: {{ .s3Bucket }}

charts/cargohold/templates/tests/cargohold-integration.yaml

@@ -9,6 +9,9 @@ spec:
     - name: "cargohold-integration"
       configMap:
         name: "cargohold-integration"
+    - name: "cargohold-config"
+      configMap:
+        name: "cargohold"
   containers:
     # NOTE: the bucket for these tests must be created.
     # If using the wire-server/fake-aws-s3 chart, `dummy-bucket` will already be created.
@@ -17,6 +20,8 @@ spec:
     volumeMounts:
     - name: "cargohold-integration"
       mountPath: "/etc/wire/integration"
+    - name: "cargohold-config"
+      mountPath: "/etc/wire/cargohold/conf"
     env:
     # these dummy values are necessary for Amazonka's "Discover"
     - name: AWS_ACCESS_KEY_ID

charts/cargohold/values.yaml

@@ -14,7 +14,8 @@ resources:
     cpu: "500m"
 config:
   logLevel: Info
+  enableFederator: false # keep enableFederator default in sync with brig and galley chart's config.enableFederator as well as wire-server chart's tag.federator
   aws:
     region: "eu-west-1"
     s3Bucket: assets
-  proxy: {}
+  proxy: {}

charts/galley/values.yaml

@@ -19,7 +19,7 @@ config:
   cassandra:
     host: aws-cassandra
     replicaCount: 3
-  enableFederator: false # keep enableFederator default in sync with brig chart's config.enableFederator as well as wire-server chart's tag.federator
+  enableFederator: false # keep enableFederator default in sync with brig and cargohold chart's config.enableFederator as well as wire-server chart's tag.federator
   settings:
     maxTeamSize: 500
     maxConvSize: 500

hack/helm_vars/wire-server/values.yaml.gotmpl

@@ -51,7 +51,7 @@ brig:
       sessionTokenTimeout: 20
       accessTokenTimeout: 30
       providerTokenTimeout: 60
-    enableFederator: true # keep in sync with galley.config.enableFederator and tags.federator!
+    enableFederator: true # keep in sync with galley.config.enableFederator, cargohold.config.enableFederator and tags.federator!
     optSettings:
       setActivationTimeout: 5
       # keep this in sync with brigSettingsTeamInvitationTimeout in spar/templates/tests/configmap.yaml
@@ -125,6 +125,7 @@ cargohold:
     aws:
       s3Bucket: dummy-bucket
       s3Endpoint: http://fake-aws-s3:9000
+    enableFederator: true # keep in sync with brig.config.enableFederator, galley.config.enableFederator and tags.federator!
   secrets:
     awsKeyId: dummykey
     awsSecretKey: dummysecret
@@ -135,7 +136,7 @@ galley:
     cassandra:
       host: cassandra-ephemeral
       replicaCount: 1
-    enableFederator: true # keep in sync with brig.config.enableFederator and tags.federator!
+    enableFederator: true # keep in sync with brig.config.enableFederator, cargohold.config.enableFederator and tags.federator!
     settings:
       maxConvAndTeamSize: 16
       maxTeamSize: 32

libs/metrics-wai/src/Data/Metrics/Servant.hs

@@ -137,6 +137,9 @@ instance RoutesToPaths (Verb method status cts a) where
 instance RoutesToPaths (NoContentVerb method) where
   getRoutes = []
 
+instance RoutesToPaths (Stream method status framing ct a) where
+  getRoutes = []
+
 -- route :<|> routes
 instance
   ( RoutesToPaths route,

libs/wire-api-federation/src/Wire/API/Federation/API/Cargohold.hs

@@ -17,15 +17,45 @@
 
 module Wire.API.Federation.API.Cargohold where
 
+import Data.Aeson (FromJSON (..), ToJSON (..))
+import Data.Id
+import Imports
 import Servant.API
 import Servant.API.Generic
-import Wire.API.Federation.API.Common
+import Wire.API.Arbitrary (Arbitrary, GenericUniform (..))
+import Wire.API.Asset
+import Wire.API.Routes.AssetBody
+import Wire.API.Util.Aeson
+
+data GetAsset = GetAsset
+  { -- | User requesting the asset. Implictly qualified with the source domain.
+    gaUser :: UserId,
+    -- | Asset key for the asset to download. Implictly qualified with the
+    -- target domain.
+    gaKey :: AssetKey,
+    -- | Optional asset token.
+    gaToken :: Maybe AssetToken
+  }
+  deriving stock (Eq, Show, Generic)
+  deriving (Arbitrary) via (GenericUniform GetAsset)
+  deriving (ToJSON, FromJSON) via (CustomEncoded GetAsset)
+
+data GetAssetResponse = GetAssetResponse
+  {gaAvailable :: Bool}
+  deriving stock (Eq, Show, Generic)
+  deriving (Arbitrary) via (GenericUniform GetAssetResponse)
+  deriving (ToJSON, FromJSON) via (CustomEncoded GetAssetResponse)
 
 data CargoholdApi routes = CargoholdApi
   { getAsset ::
       routes
         :- "get-asset"
-        :> ReqBody '[JSON] ()
-        :> Post '[JSON] EmptyResponse
+        :> ReqBody '[JSON] GetAsset
+        :> Post '[JSON] GetAssetResponse,
+    streamAsset ::
+      routes
+        :- "stream-asset"
+        :> ReqBody '[JSON] GetAsset
+        :> StreamPost NoFraming OctetStream AssetSource
   }
   deriving (Generic)

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

@@ -45,10 +45,10 @@ newtype AssetSource = AssetSource
   { getAssetSource ::
       ConduitT () ByteString (ResourceT IO) ()
   }
-  deriving newtype (FromSourceIO ByteString)
+  deriving newtype (FromSourceIO ByteString, ToSourceIO ByteString)
 
 instance ToSchema AssetSource where
-  declareNamedSchema _ = pure $ named "AssetSource" $ mempty
+  declareNamedSchema _ = pure $ named "AssetSource" mempty
 
 type AssetBody =
   StreamBody'

services/brig/brig.cabal

@@ -4,7 +4,7 @@ cabal-version: 2.0
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 350e615ec20264419ad3244b5ba5b93c8a0bbea7dc35db5f300a3fbf3ef4e4c0
+-- hash: 2a5e368730d4530d05df5395f6b3157536513b1049130b541d257aa79296cb75
 
 name:           brig
 version:        2.0
@@ -349,6 +349,7 @@ executable brig-integration
     , http-api-data
     , http-client
     , http-client-tls >=0.2
+    , http-media
     , http-types
     , imports
     , lens >=3.9

services/brig/package.yaml

@@ -226,6 +226,7 @@ executables:
     - http-api-data
     - http-client
     - http-client-tls >=0.2
+    - http-media
     - http-types
     - imports
     - lens >=3.9

services/brig/test/integration/API/User/Account.hs

@@ -1319,7 +1319,7 @@ testDeleteInternal brig cannon aws = do
 testDeleteWithProfilePic :: Brig -> CargoHold -> Http ()
 testDeleteWithProfilePic brig cargohold = do
   uid <- userId <$> createAnonUser "anon" brig
-  ast <- uploadAsset cargohold uid "this is my profile pic"
+  ast <- responseJsonError =<< uploadAsset cargohold uid Asset.defAssetSettings "this is my profile pic"
   -- Ensure that the asset is there
   downloadAsset cargohold uid (ast ^. Asset.assetKey) !!! const 200 === statusCode
   let newAssets =

services/brig/test/integration/API/User/Util.hs

@@ -27,7 +27,6 @@ import Brig.Types
 import Brig.Types.Team.LegalHold (LegalHoldClientRequest (..))
 import Brig.Types.User.Auth hiding (user)
 import qualified Brig.ZAuth
-import qualified CargoHold.Types.V3 as CHV3
 import qualified Codec.MIME.Type as MIME
 import Control.Lens (preview, (^?))
 import Control.Monad.Catch (MonadCatch)
@@ -408,24 +407,22 @@ uploadAsset ::
   (MonadCatch m, MonadIO m, MonadHttp m, HasCallStack) =>
   CargoHold ->
   UserId ->
+  AssetSettings ->
   ByteString ->
-  m CHV3.Asset
-uploadAsset c usr dat = do
-  let sts = CHV3.defAssetSettings
-      ct = MIME.Type (MIME.Application "text") []
-      mpb = CHV3.buildMultipartBody sts ct (LB.fromStrict dat)
-  rsp <-
-    post
-      ( c
-          . path "/assets/v3"
-          . zUser usr
-          . zConn "conn"
-          . content "multipart/mixed"
-          . lbytes (toLazyByteString mpb)
-      )
-      <!! const 201
-      === statusCode
-  responseJsonError rsp
+  m (Response (Maybe LByteString))
+uploadAsset c usr sts dat = do
+  let ct = MIME.Type (MIME.Application "text") []
+      mpb = buildMultipartBody sts ct (LB.fromStrict dat)
+  post
+    ( c
+        . path "/assets/v3"
+        . zUser usr
+        . zConn "conn"
+        . content "multipart/mixed"
+        . lbytes (toLazyByteString mpb)
+    )
+    <!! const 201
+    === statusCode
 
 downloadAsset ::
   (MonadIO m, MonadHttp m) =>

services/brig/test/integration/Federation/End2end.hs

@@ -23,9 +23,9 @@ import Bilge
 import Bilge.Assert ((!!!), (<!!), (===))
 import Brig.API.Client (pubClient)
 import qualified Brig.Options as BrigOpts
-import Brig.Types
+import Brig.Types hiding (assetKey)
 import Control.Arrow ((&&&))
-import Control.Lens (sequenceAOf, _1)
+import Control.Lens (sequenceAOf, view, (.~), _1)
 import qualified Data.Aeson as Aeson
 import Data.ByteString.Conversion (toByteString')
 import Data.Domain (Domain)
@@ -49,6 +49,7 @@ import qualified Test.Tasty.Cannon as WS
 import Test.Tasty.HUnit
 import Util
 import Util.Options (Endpoint)
+import Wire.API.Asset
 import Wire.API.Conversation
 import Wire.API.Conversation.Role (roleNameWireAdmin)
 import Wire.API.Event.Conversation
@@ -76,12 +77,14 @@ spec ::
   Manager ->
   Brig ->
   Galley ->
+  CargoHold ->
   Cannon ->
   Endpoint ->
   Brig ->
   Galley ->
+  CargoHold ->
   IO TestTree
-spec _brigOpts mg brig galley cannon _federator brigTwo galleyTwo =
+spec _brigOpts mg brig galley cargohold cannon _federator brigTwo galleyTwo cargoholdTwo =
   pure $
     testGroup
       "federation-end2end-user"
@@ -99,7 +102,8 @@ spec _brigOpts mg brig galley cannon _federator brigTwo galleyTwo =
         test mg "include remote users to new conversation" $ testRemoteUsersInNewConv brig galley brigTwo galleyTwo,
         test mg "send a message to a remote user" $ testSendMessage brig brigTwo galleyTwo cannon,
         test mg "send a message in a remote conversation" $ testSendMessageToRemoteConv brig brigTwo galley galleyTwo cannon,
-        test mg "delete user connected to remotes and in conversation with remotes" $ testDeleteUser brig brigTwo galley galleyTwo cannon
+        test mg "delete user connected to remotes and in conversation with remotes" $ testDeleteUser brig brigTwo galley galleyTwo cannon,
+        test mg "download remote asset" $ testRemoteAsset brig brigTwo cargohold cargoholdTwo
       ]
 
 -- | Path covered by this test:
@@ -619,3 +623,17 @@ testDeleteUser brig1 brig2 galley1 galley2 cannon1 = do
     WS.assertMatch_ (5 # Second) wsAlice $ matchDeleteUserNotification bobDel
     WS.assertMatch_ (5 # Second) wsAlice $ matchConvLeaveNotification conv1 bobDel [bobDel]
     WS.assertMatch_ (5 # Second) wsAlice $ matchConvLeaveNotification conv2 bobDel [bobDel]
+
+testRemoteAsset :: Brig -> Brig -> CargoHold -> CargoHold -> Http ()
+testRemoteAsset brig1 brig2 ch1 ch2 = do
+  alice <- userQualifiedId <$> randomUser brig1
+  bob <- userQualifiedId <$> randomUser brig2
+
+  let sts = defAssetSettings & setAssetPublic .~ True
+  ast <- responseJsonError =<< uploadAsset ch2 (qUnqualified bob) sts "hello world"
+  let qkey = view assetKey ast
+
+  downloadAsset ch1 (qUnqualified alice) qkey
+    !!! do
+      const 200 === statusCode
+      const (Just "hello world") === responseBody

services/brig/test/integration/Federation/Util.hs

@@ -50,6 +50,7 @@ import GHC.IO.Exception (IOException (ioe_errno))
 import qualified Galley.Types.Teams.SearchVisibility as Team
 import Imports
 import qualified Network.HTTP.Client as HTTP
+import Network.HTTP.Media
 import Network.Socket
 import Network.Wai.Handler.Warp (Port)
 import Network.Wai.Test (Session)
@@ -73,7 +74,7 @@ withTempMockFederator :: Opt.Opts -> LByteString -> Session a -> IO (a, [Mock.Fe
 withTempMockFederator opts resp action =
   Mock.withTempMockFederator
     [("Content-Type", "application/json")]
-    (const (pure resp))
+    (const (pure ("application" // "json", resp)))
     $ \mockPort -> do
       let opts' =
             opts

services/brig/test/integration/Main.hs

@@ -74,6 +74,7 @@ import Wire.API.Federation.Domain
 data BackendConf = BackendConf
   { remoteBrig :: Endpoint,
     remoteGalley :: Endpoint,
+    remoteCargohold :: Endpoint,
     remoteFederatorInternal :: Endpoint,
     remoteFederatorExternal :: Endpoint
   }
@@ -84,6 +85,7 @@ instance FromJSON BackendConf where
     BackendConf
       <$> o .: "brig"
       <*> o .: "galley"
+      <*> o .: "cargohold"
       <*> o .: "federatorInternal"
       <*> o .: "federatorExternal"
 
@@ -118,6 +120,7 @@ runTests iConf brigOpts otherArgs = do
       f = federatorInternal iConf
       brigTwo = mkRequest $ remoteBrig (backendTwo iConf)
       galleyTwo = mkRequest $ remoteGalley (backendTwo iConf)
+      ch2 = mkRequest $ remoteCargohold (backendTwo iConf)
 
   let turnFile = Opts.servers . Opts.turn $ brigOpts
       turnFileV2 = (Opts.serversV2 . Opts.turn) brigOpts
@@ -143,7 +146,7 @@ runTests iConf brigOpts otherArgs = do
   createIndex <- Index.Create.spec brigOpts
   browseTeam <- TeamUserSearch.tests brigOpts mg g b
   userPendingActivation <- UserPendingActivation.tests brigOpts mg db b g s
-  federationEnd2End <- Federation.End2end.spec brigOpts mg b g c f brigTwo galleyTwo
+  federationEnd2End <- Federation.End2end.spec brigOpts mg b g ch c f brigTwo galleyTwo ch2
   federationEndpoints <- API.Federation.tests mg brigOpts b c fedBrigClient
   includeFederationTests <- (== Just "1") <$> Blank.getEnv "INTEGRATION_FEDERATION_TESTS"
   internalApi <- API.Internal.tests brigOpts mg db b (brig iConf) gd g