diff --git a/docs/source/endpoints/tusupload.md b/docs/source/endpoints/tusupload.md index 5d616740a6..43062b8e97 100644 --- a/docs/source/endpoints/tusupload.md +++ b/docs/source/endpoints/tusupload.md @@ -153,7 +153,121 @@ See the `plone.rest` documentation for more information on how to configure CORS See for a list and description of the individual headers. -## Temporary Upload Directory +## Temporary Upload Directory (local-disk mode) -During upload, files are stored in a temporary directory that by default is located in the `CLIENT_HOME` directory. -If you are using a multi ZEO client setup without session stickiness you *must* configure this to a directory shared by all ZEO clients by setting the `TUS_TMP_FILE_DIR` environment variable, for example `TUS_TMP_FILE_DIR=/tmp/tus-uploads`. +When the ZODB storage is **not** S3-backed, in-progress uploads are buffered to a temporary directory that defaults to `CLIENT_HOME/tus-uploads`. +If you are using a multi-ZEO-client setup without session stickiness in this mode you *must* configure this to a directory shared by all clients by setting the `TUS_TMP_FILE_DIR` environment variable, for example `TUS_TMP_FILE_DIR=/tmp/tus-uploads`. + +This directory is unused when the storage is S3-backed (see below). + + +## S3-backed storage + +When the ZODB storage is `zodb_s3blobs.S3BlobStorage`, the TUS service streams chunks directly into an S3 multipart upload under the `tus-staging/` key prefix. +State for in-progress uploads is held in a ZODB annotation on the upload's container (key `plone.restapi.tus_uploads`), which makes the data path independent of which worker handled which chunk — any worker can serve any `PATCH`, `HEAD`, or `DELETE` for an upload, so sticky-routing is unnecessary. + +A successful final `PATCH` triggers `CompleteMultipartUpload` and hands the staging key off to `S3BlobStorage` for registration; the annotation entry is removed at that point. +Abandoned uploads (browser tab closed mid-chunk, network drop, worker crash) leave dangling multiparts behind. S3 charges for storage of in-progress parts and silently retains them until they're explicitly aborted — so deployments using S3-backed storage should configure a lifecycle rule that auto-aborts stale staging multiparts. + +### Lifecycle rule for abandoned multipart uploads + +The TUS service tries to install the rule on the first `POST` per process via `S3Client.ensure_abort_multipart_lifecycle_rule`. +The call is idempotent (it reads the bucket's existing lifecycle configuration, returns early if a rule with the right ID is already present, otherwise merges its rule with any others and writes the configuration back) and tolerant of failure (a single `WARNING` is logged and uploads continue without the rule). + +Defaults: prefix `tus-staging/`, rule ID `tus-staging-abort`, 7 days. Both `AbortIncompleteMultipartUpload` (drops in-flight parts) and `Expiration` (defensively deletes any orphaned objects under the prefix) are set. + +For auto-apply, the IAM principal needs (in addition to the base zodb-s3blobs permissions): + +- `s3:GetLifecycleConfiguration` +- `s3:PutLifecycleConfiguration` + +If you don't want to grant these, apply the rule manually once per bucket and the auto-apply call will be a no-op (it sees the existing rule and skips the write). + +### Manual setup + +`tus-lifecycle.json`: + +```json +{ + "Rules": [{ + "ID": "tus-staging-abort", + "Status": "Enabled", + "Filter": {"Prefix": "tus-staging/"}, + "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 7}, + "Expiration": {"Days": 7} + }] +} +``` + +```sh +# AWS S3 +aws s3api put-bucket-lifecycle-configuration \ + --bucket \ + --lifecycle-configuration file://tus-lifecycle.json + +# Tigris (S3-compatible) +aws --endpoint-url=https://fly.storage.tigris.dev \ + s3api put-bucket-lifecycle-configuration \ + --bucket \ + --lifecycle-configuration file://tus-lifecycle.json +``` + +**Important:** `put-bucket-lifecycle-configuration` *replaces* all rules for the bucket. +If the bucket already has lifecycle rules for other purposes, fetch them first with `aws s3api get-bucket-lifecycle-configuration --bucket `, append the rule above to the `Rules` array, then put the merged set. +The auto-apply path performs this merge automatically. + + +## Direct-to-S3 data-plane offload (optional) + +By default in S3 mode, every TUS PATCH chunk passes through a Zope worker — Plone reads the request body, calls `upload_part` against S3, and returns 204. +Under high concurrent-upload load this saturates the worker pool because each upload occupies one worker slot continuously. + +The data-plane offload removes this bottleneck by having nginx stream each non-final chunk straight to S3, holding a worker only for a small auth/lookup decision (~50 ms instead of ~hundreds of ms per chunk). +The final chunk still routes through Plone so `CompleteMultipartUpload` and content creation run transactionally. + +### What's involved + +- A new internal endpoint `@tus-authorize` returns a presigned `UploadPart` URL plus routing headers (`X-Route`, `X-S3-Url`, `X-Tus-New-Offset`) for each PATCH. +- nginx's `auth_request` directive consults this endpoint on every `PATCH` to `@tus-upload/` and either: + - streams the request body to S3 via the presigned URL (`X-Route: s3`), or + - forwards the request to Plone unchanged (`X-Route: plone`) — used for the final chunk, local-disk uploads, presign errors, or when the offload is disabled. + +### Enabling the offload + +Two pieces have to line up: + +1. **Plone**: set `TUS_S3_OFFLOAD_ENABLED=1` in the environment of every Zope process. The default is "disabled", in which case `@tus-authorize` always returns `X-Route: plone` and the offload is a no-op (this is the kill switch — flip the env var off and restart to revert without redeploying nginx). +2. **nginx**: deploy a config that runs an `auth_request` against `/_tus_authorize_internal` for `PATCH` requests to `@tus-upload/`, then proxy-passes to either S3 or Plone based on the response headers. See `plone/conf/nginx/nginx.conf` in the deployment repo for a working configuration. + +The Plone-side endpoint is safe to deploy on its own — without the matching nginx config nothing calls it. + +### IAM + +The Plone IAM principal needs `s3:PutObject` on the staging key pattern (which it already has via the base zodb-s3blobs policy). +No new permissions are required — the presigned URL is signed using the same credentials that handle every other S3 operation. + +### Limits and caveats + +- nginx returns the S3 200 response status to the client when the chunk is offloaded; Plone returns 204. + The TUS client-side library (e.g. `@rpldy/chunked-sender`) treats both as success codes, but if you use a custom client check the status handling. +- The `Upload-Offset` response header is computed by Plone in the auth subrequest and emitted by nginx via `add_header`. + nginx is configured to hide whatever value the upstream (Plone or S3) returned and substitute the authoritative value, so the client always sees the correct offset. +- nginx needs a working `resolver` directive to dispatch `proxy_pass` to the (variable) S3 host. + In containerised deployments, point at Docker's embedded DNS (`127.0.0.11`); in cloud deployments use the platform-provided resolver. +- For S3-compatible backends with strict header signing (some non-AWS implementations), the presigned URL flow assumes `UNSIGNED-PAYLOAD` for the body — the default in boto3. + Smoke-test against your specific backend before rolling out widely. + +### Smoke-testing the endpoint + +With the offload enabled, the authorize endpoint is callable directly: + +```sh +curl -i -H "Authorization: Bearer " \ + -H "X-Original-Method: PATCH" \ + -H "X-Original-Upload-Offset: 0" \ + -H "X-Original-Content-Length: 10485760" \ + "https://example.com/path/@tus-authorize/" +``` + +A successful response is `200 OK` with `X-Route: s3`, `X-S3-Url: https://...`, `X-Tus-New-Offset: 10485760`, and an empty body. +With the env var unset you'll get `X-Route: plone` regardless of input — the offload is off. diff --git a/src/plone/restapi/deserializer/dxfields.py b/src/plone/restapi/deserializer/dxfields.py index 04c3c28ce9..29391b4868 100644 --- a/src/plone/restapi/deserializer/dxfields.py +++ b/src/plone/restapi/deserializer/dxfields.py @@ -281,6 +281,13 @@ def __call__(self, value): # very quick operation if tus: tus.process_blob(value._blob) + # The NamedBlobFile was constructed with a 1-byte placeholder and + # may have had its ``size`` cached on __dict__ during validation. + # In S3 mode the on-disk blob is a tiny marker, not the real + # bytes, so without this override ``value.size`` would be pickled + # as the marker length. Prime the cache with the true length so + # the persisted size matches the uploaded file. + value.__dict__["size"] = tus.length() return value diff --git a/src/plone/restapi/services/content/configure.zcml b/src/plone/restapi/services/content/configure.zcml index 72073faafb..929f8780ca 100644 --- a/src/plone/restapi/services/content/configure.zcml +++ b/src/plone/restapi/services/content/configure.zcml @@ -140,4 +140,43 @@ name="@tus-upload" /> + + + + + + + + + diff --git a/src/plone/restapi/services/content/tus.py b/src/plone/restapi/services/content/tus.py index ea4509457d..39dcd36700 100644 --- a/src/plone/restapi/services/content/tus.py +++ b/src/plone/restapi/services/content/tus.py @@ -2,8 +2,11 @@ from Acquisition import aq_base from Acquisition.interfaces import IAcquirer from base64 import b64decode +from BTrees.OOBTree import OOBTree from email.utils import formatdate from fnmatch import fnmatch +from io import BytesIO +from persistent.mapping import PersistentMapping from plone.rest.interfaces import ICORSPolicy from plone.restapi.bbb import base_hasattr from plone.restapi.bbb import safe_hasattr @@ -14,10 +17,13 @@ from plone.restapi.services.content.utils import create from plone.rfc822.interfaces import IPrimaryFieldInfo from Products.CMFCore.utils import getToolByName +from tempfile import mkstemp from uuid import uuid4 from zExceptions import Unauthorized +from zope.annotation.interfaces import IAnnotations from zope.component import queryMultiAdapter from zope.event import notify +from zope.interface import alsoProvides from zope.interface import implementer from zope.lifecycleevent import ObjectCreatedEvent from zope.publisher.interfaces import IPublishTraverse @@ -25,14 +31,135 @@ from ZODB.blob import rename_or_copy_blob import json +import logging import os import time +logger = logging.getLogger(__name__) + + +# Annotation key on the upload's container (folder/site root). Holds an +# OOBTree[uid → PersistentMapping(descriptor)] for in-progress S3 uploads. +ANNOTATION_KEY = "plone.restapi.tus_uploads" + +# S3 staging key prefix and lifecycle rule ID for auto-aborting abandoned +# multipart uploads. Kept aligned with the manual setup instructions +# documented in CLAUDE.md. +TUS_STAGING_PREFIX = "tus-staging/" +TUS_LIFECYCLE_RULE_ID = "tus-staging-abort" +TUS_LIFECYCLE_DAYS = 7 + +# Feature flag controlling the nginx-mediated direct-to-S3 offload. +# When unset/false, the @tus-authorize endpoint always returns +# X-Route: plone, which keeps PATCH bodies flowing through the Zope worker +# (the legacy/safe path). When true, @tus-authorize hands nginx a +# presigned UploadPart URL for non-final chunks; only the final chunk goes +# back through Plone for the transactional CompleteMultipartUpload + +# content-creation step. See the deployment doc for nginx config required +# to make use of this. +TUS_S3_OFFLOAD_ENV = "TUS_S3_OFFLOAD_ENABLED" +TUS_PRESIGNED_URL_EXPIRES = 300 # seconds — generous for slow uploads + + +def _s3_offload_enabled(): + return os.environ.get(TUS_S3_OFFLOAD_ENV, "").lower() in ( + "1", + "true", + "yes", + "on", + ) + +# Process-local set of bucket names already checked / had the rule applied +# during this process's lifetime. Auto-apply runs at most once per bucket +# per process, regardless of success — never block uploads on lifecycle. +_LIFECYCLE_APPLIED: set = set() + + +def _ensure_lifecycle_rule_once(s3_client): + """Apply the abort-multipart lifecycle rule on first POST per process. + + Idempotent within a process; tolerant of failure (logs a warning and + moves on). The S3Client method itself is idempotent against the bucket + too, so even repeated calls would be safe — this just avoids the round + trip after the first call. + """ + bucket = getattr(s3_client, "bucket_name", None) + if not bucket or bucket in _LIFECYCLE_APPLIED: + return + try: + s3_client.ensure_abort_multipart_lifecycle_rule( + rule_id=TUS_LIFECYCLE_RULE_ID, + prefix=TUS_STAGING_PREFIX, + days=TUS_LIFECYCLE_DAYS, + ) + except Exception: + logger.warning( + "TUS/S3: lifecycle setup raised unexpectedly for bucket=%s; " + "continuing without it.", + bucket, + exc_info=True, + ) + # Mark applied either way so we don't retry on every POST. + _LIFECYCLE_APPLIED.add(bucket) + + +def _resolve_s3_blob_storage(context): + """Return (storage, s3_client) if the context's ZODB storage is an + S3BlobStorage, else None. Import-guarded so environments without + zodb-s3blobs installed continue to work. + + Returns the **Connection's** MVCC instance of the storage (``jar._storage``), + not ``db.storage``. ZODB's ``DB.open()`` calls ``new_instance()`` on the + storage for each connection, and commits happen through that per-connection + instance. The S3BlobStorage keeps pending/staged state on the instance, so + registrations made on ``db.storage`` (the primary) would be invisible at + commit time on the per-connection instance. + """ + try: + from zodb_s3blobs.storage import S3BlobStorage + except ImportError: + logger.debug("TUS/S3: zodb_s3blobs not installed; staying in local mode") + return None + jar = getattr(aq_base(context), "_p_jar", None) + if jar is None: + logger.debug("TUS/S3: context has no _p_jar; staying in local mode") + return None + storage = getattr(jar, "_storage", None) + if storage is None: + db = jar.db() + storage = db.storage if db is not None else None + if not isinstance(storage, S3BlobStorage): + logger.debug( + "TUS/S3: storage %r is not S3BlobStorage; staying in local mode", + type(storage).__name__, + ) + return None + logger.debug( + "TUS/S3: resolved per-connection S3BlobStorage id=%s", id(storage) + ) + return storage, storage._s3_client + + +def _container_uploads(container, create=False): + """Return the OOBTree of in-progress S3 uploads for this container. + + If ``create`` is False and no annotation exists, returns None. + """ + annotations = IAnnotations(container) + uploads = annotations.get(ANNOTATION_KEY) + if uploads is None: + if not create: + return None + uploads = OOBTree() + annotations[ANNOTATION_KEY] = uploads + return uploads + + TUS_OPTIONS_RESPONSE_HEADERS = { "Tus-Resumable": "1.0.0", "Tus-Version": "1.0.0", - "Tus-Extension": "creation,expiration", + "Tus-Extension": "creation,expiration,termination", } @@ -87,6 +214,26 @@ def error(self, type, message, status=400): self.request.response.setStatus(status) return {"error": {"type": type, "message": message}} + def disable_csrf_protection(self): + """Mark the request as auto-CSRF-exempt. + + plone.protect's auto-CSRF check fires whenever a request triggers a + ZODB write. The annotation-backed S3 path writes the upload + descriptor on POST and updates ``last_active`` on every PATCH; the + DELETE path removes the annotation. None of these requests carry a + plone.protect token (the TUS protocol has no provision for one), + and the endpoints already require Bearer/Basic auth — so the + protection is redundant. Mirrors the pattern used in + ``plone.restapi.services.upgrade.post`` and similar services. + """ + import plone.protect.interfaces + + if "IDisableCSRFProtection" in dir(plone.protect.interfaces): + alsoProvides( + self.request, + plone.protect.interfaces.IDisableCSRFProtection, + ) + class UploadPost(TUSBaseService): """TUS upload endpoint for creating a new upload resource.""" @@ -95,6 +242,8 @@ def reply(self): if not self.check_tus_version(): return self.unsupported_version() + self.disable_csrf_protection() + length = self.request.getHeader("Upload-Length", "") try: length = int(length) @@ -115,7 +264,19 @@ def reply(self): else: metadata["mode"] = "create" - tus_upload = TUSUpload(uuid4().hex, metadata=metadata) + uid = uuid4().hex + resolved = _resolve_s3_blob_storage(self.context) + if resolved is not None: + _ensure_lifecycle_rule_once(resolved[1]) + tus_upload = S3TUSUpload( + uid, + container=self.context, + metadata=metadata, + s3_storage=resolved[0], + s3_client=resolved[1], + ) + else: + tus_upload = TUSUpload(uid, metadata=metadata) self.request.response.setStatus(201) url = self.request.getURL() @@ -147,11 +308,52 @@ def publishTraverse(self, request, name): def tus_upload(self): if self.uid is None: + logger.warning( + "TUS: %s request with no uid in URL (context=%s)", + self.request.get("REQUEST_METHOD", "?"), + "/".join(self.context.getPhysicalPath()), + ) return + # Prefer the annotation-backed (S3) path: state is shared across + # workers via ZODB, no stickiness required. + uploads = _container_uploads(self.context) + if uploads is not None and self.uid in uploads: + resolved = _resolve_s3_blob_storage(self.context) + if resolved is None: + # Annotation says S3 but storage no longer reports as S3. This + # would be a configuration regression rather than a routing + # miss, so log loudly. + logger.error( + "TUS: annotation entry exists for uid=%s but S3 storage " + "did not resolve. context=%s", + self.uid, + "/".join(self.context.getPhysicalPath()), + ) + return + return S3TUSUpload( + self.uid, + container=self.context, + s3_storage=resolved[0], + s3_client=resolved[1], + ) + + # Local-disk fallback (dev/test or non-S3 deployments). tus_upload = TUSUpload(self.uid) length = tus_upload.length() if length == 0: + logger.warning( + "TUS: upload uid not found. uid=%s method=%s context=%s " + "tmp_dir=%s metadata_exists=%s file_exists=%s " + "annotation_keys=%s", + self.uid, + self.request.get("REQUEST_METHOD", "?"), + "/".join(self.context.getPhysicalPath()), + tus_upload.tmp_dir, + os.path.exists(tus_upload.metadata_path), + os.path.exists(tus_upload.filepath), + list(uploads.keys()) if uploads is not None else None, + ) return return tus_upload @@ -205,6 +407,8 @@ def reply(self): if not self.check_tus_version(): return self.unsupported_version() + self.disable_csrf_protection() + content_type = self.request.getHeader("Content-Type") if content_type != "application/offset+octet-stream": return self.error("Bad Request", "Missing or invalid Content-Type header") @@ -215,10 +419,21 @@ def reply(self): except ValueError: return self.error("Bad Request", "Missing or invalid Upload-Offset header") + content_length_raw = self.request.getHeader("Content-Length", "") + try: + content_length = int(content_length_raw) if content_length_raw else None + except ValueError: + content_length = None + request_body = self.request._file if hasattr(request_body, "raw"): # Unwrap io.BufferedRandom request_body = request_body.raw - tus_upload.write(request_body, offset) + try: + tus_upload.write( + request_body, offset, content_length=content_length + ) + except TUSUploadError as e: + return self.error(e.error_type, str(e), e.status) offset = tus_upload.offset() if tus_upload.finished: @@ -290,7 +505,229 @@ def create_or_modify_content(self, tus_upload): self.request.response.setHeader("Location", obj.absolute_url()) +@implementer(IPublishTraverse) +class UploadAuthorize(UploadFileBase): + """Authorise a single TUS PATCH chunk for direct-to-S3 forwarding. + + This is *not* part of the TUS protocol — it's an internal endpoint used + by nginx's ``auth_request`` directive in the data-plane offload setup. + nginx makes a body-less GET subrequest here for every PATCH the client + sends; the response headers tell nginx whether to: + + * stream the body straight to S3 via a presigned ``UploadPart`` URL + (``X-Route: s3`` + ``X-S3-Url``), or + * forward the request to Plone unchanged (``X-Route: plone``), + which is the path the final chunk takes so Plone can run + ``CompleteMultipartUpload`` + content creation transactionally. + + Behaviour is controlled by the ``TUS_S3_OFFLOAD_ENABLED`` env var. When + disabled (default), every chunk routes back to Plone — equivalent to + not having the offload at all, used as a kill switch. + + The endpoint is auth-checked exactly like the other TUS handlers + (``Add portal content`` / ``Modify portal content``), so a leaked + ``@tus-authorize`` URL conveys no more privilege than a leaked + ``@tus-upload`` URL. + """ + + def _route_to_plone(self, *, reason=None): + """Build a 'send the body to Plone' authorize response.""" + self.request.response.setHeader("X-Route", "plone") + if reason: + self.request.response.setHeader("X-Tus-Authorize-Reason", reason) + self.request.response.setStatus(200, lock=1) + logger.debug( + "TUS/S3: authorize routed to Plone uid=%s reason=%s", + getattr(self, "uid", None), + reason, + ) + return "" + + def _original_int_header(self, *names): + """Return the first parseable header value from ``names``, or None. + + nginx auth_request sub-requests have no original body, so + ``Upload-Offset`` / ``Content-Length`` are forwarded under + ``X-Original-*`` aliases by the nginx config. Accept either name + so the endpoint is callable for ad-hoc testing too. + """ + for name in names: + raw = self.request.getHeader(name, "") + if raw == "" or raw is None: + continue + try: + return int(raw) + except (TypeError, ValueError): + return None + return None + + def reply(self): + # If offload is disabled, route everything back to Plone — this + # makes the endpoint safe to deploy ahead of the nginx config and + # acts as a runtime kill switch. + if not _s3_offload_enabled(): + return self._route_to_plone(reason="offload-disabled") + + tus_upload = self.tus_upload() + if tus_upload is None: + return self.error("Not Found", "", 404) + + metadata = tus_upload.metadata() + self.check_add_modify_permission(metadata.get("mode", "create")) + + # Local-disk uploads can't be offloaded — there's no S3 multipart + # to PUT into. Route to Plone. + if not isinstance(tus_upload, S3TUSUpload): + return self._route_to_plone(reason="local-mode") + + offset = self._original_int_header("X-Original-Upload-Offset", "Upload-Offset") + content_length = self._original_int_header( + "X-Original-Content-Length", "Content-Length" + ) + if offset is None or content_length is None: + return self.error( + "Bad Request", + "Missing or invalid Upload-Offset / Content-Length", + ) + + # ``metadata()`` returns a copy of the persisted entry — it + # contains staging_key, multipart_upload_id, length, chunk_size, + # and the descriptor fields. We don't mutate it here. + entry = tus_upload.metadata() + if not entry: + return self.error("Not Found", "", 404) + + length = int(entry["length"]) + chunk_size = entry.get("chunk_size") + is_final = (offset + content_length) >= length + + # Validate alignment / size against the established chunk_size. + # First PATCH learns chunk_size — we don't reject it here, but we + # do need to know the size to compute a part_number, so we accept + # the client's Content-Length as the chunk size for this request. + if chunk_size is None: + if not is_final and content_length < 5 * 1024 * 1024: + return self.error( + "Bad Request", + "Non-final chunk must be at least 5 MiB", + ) + effective_chunk_size = content_length + else: + if offset % chunk_size != 0: + return self.error( + "Bad Request", + f"Upload-Offset {offset} is not a multiple of " + f"chunk_size {chunk_size}", + ) + if not is_final and content_length != chunk_size: + return self.error( + "Bad Request", + f"Non-final chunk size {content_length} does not match " + f"chunk_size {chunk_size}", + ) + effective_chunk_size = chunk_size + + # Persist chunk_size on the first chunk's authorize, so that when + # the final chunk arrives at Plone (offload-bypassed up to this + # point) it has the correct value for part_number calculation. + # One write per upload — see S3TUSUpload.remember_chunk_size. + # Only this branch writes, so CSRF only needs disabling here + # (nginx's auth_request subrequest carries no CSRF token; the + # endpoint is already gated by the upload's Add/Modify perm). + if chunk_size is None: + self.disable_csrf_protection() + tus_upload.remember_chunk_size(effective_chunk_size) + + # Final chunk goes through Plone — that's where the transactional + # CompleteMultipartUpload + content-creation work happens. + if is_final: + return self._route_to_plone(reason="final-chunk") + + part_number = (offset // effective_chunk_size) + 1 + try: + url = tus_upload._s3_client.generate_upload_part_presigned_url( + entry["staging_key"], + entry["multipart_upload_id"], + part_number, + expires_in=TUS_PRESIGNED_URL_EXPIRES, + ) + except Exception: + logger.exception( + "TUS/S3: presigning failed for uid=%s part=%d; falling " + "back to Plone-routed path for this chunk.", + self.uid, + part_number, + ) + return self._route_to_plone(reason="presign-failed") + + new_offset = offset + content_length + self.request.response.setHeader("X-Route", "s3") + self.request.response.setHeader("X-S3-Url", url) + self.request.response.setHeader("X-Tus-New-Offset", str(new_offset)) + self.request.response.setHeader("X-Tus-Part-Number", str(part_number)) + # Surface whether this was the chunk_size-learning PATCH so an + # operator inspecting headers can tell. nginx doesn't act on it. + if chunk_size is None: + self.request.response.setHeader( + "X-Tus-Chunk-Size-Learned", str(effective_chunk_size) + ) + self.request.response.setStatus(200, lock=1) + logger.debug( + "TUS/S3: authorize routed to S3 uid=%s part=%d offset=%d size=%d " + "presigned_expires_in=%d", + self.uid, + part_number, + offset, + content_length, + TUS_PRESIGNED_URL_EXPIRES, + ) + return "" + + +@implementer(IPublishTraverse) +class UploadDelete(UploadFileBase): + """TUS upload endpoint for handling DELETE (termination) requests.""" + + def reply(self): + tus_upload = self.tus_upload() + if tus_upload is None: + return self.error("Not Found", "", 404) + + metadata = tus_upload.metadata() + self.check_add_modify_permission(metadata.get("mode", "create")) + + if not self.check_tus_version(): + return self.unsupported_version() + + self.disable_csrf_protection() + + tus_upload.close() + tus_upload.cleanup() + self.request.response.setHeader("Tus-Resumable", "1.0.0") + self.request.response.setStatus(204, lock=1) + return self.reply_no_content() + + +class TUSUploadError(Exception): + """Internal error raised by TUS upload classes during ``write()``. + + Carries enough information for the service layer to translate into a + well-formed TUS error response. + """ + + def __init__(self, message, error_type="Bad Request", status=400): + super().__init__(message) + self.error_type = error_type + self.status = status + + class TUSUpload: + """Local-disk TUS upload state. Used in dev/test or non-S3 deployments. + + State lives in ``CLIENT_HOME/tus-uploads/`` (or ``$TUS_TMP_FILE_DIR``). + Bytes accumulate in a single file at ``self.filepath``; metadata is + written alongside as ``.json``. + """ file_prefix = "tus_upload_" expiration_period = 60 * 60 @@ -320,6 +757,15 @@ def initalize(self, metadata): self.cleanup_expired() with open(self.metadata_path, "w") as f: json.dump(metadata, f) + self._metadata = metadata + + def metadata(self): + """Returns the metadata of the current upload.""" + if self._metadata is None: + if os.path.exists(self.metadata_path): + with open(self.metadata_path, "rb") as f: + self._metadata = json.load(f) + return self._metadata or {} def length(self): """Returns the total upload length.""" @@ -334,8 +780,21 @@ def offset(self): return os.path.getsize(self.filepath) return 0 - def write(self, infile, offset=0): - """Write to uploaded file at the given offset.""" + def expires(self): + """Returns the expiration time of the current upload.""" + if os.path.exists(self.filepath): + expiration = os.stat(self.filepath).st_mtime + self.expiration_period + else: + expiration = time.time() + self.expiration_period + return formatdate(expiration, False, True) + + def write(self, infile, offset=0, content_length=None): + """Write to uploaded file at the given offset. + + ``content_length`` is accepted for signature compatibility with + ``S3TUSUpload.write`` but ignored — the local-disk path drains the + body until EOF. + """ mode = "wb" if os.path.exists(self.filepath): mode = "ab+" @@ -362,14 +821,6 @@ def close(self): if self._file is not None and not self._file.closed: self._file.close() - def metadata(self): - """Returns the metadata of the current upload.""" - if self._metadata is None: - if os.path.exists(self.metadata_path): - with open(self.metadata_path, "rb") as f: - self._metadata = json.load(f) - return self._metadata or {} - def cleanup(self): """Remove temporary upload files.""" if os.path.exists(self.filepath): @@ -383,24 +834,446 @@ def cleanup_expired(self): if fnmatch(filename, "tus_upload_*.json"): metadata_path = os.path.join(self.tmp_dir, filename) filepath = metadata_path[:-5] + mtime_src = None + for candidate in (filepath, metadata_path): + if os.path.exists(candidate): + mtime_src = candidate + break + if mtime_src is None: + continue + mtime = os.stat(mtime_src).st_mtime + if (time.time() - mtime) <= self.expiration_period: + continue + os.remove(metadata_path) if os.path.exists(filepath): - mtime = os.stat(filepath).st_mtime - else: - mtime = os.stat(metadata_path).st_mtime + os.remove(filepath) + + def process_blob(self, blob): + """Hand the uploaded bytes off to the ZODB blob's uncommitted file.""" + rename_or_copy_blob(self.filepath, blob._p_blob_uncommitted) + + +class S3TUSUpload(TUSUpload): + """Multi-instance-safe TUS upload backed by S3 multipart + ZODB annotation. + + State lives in two places: + - **Container annotation** (``IAnnotations(container)[ANNOTATION_KEY]``): + a small descriptor (length, filename, content-type, fieldname, mode, + multipart_upload_id, staging_key, chunk_size, created, last_active). + Shared across workers via ZODB; no local disk involvement. + - **S3 multipart upload**: parts themselves. Source of truth for what's + been uploaded; recovered via ``list_parts`` for HEAD/Complete. + + Per-PATCH flow is intentionally tiny: the request body streams straight + into ``upload_part`` with ``part_number = offset / chunk_size + 1``. + There's no local buffer, no flush threshold, no double-pass over data. + Sticky routing is unnecessary because any worker can resolve uid → + annotation → multipart_upload_id and proceed. + """ + + # Subclass init must NOT touch tmp_dir/filepath like the parent does. + file_prefix = TUSUpload.file_prefix + expiration_period = TUSUpload.expiration_period + + def __init__( + self, uid, container, metadata=None, s3_storage=None, s3_client=None + ): + # Deliberately bypass TUSUpload.__init__ — it creates a tmp dir we + # don't need. Set the minimum attributes the parent class methods + # might still touch. + self.uid = uid + self.container = container + self.tmp_dir = None + self.filepath = None + self.metadata_path = None + self._metadata = None + self._file = None + self.finished = False + self._s3_storage = s3_storage + self._s3_client = s3_client + # Set after process_blob hands the staging key to the storage so + # cleanup() knows whether to delete the object on abort. + self._handoff_done = False + + if metadata is not None: + self._initialize(metadata) + + @property + def s3_mode(self): + return True + + @property + def staging_key(self): + return self._entry().get("staging_key") + + @property + def multipart_upload_id(self): + return self._entry().get("multipart_upload_id") + + def _entry(self): + """Return the OOBTree entry for this upload, or empty mapping.""" + uploads = _container_uploads(self.container) + if uploads is None: + return {} + return uploads.get(self.uid) or {} + + def _save_entry(self, entry): + uploads = _container_uploads(self.container, create=True) + uploads[self.uid] = entry + + def remember_chunk_size(self, chunk_size): + """Persist chunk_size on the upload entry if it isn't set yet. + + Called from the authorize endpoint so that when the data-plane + offload is on, the first chunk's size is recorded even though + the PATCH body bypasses Plone. Without this, the first PATCH + that ever reaches Plone (typically the final chunk, which is + smaller) would "learn" the wrong chunk_size from its own + content_length and compute a bogus part_number — leading to + S3 InvalidArgument on UploadPart. + + One write per upload: gated on chunk_size being None, so every + chunk after the first finds it already set and does nothing. + """ + entry = self._entry() + if not entry: + return + if entry.get("chunk_size") is None: + entry["chunk_size"] = int(chunk_size) + entry["last_active"] = time.time() + self._save_entry(entry) + self._metadata = entry + + def _initialize(self, metadata): + """Create a fresh multipart upload and persist the descriptor.""" + self.cleanup_expired() + staging_key = f"{TUS_STAGING_PREFIX}{self.uid}" + upload_id = self._s3_client.create_multipart_upload(staging_key) + now = time.time() + entry = PersistentMapping( + { + # Upload descriptor (immutable once written, except chunk_size + # which is learned from the first PATCH). + "length": int(metadata.get("length", 0)), + "filename": metadata.get("filename", ""), + "content-type": metadata.get( + "content-type", "application/octet-stream" + ), + "@type": metadata.get("@type"), + "fieldname": metadata.get("fieldname"), + "mode": metadata.get("mode", "create"), + "chunk_size": None, + # S3 state. + "staging_key": staging_key, + "multipart_upload_id": upload_id, + # Liveness. + "created": now, + "last_active": now, + } + ) + self._save_entry(entry) + self._metadata = entry + logger.info( + "TUS/S3: created upload uid=%s staging_key=%s upload_id=%s " + "container=%s length=%d", + self.uid, + staging_key, + upload_id, + "/".join(self.container.getPhysicalPath()), + entry["length"], + ) + + def metadata(self): + """Return the descriptor as a dict (compatible with TUSUpload.metadata).""" + if self._metadata is None: + self._metadata = self._entry() + # The deserializer reads keys like 'content-type', 'filename', + # '@type', 'mode' off this — they're stored under the same names. + return dict(self._metadata) if self._metadata else {} + + def length(self): + return int(self._entry().get("length", 0)) + + def offset(self): + """TUS offset = highest contiguous prefix of part numbers × chunk_size. - if (time.time() - mtime) > self.expiration_period: - os.remove(metadata_path) - if os.path.exists(filepath): - os.remove(filepath) + Sparse parts beyond the first gap are ignored — the client will + re-PATCH from there and S3 will overwrite by part number. + + If the upload has been completed in this request, the multipart + upload no longer exists in S3 and ``list_parts`` would raise + ``NoSuchUpload``. Short-circuit to ``length`` in that case — by + definition a completed upload's offset equals its length. + """ + if self.finished: + return self.length() + entry = self._entry() + chunk_size = entry.get("chunk_size") + if not chunk_size: + return 0 + parts = self._s3_client.list_parts( + entry["staging_key"], entry["multipart_upload_id"] + ) + # Highest contiguous run starting from PartNumber=1. + contiguous = 0 + total = 0 + for part in parts: + if part["PartNumber"] != contiguous + 1: + break + contiguous += 1 + total += part["Size"] + return total def expires(self): - """Returns the expiration time of the current upload.""" - if os.path.exists(self.filepath): - expiration = os.stat(self.filepath).st_mtime + self.expiration_period + last_active = self._entry().get("last_active") or time.time() + return formatdate(last_active + self.expiration_period, False, True) + + def write(self, infile, offset=0, content_length=None): + """Stream one chunk straight to S3 as a single multipart part.""" + entry = self._entry() + if not entry: + raise TUSUploadError( + f"Upload {self.uid} not found in container annotation", + error_type="Not Found", + status=404, + ) + + length = int(entry["length"]) + chunk_size = entry.get("chunk_size") + + if content_length is None: + raise TUSUploadError( + "Missing or invalid Content-Length header for PATCH" + ) + + is_final = (offset + content_length) >= length + if chunk_size is None: + # Learn chunk_size from the first PATCH. + if not is_final and content_length < 5 * 1024 * 1024: + raise TUSUploadError( + "Non-final chunk must be at least 5 MiB (S3 multipart " + "minimum). Configure the client to use a larger chunkSize." + ) + chunk_size = content_length else: - expiration = time.time() + self.expiration_period - return formatdate(expiration, False, True) + if offset % chunk_size != 0: + raise TUSUploadError( + f"Upload-Offset {offset} is not a multiple of " + f"chunk_size {chunk_size}" + ) + if not is_final and content_length != chunk_size: + raise TUSUploadError( + f"Non-final chunk size {content_length} does not match " + f"established chunk_size {chunk_size}" + ) + + part_number = (offset // chunk_size) + 1 + # Read exactly content_length bytes into a seekable buffer. boto3's + # SigV4 signer needs to either rewind the body (to compute SHA256 + # then re-read for upload) or hash a bytes-like payload directly; + # a streaming reader without seek/tell falls between those paths + # and triggers `TypeError: object supporting the buffer API required`. + # The request body is already buffered in Zope's tempfile so this + # doesn't add real memory pressure beyond the chunk itself. + data = infile.read(content_length) + if len(data) != content_length: + raise TUSUploadError( + f"Short read on PATCH body: expected {content_length} bytes, " + f"got {len(data)}" + ) + body = BytesIO(data) + etag = self._s3_client.upload_part( + entry["staging_key"], + entry["multipart_upload_id"], + part_number, + body, + ) + logger.info( + "TUS/S3: uploaded part uid=%s part=%d size=%d offset=%d etag=%s", + self.uid, + part_number, + content_length, + offset, + etag, + ) + + # Update annotation: chunk_size on first PATCH; last_active always. + entry["last_active"] = time.time() + if entry.get("chunk_size") is None: + entry["chunk_size"] = chunk_size + self._save_entry(entry) + self._metadata = entry + + if is_final: + self._complete() + + def _complete(self): + """Verify all parts are present, call CompleteMultipartUpload.""" + entry = self._entry() + parts = self._s3_client.list_parts( + entry["staging_key"], entry["multipart_upload_id"] + ) + chunk_size = entry["chunk_size"] + length = entry["length"] + # Expected part count: ceil(length / chunk_size). + expected = (length + chunk_size - 1) // chunk_size + actual_numbers = {p["PartNumber"] for p in parts} + expected_numbers = set(range(1, expected + 1)) + missing = sorted(expected_numbers - actual_numbers) + if missing: + logger.error( + "TUS/S3: cannot complete upload uid=%s — missing parts %s " + "(expected=%d, got=%d)", + self.uid, + missing, + expected, + len(actual_numbers), + ) + raise TUSUploadError( + f"Cannot complete upload: missing parts {missing}", + error_type="Conflict", + status=409, + ) + try: + self._s3_client.complete_multipart_upload( + entry["staging_key"], + entry["multipart_upload_id"], + parts, + ) + except Exception: + logger.exception( + "TUS/S3: complete_multipart_upload failed uid=%s " + "staging_key=%s parts=%d", + self.uid, + entry["staging_key"], + len(parts), + ) + raise + logger.info( + "TUS/S3: completed multipart upload uid=%s parts=%d total_bytes=%d " + "staging_key=%s", + self.uid, + len(parts), + length, + entry["staging_key"], + ) + self.finished = True + + def open(self): + """Not used in S3 mode (deserializer reads via process_blob marker).""" + raise NotImplementedError( + "S3TUSUpload does not expose the uploaded bytes locally" + ) + + def close(self): + """No-op — there is no local file handle in S3 mode.""" + + def cleanup(self): + """Drop annotation entry; abort multipart if upload was abandoned.""" + entry = self._entry() + if entry: + if ( + not self.finished + and entry.get("multipart_upload_id") + and self._s3_client is not None + ): + try: + self._s3_client.abort_multipart_upload( + entry["staging_key"], entry["multipart_upload_id"] + ) + except Exception: + logger.warning( + "TUS/S3: failed to abort multipart for upload %s", + self.uid, + exc_info=True, + ) + if ( + self.finished + and not self._handoff_done + and self._s3_client is not None + ): + # Completed S3 upload but handoff to ZODB blob never ran. + # Leave the orphan for the lifecycle rule to clean up rather + # than risk deleting a key the storage might end up + # referencing. + logger.warning( + "TUS/S3: upload uid=%s completed but handoff was not " + "performed; staging_key=%s left for lifecycle cleanup", + self.uid, + entry.get("staging_key"), + ) + uploads = _container_uploads(self.container) + if uploads is not None and self.uid in uploads: + del uploads[self.uid] + + def cleanup_expired(self): + """Remove annotation entries (and S3 multiparts) older than the period. + + Scoped to this container's annotation only — cheap and targeted, runs + at upload-create time. + """ + uploads = _container_uploads(self.container) + if uploads is None: + return + cutoff = time.time() - self.expiration_period + expired = [ + uid + for uid, entry in uploads.items() + if (entry.get("last_active") or entry.get("created") or 0) < cutoff + ] + for uid in expired: + entry = uploads[uid] + if self._s3_client is not None and entry.get("multipart_upload_id"): + try: + self._s3_client.abort_multipart_upload( + entry["staging_key"], entry["multipart_upload_id"] + ) + except Exception: + logger.debug( + "TUS/S3: expired-upload abort failed (key=%s)", + entry.get("staging_key"), + exc_info=True, + ) + del uploads[uid] def process_blob(self, blob): - rename_or_copy_blob(self.filepath, blob._p_blob_uncommitted) + """Register the staging key with the S3 storage as a blob handoff. + + Mirrors the original implementation: write a tiny diagnostic marker, + rename it into ``_p_blob_uncommitted``, and tell the storage which + S3 key the blob's bytes actually live at. + """ + if self._s3_storage is None: + raise RuntimeError( + "S3TUSUpload missing storage reference at handoff" + ) + size = self.length() + entry = self._entry() + staging_key = entry.get("staging_key") + marker_content = ( + f"S3BLOB-STAGED\n{staging_key}\n{size}\n".encode() + ) + # Use a uniquely-named tempfile so two concurrent uploads in the same + # process never race on the marker source path. + fd, marker_source = mkstemp(prefix="tus-marker-", suffix=".bin") + try: + with os.fdopen(fd, "wb") as f: + f.write(marker_content) + target_path = blob._p_blob_uncommitted + rename_or_copy_blob(marker_source, target_path) + finally: + if os.path.exists(marker_source): + os.remove(marker_source) + self._s3_storage.register_staged_s3_key(target_path, staging_key, size) + self._handoff_done = True + logger.info( + "TUS/S3: handoff complete uid=%s staging_key=%s size=%d " + "marker_path=%s storage_id=%s", + self.uid, + staging_key, + size, + target_path, + id(self._s3_storage), + ) + diff --git a/src/plone/restapi/tests/test_tus.py b/src/plone/restapi/tests/test_tus.py index 838c935a2c..429b407fe8 100644 --- a/src/plone/restapi/tests/test_tus.py +++ b/src/plone/restapi/tests/test_tus.py @@ -11,6 +11,7 @@ from plone.restapi.services.content.tus import TUSUpload from plone.restapi.testing import PLONE_RESTAPI_DX_FUNCTIONAL_TESTING from plone.restapi.testing import RelativeSession +from zope.annotation.interfaces import IAnnotations from zope.component import getGlobalSiteManager from zope.component import provideAdapter from zope.interface import Interface @@ -22,6 +23,7 @@ import os import shutil import tempfile +import time import transaction import unittest @@ -82,7 +84,9 @@ def test_tus_option_headers(self): headers = response.headers self.assertEqual(response.status_code, 204) self.assertEqual(headers["Tus-Version"], "1.0.0") - self.assertEqual(headers["Tus-Extension"], "creation,expiration") + self.assertEqual( + headers["Tus-Extension"], "creation,expiration,termination" + ) self.assertEqual(headers["Tus-Resumable"], "1.0.0") def test_tus_post_without_version_header_returns_412(self): @@ -761,3 +765,618 @@ def tearDown(self): tmp_dir = os.path.join(client_home, "tus-uploads") if os.path.isdir(tmp_dir): shutil.rmtree(tmp_dir) + + +class FakeS3Client: + """In-memory stand-in for ``zodb_s3blobs.s3client.S3Client``. + + Records the parts uploaded for each (key, upload_id) and tracks the + multipart upload lifecycle (created / completed / aborted). Just enough + surface to exercise ``S3TUSUpload``. + """ + + def __init__(self, bucket_name="test-bucket"): + self.bucket_name = bucket_name + # {key: upload_id} + self._uploads = {} + # {(key, upload_id): {part_number: {'ETag': ..., 'Size': ...}}} + self._parts = {} + self._completed = set() + self._aborted = set() + self._next_upload_id = 1 + # Lifecycle state: in-memory rules list + counters for assertions. + self._lifecycle_rules = [] + self._lifecycle_get_calls = 0 + self._lifecycle_put_calls = 0 + # Configurable error injection for tests. + self._lifecycle_get_raises = None + self._lifecycle_put_raises = None + + def generate_upload_part_presigned_url( + self, s3_key, upload_id, part_number, expires_in=300 + ): + # Deterministic stub URL so tests can assert routing/output. + return ( + f"https://fake-s3.example/{s3_key}" + f"?uploadId={upload_id}" + f"&partNumber={part_number}" + f"&X-Amz-Signature=stub-signature" + ) + + def ensure_abort_multipart_lifecycle_rule(self, rule_id, prefix, days=7): + self._lifecycle_get_calls += 1 + if self._lifecycle_get_raises is not None: + raise self._lifecycle_get_raises + if any(r.get("ID") == rule_id for r in self._lifecycle_rules): + return True + self._lifecycle_put_calls += 1 + if self._lifecycle_put_raises is not None: + raise self._lifecycle_put_raises + self._lifecycle_rules.append( + { + "ID": rule_id, + "Status": "Enabled", + "Filter": {"Prefix": prefix}, + "AbortIncompleteMultipartUpload": { + "DaysAfterInitiation": days + }, + "Expiration": {"Days": days}, + } + ) + return True + + def create_multipart_upload(self, s3_key): + upload_id = f"mpu-{self._next_upload_id}" + self._next_upload_id += 1 + self._uploads[s3_key] = upload_id + self._parts[(s3_key, upload_id)] = {} + return upload_id + + def upload_part(self, s3_key, upload_id, part_number, body): + # Drain the body to simulate boto3 reading it. + data = body.read() if hasattr(body, "read") else body + size = len(data) + etag = f'"etag-{s3_key}-{part_number}"' + self._parts[(s3_key, upload_id)][part_number] = { + "ETag": etag, + "Size": size, + "Data": data, + } + return etag + + def list_parts(self, s3_key, upload_id): + parts = self._parts.get((s3_key, upload_id), {}) + return [ + {"PartNumber": n, "ETag": p["ETag"], "Size": p["Size"]} + for n, p in sorted(parts.items()) + ] + + def complete_multipart_upload(self, s3_key, upload_id, parts): + self._completed.add((s3_key, upload_id)) + + def abort_multipart_upload(self, s3_key, upload_id): + self._aborted.add((s3_key, upload_id)) + + +class FakeS3Storage: + """Minimal stand-in for S3BlobStorage.register_staged_s3_key.""" + + def __init__(self): + self.registered = [] + + def register_staged_s3_key(self, blob_path, staging_key, size): + self.registered.append((blob_path, staging_key, size)) + + +class TestS3TUSUpload(unittest.TestCase): + """Unit tests for the annotation-backed S3 TUS path.""" + + layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING + + def setUp(self): + from plone import api as plone_api + + self.app = self.layer["app"] + self.portal = self.layer["portal"] + login(self.portal, SITE_OWNER_NAME) + self.folder = plone_api.content.create( + container=self.portal, + type="Folder", + id="s3tustestfolder", + title="S3 TUS Test Folder", + ) + transaction.commit() + self.s3 = FakeS3Client() + self.storage = FakeS3Storage() + + def _make_upload(self, uid="abc123", length=30 * 1024 * 1024, **extra): + from plone.restapi.services.content.tus import S3TUSUpload + + meta = { + "length": length, + "filename": "test.bin", + "content-type": "application/octet-stream", + "@type": "File", + "mode": "create", + } + meta.update(extra) + return S3TUSUpload( + uid, + container=self.folder, + metadata=meta, + s3_storage=self.storage, + s3_client=self.s3, + ) + + def test_initialization_creates_multipart_and_writes_annotation(self): + from plone.restapi.services.content.tus import ( + ANNOTATION_KEY, + _container_uploads, + ) + + tus = self._make_upload(uid="uid-1") + # Multipart created with deterministic staging key + self.assertEqual(self.s3._uploads, {"tus-staging/uid-1": "mpu-1"}) + # Annotation written + uploads = _container_uploads(self.folder) + self.assertIsNotNone(uploads) + self.assertIn("uid-1", uploads) + entry = uploads["uid-1"] + self.assertEqual(entry["staging_key"], "tus-staging/uid-1") + self.assertEqual(entry["multipart_upload_id"], "mpu-1") + self.assertEqual(entry["length"], 30 * 1024 * 1024) + self.assertIsNone(entry["chunk_size"]) + # Sanity: ANNOTATION_KEY is the right place + self.assertIn(ANNOTATION_KEY, IAnnotations(self.folder)) + + def test_metadata_and_length_read_from_annotation(self): + tus = self._make_upload(filename="report.pdf", length=12345) + self.assertEqual(tus.length(), 12345) + self.assertEqual(tus.metadata()["filename"], "report.pdf") + + def test_offset_zero_until_first_chunk_uploaded(self): + tus = self._make_upload() + self.assertEqual(tus.offset(), 0) + + def test_offset_uses_contiguous_prefix_of_parts(self): + # 30 MiB total, 10 MiB chunks. Upload parts 1, 2, and 4 — gap at 3. + # Contiguous offset = 2 * chunk_size. + chunk = 10 * 1024 * 1024 + tus = self._make_upload(length=4 * chunk) + # Manually inject parts since we don't have a real request body here. + key = "tus-staging/abc123" + upload_id = "mpu-1" + for part_number in (1, 2, 4): + self.s3._parts[(key, upload_id)][part_number] = { + "ETag": f'"e-{part_number}"', + "Size": chunk, + "Data": b"", + } + # Set chunk_size in the annotation so offset() uses it. + from plone.restapi.services.content.tus import _container_uploads + + entry = _container_uploads(self.folder)["abc123"] + entry["chunk_size"] = chunk + _container_uploads(self.folder)["abc123"] = entry + self.assertEqual(tus.offset(), 2 * chunk) + + def test_write_uses_deterministic_part_number(self): + chunk = 10 * 1024 * 1024 + length = 3 * chunk + tus = self._make_upload(uid="dpn", length=length) + # First chunk at offset 0. + tus.write(BytesIO(b"x" * chunk), offset=0, content_length=chunk) + # Second chunk at offset chunk. + tus.write(BytesIO(b"y" * chunk), offset=chunk, content_length=chunk) + parts = self.s3._parts[("tus-staging/dpn", "mpu-1")] + self.assertEqual(sorted(parts.keys()), [1, 2]) + self.assertEqual(parts[1]["Size"], chunk) + self.assertEqual(parts[2]["Size"], chunk) + + def test_write_first_patch_learns_chunk_size(self): + from plone.restapi.services.content.tus import _container_uploads + + chunk = 10 * 1024 * 1024 + tus = self._make_upload(uid="learn", length=3 * chunk) + tus.write(BytesIO(b"x" * chunk), offset=0, content_length=chunk) + entry = _container_uploads(self.folder)["learn"] + self.assertEqual(entry["chunk_size"], chunk) + + def test_write_rejects_non_final_chunk_below_5mib(self): + from plone.restapi.services.content.tus import TUSUploadError + + tus = self._make_upload(uid="small", length=100 * 1024 * 1024) + small = 4 * 1024 * 1024 # below 5 MiB + with self.assertRaises(TUSUploadError) as ctx: + tus.write(BytesIO(b"x" * small), offset=0, content_length=small) + self.assertIn("5 MiB", str(ctx.exception)) + + def test_write_rejects_offset_not_multiple_of_chunk_size(self): + from plone.restapi.services.content.tus import TUSUploadError + + chunk = 10 * 1024 * 1024 + tus = self._make_upload(uid="badoff", length=3 * chunk) + tus.write(BytesIO(b"x" * chunk), offset=0, content_length=chunk) + # Now send at a misaligned offset. + with self.assertRaises(TUSUploadError) as ctx: + tus.write( + BytesIO(b"y" * chunk), + offset=chunk + 1, + content_length=chunk, + ) + self.assertIn("not a multiple", str(ctx.exception)) + + def test_write_rejects_size_mismatch_for_non_final_chunk(self): + from plone.restapi.services.content.tus import TUSUploadError + + chunk = 10 * 1024 * 1024 + tus = self._make_upload(uid="badsize", length=3 * chunk) + tus.write(BytesIO(b"x" * chunk), offset=0, content_length=chunk) + # Subsequent non-final PATCH must equal chunk_size. + bad = chunk - 100 + with self.assertRaises(TUSUploadError) as ctx: + tus.write(BytesIO(b"y" * bad), offset=chunk, content_length=bad) + self.assertIn("does not match", str(ctx.exception)) + + def test_final_chunk_completes_multipart(self): + chunk = 10 * 1024 * 1024 + length = 2 * chunk + 100 # final part is small (100 bytes) + tus = self._make_upload(uid="finish", length=length) + tus.write(BytesIO(b"a" * chunk), offset=0, content_length=chunk) + tus.write(BytesIO(b"b" * chunk), offset=chunk, content_length=chunk) + # Final small chunk + tus.write(BytesIO(b"c" * 100), offset=2 * chunk, content_length=100) + self.assertTrue(tus.finished) + self.assertIn( + ("tus-staging/finish", "mpu-1"), self.s3._completed + ) + + def test_complete_with_missing_parts_raises_409(self): + from plone.restapi.services.content.tus import TUSUploadError + + chunk = 10 * 1024 * 1024 + length = 3 * chunk + tus = self._make_upload(uid="gap", length=length) + tus.write(BytesIO(b"a" * chunk), offset=0, content_length=chunk) + # Skip middle chunk; pretend client thinks last chunk completes + # (length boundary). The write at offset=2*chunk would set is_final. + with self.assertRaises(TUSUploadError) as ctx: + tus.write( + BytesIO(b"c" * chunk), + offset=2 * chunk, + content_length=chunk, + ) + self.assertEqual(ctx.exception.status, 409) + self.assertIn("missing parts", str(ctx.exception)) + + def test_cleanup_unfinished_aborts_multipart_and_drops_entry(self): + from plone.restapi.services.content.tus import _container_uploads + + tus = self._make_upload(uid="abort") + tus.cleanup() + self.assertIn( + ("tus-staging/abort", "mpu-1"), self.s3._aborted + ) + self.assertNotIn("abort", _container_uploads(self.folder)) + + def test_cleanup_finished_does_not_abort(self): + chunk = 10 * 1024 * 1024 + length = chunk + tus = self._make_upload(uid="ok", length=length) + tus.write(BytesIO(b"x" * chunk), offset=0, content_length=chunk) + self.assertTrue(tus.finished) + # Simulate handoff completing before cleanup + tus._handoff_done = True + tus.cleanup() + self.assertNotIn( + ("tus-staging/ok", "mpu-1"), self.s3._aborted + ) + + def test_cleanup_expired_removes_old_entries(self): + from plone.restapi.services.content.tus import _container_uploads + + tus_old = self._make_upload(uid="old") + tus_new = self._make_upload(uid="new") + # Backdate "old" past the expiration period. + uploads = _container_uploads(self.folder) + uploads["old"]["last_active"] = ( + time.time() - tus_old.expiration_period - 1 + ) + # Triggering cleanup_expired through any S3TUSUpload instance. + tus_new.cleanup_expired() + self.assertNotIn("old", _container_uploads(self.folder)) + self.assertIn("new", _container_uploads(self.folder)) + self.assertIn( + ("tus-staging/old", "mpu-1"), self.s3._aborted + ) + + def tearDown(self): + from plone.restapi.services.content.tus import ( + ANNOTATION_KEY, + ) + + ann = IAnnotations(self.folder) + if ANNOTATION_KEY in ann: + del ann[ANNOTATION_KEY] + # Tear the folder down so the next test starts clean. + api.content.delete(obj=self.folder) + transaction.commit() + + +class _FakeResponse: + def __init__(self): + self.headers = {} + self.status = 200 + + def setHeader(self, name, value): + self.headers[name] = value + + def getHeader(self, name, default=None): + return self.headers.get(name, default) + + def setStatus(self, status, lock=0): + self.status = status + + def setBody(self, body, is_error=0): + self.body = body + + +class _FakeRequest: + """Minimal stand-in for a Zope/Plone request object. + + Exposes the surface UploadAuthorize touches: getHeader, response, plus + the dict-like ``get`` for environment-style lookups (REQUEST_METHOD). + """ + + def __init__(self, headers=None, method="GET"): + self._headers = dict(headers or {}) + self.response = _FakeResponse() + self._env = {"REQUEST_METHOD": method} + self._rest_cors_preflight = False + + def getHeader(self, name, default=None): + for k, v in self._headers.items(): + if k.lower() == name.lower(): + return v + return default + + def get(self, key, default=None): + return self._env.get(key, default) + + +class TestUploadAuthorize(unittest.TestCase): + """Unit tests for the @tus-authorize endpoint's routing logic.""" + + layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING + + def setUp(self): + from plone import api as plone_api + + self.app = self.layer["app"] + self.portal = self.layer["portal"] + login(self.portal, SITE_OWNER_NAME) + self.folder = plone_api.content.create( + container=self.portal, + type="Folder", + id="authorizetest", + title="Authorize Test", + ) + transaction.commit() + self.s3 = FakeS3Client(bucket_name="authorize-bucket") + self.storage = FakeS3Storage() + # Force the offload on for this class. Tests that need it off can + # delete the env var locally. + os.environ["TUS_S3_OFFLOAD_ENABLED"] = "1" + + def tearDown(self): + from plone.restapi.services.content.tus import ANNOTATION_KEY + + os.environ.pop("TUS_S3_OFFLOAD_ENABLED", None) + ann = IAnnotations(self.folder) + if ANNOTATION_KEY in ann: + del ann[ANNOTATION_KEY] + api.content.delete(obj=self.folder) + transaction.commit() + + def _make_s3_upload(self, uid, length, chunk_size=None): + from plone.restapi.services.content.tus import S3TUSUpload + + meta = { + "length": length, + "filename": "x.bin", + "content-type": "application/octet-stream", + "@type": "File", + "mode": "create", + } + tus = S3TUSUpload( + uid, + container=self.folder, + metadata=meta, + s3_storage=self.storage, + s3_client=self.s3, + ) + if chunk_size is not None: + from plone.restapi.services.content.tus import _container_uploads + + _container_uploads(self.folder)[uid]["chunk_size"] = chunk_size + return tus + + def _authorize(self, uid, headers): + """Invoke UploadAuthorize.reply() with patched S3 resolution. + + Bypasses ``__init__`` (the ``Service`` base inherits ``object``'s + zero-arg init and relies on the publisher to set ``context`` / + ``request``); we set them by hand for the unit test. + """ + from plone.restapi.services.content import tus as tus_module + from plone.restapi.services.content.tus import UploadAuthorize + + request = _FakeRequest(headers=headers) + service = UploadAuthorize.__new__(UploadAuthorize) + service.context = self.folder + service.request = request + service.uid = uid + service.__name__ = "@tus-authorize" + original_resolve = tus_module._resolve_s3_blob_storage + tus_module._resolve_s3_blob_storage = lambda ctx: ( + self.storage, + self.s3, + ) + try: + body = service.reply() + finally: + tus_module._resolve_s3_blob_storage = original_resolve + return request.response, body + + def test_offload_disabled_routes_to_plone(self): + os.environ.pop("TUS_S3_OFFLOAD_ENABLED", None) + self._make_s3_upload("uid1", length=20 * 1024 * 1024) + response, _ = self._authorize( + "uid1", + { + "X-Original-Upload-Offset": "0", + "X-Original-Content-Length": str(10 * 1024 * 1024), + }, + ) + self.assertEqual(response.headers.get("X-Route"), "plone") + self.assertEqual(response.status, 200) + self.assertNotIn("X-S3-Url", response.headers) + + def test_unknown_uid_returns_404(self): + # No upload created — annotation lookup will miss. + response, _ = self._authorize( + "missing-uid", + { + "X-Original-Upload-Offset": "0", + "X-Original-Content-Length": "10485760", + }, + ) + self.assertEqual(response.status, 404) + + def test_first_chunk_returns_presigned_url(self): + chunk = 10 * 1024 * 1024 + self._make_s3_upload("uid2", length=3 * chunk) + response, _ = self._authorize( + "uid2", + { + "X-Original-Upload-Offset": "0", + "X-Original-Content-Length": str(chunk), + }, + ) + self.assertEqual(response.headers.get("X-Route"), "s3") + self.assertIn( + "tus-staging/uid2", response.headers.get("X-S3-Url", "") + ) + self.assertIn( + "partNumber=1", response.headers.get("X-S3-Url", "") + ) + self.assertEqual( + response.headers.get("X-Tus-New-Offset"), str(chunk) + ) + self.assertEqual(response.headers.get("X-Tus-Part-Number"), "1") + + def test_subsequent_chunk_uses_correct_part_number(self): + chunk = 10 * 1024 * 1024 + self._make_s3_upload("uid3", length=3 * chunk, chunk_size=chunk) + response, _ = self._authorize( + "uid3", + { + "X-Original-Upload-Offset": str(chunk), + "X-Original-Content-Length": str(chunk), + }, + ) + self.assertEqual(response.headers.get("X-Tus-Part-Number"), "2") + self.assertIn( + "partNumber=2", response.headers.get("X-S3-Url", "") + ) + self.assertEqual( + response.headers.get("X-Tus-New-Offset"), str(2 * chunk) + ) + + def test_final_chunk_routes_to_plone(self): + chunk = 10 * 1024 * 1024 + # Length = 2 chunks + small tail; the third PATCH would complete it. + length = 2 * chunk + 100 + self._make_s3_upload("uid4", length=length, chunk_size=chunk) + response, _ = self._authorize( + "uid4", + { + "X-Original-Upload-Offset": str(2 * chunk), + "X-Original-Content-Length": "100", + }, + ) + self.assertEqual(response.headers.get("X-Route"), "plone") + self.assertEqual( + response.headers.get("X-Tus-Authorize-Reason"), "final-chunk" + ) + + def test_misaligned_offset_rejected(self): + chunk = 10 * 1024 * 1024 + self._make_s3_upload("uid5", length=3 * chunk, chunk_size=chunk) + response, _ = self._authorize( + "uid5", + { + "X-Original-Upload-Offset": str(chunk + 1), + "X-Original-Content-Length": str(chunk), + }, + ) + self.assertEqual(response.status, 400) + + +class TestTUSLifecycleAutoApply(unittest.TestCase): + """Tests for the once-per-process lifecycle rule auto-apply.""" + + def setUp(self): + # Reset the process-local set between tests. + from plone.restapi.services.content.tus import _LIFECYCLE_APPLIED + + _LIFECYCLE_APPLIED.clear() + self.s3 = FakeS3Client(bucket_name="bucket-a") + + def test_first_call_applies_and_marks_bucket(self): + from plone.restapi.services.content.tus import ( + _ensure_lifecycle_rule_once, + _LIFECYCLE_APPLIED, + ) + + _ensure_lifecycle_rule_once(self.s3) + self.assertEqual(self.s3._lifecycle_get_calls, 1) + self.assertEqual(self.s3._lifecycle_put_calls, 1) + self.assertIn("bucket-a", _LIFECYCLE_APPLIED) + + def test_second_call_is_a_noop(self): + from plone.restapi.services.content.tus import ( + _ensure_lifecycle_rule_once, + ) + + _ensure_lifecycle_rule_once(self.s3) + _ensure_lifecycle_rule_once(self.s3) + # Still only one round trip — the second call is short-circuited + # by the process-local set. + self.assertEqual(self.s3._lifecycle_get_calls, 1) + self.assertEqual(self.s3._lifecycle_put_calls, 1) + + def test_failure_is_swallowed_and_bucket_still_marked(self): + from plone.restapi.services.content.tus import ( + _ensure_lifecycle_rule_once, + _LIFECYCLE_APPLIED, + ) + + self.s3._lifecycle_get_raises = RuntimeError("boom") + # Should not raise; should still record the bucket so subsequent + # POSTs don't keep retrying. + _ensure_lifecycle_rule_once(self.s3) + self.assertIn("bucket-a", _LIFECYCLE_APPLIED) + + def test_separate_buckets_are_each_handled_once(self): + from plone.restapi.services.content.tus import ( + _ensure_lifecycle_rule_once, + ) + + s3_b = FakeS3Client(bucket_name="bucket-b") + _ensure_lifecycle_rule_once(self.s3) + _ensure_lifecycle_rule_once(s3_b) + _ensure_lifecycle_rule_once(self.s3) + _ensure_lifecycle_rule_once(s3_b) + self.assertEqual(self.s3._lifecycle_get_calls, 1) + self.assertEqual(s3_b._lifecycle_get_calls, 1)