diff --git a/waterbutler/core/utils.py b/waterbutler/core/utils.py
index d319d5b84..983e9b3b6 100644
--- a/waterbutler/core/utils.py
+++ b/waterbutler/core/utils.py
@@ -250,7 +250,10 @@ async def __anext__(self):
             else:
                 return path.path.replace(self.parent_path.path, '', 1), EmptyStream()
 
-        return path.path.replace(self.parent_path.path, '', 1), await self.provider.download(path)
+        return (
+            path.path.replace(self.parent_path.path, '', 1),
+            await self.provider.download(path, metadata=current[1])
+        )
 
 
 class ZipStreamGeneratorPaginated(BaseZipStreamGenerator):
@@ -308,7 +311,7 @@ async def __anext__(self):
 
         return (
             path.path.replace(self.parent_path.path, '', 1),
-            await self.provider.download(path),
+            await self.provider.download(path, metadata=current[1]),
         )
 
 
diff --git a/waterbutler/providers/osfstorage/provider.py b/waterbutler/providers/osfstorage/provider.py
index 77b6e1215..d6efb6c9f 100644
--- a/waterbutler/providers/osfstorage/provider.py
+++ b/waterbutler/providers/osfstorage/provider.py
@@ -195,7 +195,7 @@ def build_signed_url(method, url, data=None, params=None, ttl=100, **kwargs):
 
         return url, data, params
 
-    async def download(self, path, version=None, revision=None, mode=None, **kwargs):
+    async def download(self, path, version=None, revision=None, mode=None, metadata=None, **kwargs):
         if not path.identifier:
             raise exceptions.NotFoundError(str(path))
 
@@ -209,28 +209,37 @@ async def download(self, path, version=None, revision=None, mode=None, **kwargs)
             # version could be 0 here
             version = revision
 
-        # Capture user_id for analytics if user is logged in
-        user_param = {}
-        if self.auth.get('id', None):
-            user_param = {'user': self.auth['id']}
+        storage = None
+        if metadata is not None:
+            storage = metadata.raw.get('storage', None)
 
-        # osf storage metadata will return a virtual path within the provider
-        resp = await self.make_signed_request(
-            'GET',
-            self.build_url(path.identifier, 'download', version=version, mode=mode),
-            expects=(200, ),
-            params=user_param,
-            throws=exceptions.DownloadError,
-        )
-        data = await resp.json()
+        if version is None and storage and storage.get('data', None):
+            data = storage
+        else:
+            # Capture user_id for analytics if user is logged in
+            user_param = {}
+            if self.auth.get('id', None):
+                user_param = {'user': self.auth['id']}
+
+            # osf storage metadata will return a virtual path within the provider
+            resp = await self.make_signed_request(
+                'GET',
+                self.build_url(path.identifier, 'download', version=version, mode=mode),
+                expects=(200, ),
+                params=user_param,
+                throws=exceptions.DownloadError,
+            )
+            data = await resp.json()
 
         provider_object = self.make_provider(data['settings'])
-        name = data['data'].pop('name')
-        data['data']['path'] = await provider_object.validate_path('/' + data['data']['path'])
+        file_data = dict(data['data'])
+        name = file_data.pop('name')
+        file_data['path'] = await provider_object.validate_path('/' + file_data['path'])
         download_kwargs = {}
         download_kwargs.update(kwargs)
-        download_kwargs.update(data['data'])
+        download_kwargs.update(file_data)
         download_kwargs['display_name'] = kwargs.get('display_name') or name
+
         return await provider_object.download(**download_kwargs)
 
     async def upload(self, stream, path, **kwargs):
@@ -555,6 +564,8 @@ async def _children_metadata(self, path, limit=None, after=None, **kwargs):
             query['limit'] = limit
             if after is not None:
                 query['after'] = after
+        elif kwargs.get('orm') is not None:
+            query['orm'] = kwargs['orm']
 
         resp = await self.make_signed_request(
             'GET',