diff --git a/Dockerfile b/Dockerfile index 43cbd38..553da1c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -30,9 +30,6 @@ USER gen3 # Final stage FROM base -COPY --from=builder /${appname} /${appname} -COPY --from=builder /venv /venv - # switch to root user to install vim USER root @@ -41,6 +38,9 @@ RUN dnf -y install vim # Switch to non-root user 'gen3' for the serving process USER gen3 +COPY --from=builder /${appname} /${appname} +COPY --from=builder /venv /venv + WORKDIR /${appname} RUN chmod 755 bin/run.sh diff --git a/gen3workflow/routes/s3.py b/gen3workflow/routes/s3.py index 860f658..e8d6bbe 100644 --- a/gen3workflow/routes/s3.py +++ b/gen3workflow/routes/s3.py @@ -10,9 +10,10 @@ from fastapi.security import HTTPAuthorizationCredentials from botocore.credentials import Credentials import hmac +from starlette.background import BackgroundTask from starlette.datastructures import Headers from starlette.requests import ClientDisconnect -from starlette.responses import Response +from starlette.responses import Response, StreamingResponse from starlette.status import ( HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED, @@ -203,7 +204,7 @@ async def s3_endpoint(path: str, request: Request): # "All buckets" listing requests also land here and are not supported, since users can only # access their own bucket. if request.method == "GET" and path in ("", "s3"): - err_msg = f"If you are using the S3 endpoint: 's3 ls' not supported, use 's3 ls s3://' instead. If you are trying to reach the Gen3-Workflow API, try '/_status'." + err_msg = f"'{request.method} /{path}': If you are using the S3 endpoint: 's3 ls' not supported, use 's3 ls s3://' instead. If you are trying to reach the Gen3-Workflow API, try '/_status'." logger.error(err_msg) raise HTTPException(HTTP_400_BAD_REQUEST, err_msg) @@ -432,12 +433,15 @@ async def s3_endpoint(path: str, request: Request): proceed = True exception = None try: - response = await request.app.async_client.request( - method=request.method, - url=s3_api_url, - headers=out_headers, - params=query_params, - data=body, + response = await request.app.async_client.send( + request.app.async_client.build_request( + method=request.method, + url=s3_api_url, + headers=out_headers, + params=query_params, + content=body, + ), + stream=True, ) if response.status_code >= 300: @@ -492,38 +496,68 @@ async def s3_endpoint(path: str, request: Request): await asyncio.sleep(delay) # Return the response from AWS S3. - # - mask the details of 403 errors from the end user: authentication is done internally by this + # Return all the headers from the AWS response, except: + # - hop-by-hop headers (apply only to a single transport connection and should be stripped by + # proxies). + # - `x-amz-bucket-region` which for some reason causes this error for tasks ran through + # Nextflow: `The AWS Access Key Id you provided does not exist in our records`. + # - `x-amz-decoded-content-length`: request-direction header; should not appear on responses. + filtered_headers = { + h: v + for h, v in response.headers.items() + if h.lower() + not in { + "x-amz-bucket-region", + "x-amz-decoded-content-length", + # hop-by-hop headers: + "connection", + "keep-alive", + "transfer-encoding", + "te", + "trailer", + "upgrade", + "proxy-authenticate", + "proxy-authorization", + } + } + + # mask the details of 403 errors from the end user: authentication is done internally by this # function, so 403 errors are internal service errors. - # - return all the headers from the AWS response, except: - # - `x-amz-bucket-region` which for some reason causes this error for tasks ran through - # Nextflow: `The AWS Access Key Id you provided does not exist in our records`. - # - `content-length`/`x-amz-decoded-content-length`: when the `content-length` is provided, - # Starlette's Response does not recompute it from the actual content bytes. Example case: if - # the S3 server is Minio, the `content-length` header for a HEAD request can describe what - # a GET on that object _would_ return. Recomputing it is safer. - # - `content-encoding`: forwarding response headers that describe the original bytes - # returned by the S3 server can cause a mismatch, because our httpx client may not return - # those original bytes. Example case: httpx transparently decompresses gzip content. - # `response.content` contains the decoded bytes, while the `content-encoding` header still - # says gzip. - # Alternatively, we could get the raw response bytes (no httpx post-handling) and keep the - # original headers: replace the `client.request` call with `client.build_request` + - # `client.send(..., stream=True)`, and get the response bytes with `response.aiter_raw()`. - # Downside: holding the body in memory instead of benefiting from httpx's streaming. - return Response( - content=( - response.content if response.status_code != HTTP_403_FORBIDDEN else None - ), + if response.status_code == HTTP_403_FORBIDDEN: + await response.aclose() # discard the body we're not returning + return Response(status_code=403, headers=filtered_headers) + + if response.headers.get("content-encoding", "").lower() == "gzip": + # the backend compressed this response (e.g. Minio does this for small, compressible + # bodies like XML listings), httpx decodes it for us + filtered_headers = { + h: v + for h, v in filtered_headers.items() + # Filter out more headers: + # - `content-length`: when it is provided, Starlette's Response does not recompute it + # from the actual content bytes. Example case: if the S3 server is Minio, the + # `content-length` header for a HEAD request can describe what a GET on that object + # _would_ return. Recomputing it is safer. + # - `content-encoding`: forwarding response headers that describe the original bytes + # returned by the S3 server can cause a mismatch, because our httpx client may not + # return those original bytes. Example case: here, httpx transparently decompresses + # gzip content. `response.content` contains the decoded bytes, while the + # `content-encoding` header still says gzip. + if h.lower() not in {"content-length", "content-encoding"} + } + decoded = await response.aread() + await response.aclose() + return Response( + content=decoded, + status_code=response.status_code, + headers=filtered_headers, + ) + + # the response is not compressed: stream the raw response bytes (skip the automatic httpx + # post-handling) + return StreamingResponse( + response.aiter_raw(), status_code=response.status_code, - headers={ - k: v - for k, v in response.headers.items() - if k.lower() - not in { - "x-amz-bucket-region", - "content-length", - "x-amz-decoded-content-length", - "content-encoding", - } - }, + headers=filtered_headers, + background=BackgroundTask(response.aclose), ) diff --git a/tests/conftest.py b/tests/conftest.py index 9960083..be499dd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -255,9 +255,12 @@ def mock_aws_s3_request_function(url: str): if f"?uploadId={upload_id}&partNumber=" in url: headers["etag"] = "test-etag" + content_bytes = resp_xml.encode("utf-8") + headers["content-length"] = str(len(content_bytes)) + return httpx.Response( status_code=200, - text=resp_xml, + stream=httpx.ByteStream(content_bytes), headers=headers, )