Merge pull request #1 from CSCfi/devel

Add resumablejs uploads and container / object replication

Author: sampsapenna

README.md

@@ -1,7 +1,7 @@
 ### swift-upload-runner – better browser file upload and download for swift-browser-ui
 
 `swift-upload-runner` makes it possible to properly batch upload from browser
-into object storage backends using Openstack Swift API. It also will make possible
+into object storage backends using Openstack Swift API. It will also make possible
 proper container uploads without plaintext passwords or tokens, to prevent
 token expiration.
 

swift_upload_runner/api.py

@@ -2,10 +2,12 @@
 
 
 import aiohttp.web
-# import asyncio
+import asyncio
 
-from .common import get_auth_instance
+from .common import get_auth_instance, get_upload_instance
+from .common import parse_multipart_in
 from .download import FileDownloadProxy, ContainerArchiveDownloadProxy
+from .replicate import ObjectReplicationProxy
 
 
 async def handle_get_object(
@@ -25,8 +27,8 @@ async def handle_get_object(
     resp = aiohttp.web.StreamResponse()
 
     # Create headers
-    resp.headers["Content-Type"] = download.get_type()
-    resp.headers["Content-Length"] = str(download.get_size())
+    resp.headers["Content-Type"] = await download.a_get_type()
+    resp.headers["Content-Length"] = str(await download.a_get_size())
 
     await resp.prepare(request)
 
@@ -36,10 +38,134 @@ async def handle_get_object(
     return resp
 
 
+async def handle_replicate_container(
+        request: aiohttp.web.Request
+) -> aiohttp.web.Response:
+    """Handle request to replicating a container from a source."""
+    auth = get_auth_instance(request)
+
+    project = request.match_info["project"]
+    container = request.match_info["container"]
+
+    source_project = request.query["from_project"]
+    source_container = request.query["from_container"]
+
+    replicator = ObjectReplicationProxy(
+        auth,
+        request.app["client"],
+        project,
+        container,
+        source_project,
+        source_container
+    )
+
+    asyncio.ensure_future(replicator.a_copy_from_container())
+
+    return aiohttp.web.Response(status=202)
+
+
+async def handle_replicate_object(
+        request: aiohttp.web.Request
+) -> aiohttp.web.Response:
+    """Handle a request to replicating an object from a source."""
+    auth = get_auth_instance(request)
+
+    project = request.match_info["project"]
+    container = request.match_info["container"]
+
+    source_project = request.query["from_project"]
+    source_container = request.query["from_container"]
+    source_object = request.query["from_object"]
+
+    replicator = ObjectReplicationProxy(
+        auth,
+        request.app["client"],
+        project,
+        container,
+        source_project,
+        source_container
+    )
+
+    asyncio.ensure_future(replicator.a_copy_object(source_object))
+
+    return aiohttp.web.Response(status=202)
+
+
+async def handle_post_object_chunk(
+        request: aiohttp.web.Request
+) -> aiohttp.web.Response:
+    """Handle a request for posting an object chunk."""
+    if "from_object" in request.query.keys():
+        return await handle_replicate_object(request)
+    if "from_container" in request.query.keys():
+        return await handle_replicate_container(request)
+
+    project = request.match_info["project"]
+    container = request.match_info["container"]
+
+    query, data = await parse_multipart_in(request)
+
+    upload_session = await get_upload_instance(
+        request,
+        project,
+        container,
+        p_query=query
+    )
+
+    return await upload_session.a_add_chunk(
+        query,
+        data
+    )
+
+
+async def handle_get_object_chunk(
+        request: aiohttp.web.Request
+) -> aiohttp.web.Response:
+    """Handle a request for checking if a chunk exists."""
+    get_auth_instance(request)
+
+    project = request.match_info["project"]
+    container = request.match_info["container"]
+
+    try:
+        # Infuriatingly resumable.js starts counting chunks from 1
+        # thus, reducing said 1 from the resulting chunk number
+        chunk_number = int(request.query["resumableChunkNumber"]) - 1
+    except KeyError:
+        raise aiohttp.web.HTTPBadRequest(reason="Malformed query string")
+
+    upload_session = await get_upload_instance(
+        request,
+        project,
+        container
+    )
+
+    return await upload_session.a_check_segment(
+        chunk_number
+    )
+
+
+async def handle_post_object_options(
+        request: aiohttp.web.Request
+) -> aiohttp.web.Response:
+    """Handle options request for posting the object chunk."""
+    resp = aiohttp.web.Response(
+        headers={
+            "Access-Control-Allow-Methods": "POST, OPTIONS, GET",
+            "Access-Control-Max-Age": "84600",
+        }
+    )
+
+    return resp
+
+
 async def handle_get_container(
         request: aiohttp.web.Request
 ) -> aiohttp.web.StreamResponse:
     """Handle a request for getting container contents as an archive."""
+    if "resumableChunkNumber" in request.query.keys():
+        return await handle_get_object_chunk(request)
+
     auth = get_auth_instance(request)
 
     resp = aiohttp.web.StreamResponse()

swift_upload_runner/auth.py

@@ -33,7 +33,9 @@ async def handle_login(
         project = request.match_info["project"]
         login_form = await request.post()
         token = login_form["token"]
-        request.app[session_key] = initiate_os_session(
+        request.app[session_key] = {}
+        request.app[session_key]["uploads"] = {}
+        request.app[session_key]["auth"] = initiate_os_session(
             token,
             project
         )
@@ -96,7 +98,7 @@ async def handle_validate_authentication(
         path = request.url.path
     except KeyError:
         raise aiohttp.web.HTTPUnauthorized(
-            reason="Query string missing validity or signature."
+            reason="Query string missing validity or signature"
         )
 
     await test_signature(

swift_upload_runner/common.py

@@ -5,10 +5,12 @@
 
 
 import aiohttp.web
-
 import keystoneauth1.session
 
 
+import swift_upload_runner.upload as upload
+
+
 def generate_download_url(
         host,
         container=None,
@@ -23,21 +25,110 @@ def generate_download_url(
     return f'{host}/{container}/{object_name}'
 
 
-def get_auth_instance(
+def get_download_host(
+        auth: keystoneauth1.session.Session,
+        project: str
+) -> str:
+    """Get the actual download host with shared container support."""
+    ret = auth.get_endpoint(service_type="object-store")
+
+    if project not in ret:
+        ret = ret.replace(ret.split("/")[-1], project)
+
+    return ret
+
+
+def get_session_id(
         request: aiohttp.web.Request
-) -> keystoneauth1.session.Session:
-    """Return the session specific keystone auth instance"""
+) -> str:
+    """Return the session id from request."""
     try:
-        return request.app[request.cookies["RUNNER_SESSION_ID"]]
+        return request.cookies["RUNNER_SESSION_ID"]
     except KeyError:
         try:
-            return request.app[request.query["session"]]
+            return request.query["session"]
         except KeyError:
             raise aiohttp.web.HTTPUnauthorized(
-                reason="Runner session ID missing"
+                reason="Missing runner session ID"
             )
 
 
+def get_auth_instance(
+        request: aiohttp.web.Request
+) -> keystoneauth1.session.Session:
+    """Return the session specific keystone auth instance"""
+    return request.app[get_session_id(request)]["auth"]
+
+
+async def parse_multipart_in(
+        request: aiohttp.web.Request,
+) -> typing.Tuple[typing.Dict[str, typing.Any], aiohttp.MultipartReader]:
+    """Parse the form headers into a dictionary and chunk data as reader."""
+    reader = await request.multipart()
+
+    ret_d = {}
+
+    while True:
+        field = await reader.next()
+        if field.name == "file":  # type: ignore
+            ret_d["filename"] = field.filename  # type: ignore
+            return ret_d, field  # type: ignore
+        if field.name == "resumableChunkNumber":  # type: ignore
+            ret_d["resumableChunkNumber"] = \
+                int(await field.text())  # type: ignore
+        else:
+            ret_d[
+                str(field.name)  # type: ignore
+            ] = await field.text()  # type: ignore
+
+
+async def get_upload_instance(
+        request: aiohttp.web.Request,
+        pro: str,
+        cont: str,
+        p_query: typing.Optional[dict] = None,
+) -> upload.ResumableFileUploadProxy:
+    """Return the specific upload proxy for the resumable upload."""
+    session = get_session_id(request)
+
+    if p_query:
+        query: dict = p_query
+    else:
+        query = request.query
+
+    # Check the existence of the dictionary structure
+    try:
+        request.app[session]["uploads"][pro]
+    except KeyError:
+        request.app[session]["uploads"][pro] = {}
+
+    try:
+        request.app[session]["uploads"][pro][cont]
+    except KeyError:
+        request.app[session]["uploads"][pro][cont] = {}
+
+    try:
+        ident = query["resumableIdentifier"]
+    except KeyError:
+        raise aiohttp.web.HTTPBadRequest(reason="Malformed query string")
+    try:
+        upload_session = request.app[session]["uploads"][pro][cont][ident]
+    except KeyError:
+        auth = get_auth_instance(request)
+        upload_session = upload.ResumableFileUploadProxy(
+            auth,
+            query,
+            request.match_info,
+            request.app["client"]
+        )
+        await upload_session.a_check_container()
+        if upload_session.get_segmented():
+            await upload_session.a_sync_segments()
+        request.app[session]["uploads"][pro][cont][ident] = upload_session
+
+    return upload_session
+
+
 def get_path_from_list(
         to_parse: typing.List[str],
         path_prefix: str