Backport/bucket name restriction (#142)

* Allow colon characters in bucket names

* Change bucket id validation in s3 only and add test

* Version bump

---------

Co-authored-by: Christoph Zwerschke <c.zwerschke@dkfz-heidelberg.de>

.pyproject_generation/pyproject_custom.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hexkit"
-version = "3.7.0"
+version = "3.8.0"
 description = "A Toolkit for Building Microservices using the Hexagonal Architecture"
 requires-python = ">=3.9"
 classifiers = [

pyproject.toml

@@ -23,7 +23,7 @@ classifiers = [
     "Intended Audience :: Developers",
 ]
 name = "hexkit"
-version = "3.7.0"
+version = "3.8.0"
 description = "A Toolkit for Building Microservices using the Hexagonal Architecture"
 dependencies = [
     "pydantic >=2, <3",

src/hexkit/providers/s3/provider.py

@@ -22,6 +22,7 @@
 # ruff: noqa: PLR0913
 
 import asyncio
+import re
 from functools import lru_cache
 from pathlib import Path
 from typing import Any, Optional
@@ -31,6 +32,7 @@
 import botocore.config
 import botocore.configloader
 import botocore.exceptions
+import botocore.handlers
 from boto3.s3.transfer import TransferConfig
 from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings
@@ -39,6 +41,10 @@
 from hexkit.utils import calc_part_size
 
 __all__ = ["ObjectStorageProtocol", "PresignedPostURL"]
+# Allow colon character in bucket names to accommodate Ceph multi tenancy S3
+botocore.handlers.VALID_BUCKET = re.compile(
+    r"^(?:[a-zA-Z0-9_]{1,191}:)?[a-z0-9\-]{3,63}$"
+)
 
 
 class S3Config(BaseSettings):
@@ -116,6 +122,8 @@ def read_aws_config_ini(aws_config_ini: Path) -> botocore.config.Config:
 class S3ObjectStorage(ObjectStorageProtocol):
     """S3-based provider implementing the ObjectStorageProtocol."""
 
+    _re_bucket_id = botocore.handlers.VALID_BUCKET
+
     def __init__(
         self,
         *,

tests/integration/test_s3.py

@@ -105,6 +105,19 @@ async def test_object_existence_checks(s3: S3Fixture, tmp_file: FileObject):  #
     )
 
 
+async def test_bucket_name_with_tenant(s3: S3Fixture):
+    """Test if bucket names containing a tenant work correctly."""
+    check_bucket = s3.storage.does_bucket_exist
+    assert not await check_bucket("non-existing-bucket")
+    assert not await check_bucket("tenant:non-existing-bucket")
+    with pytest.raises(ObjectStorageProtocol.BucketIdValidationError):
+        assert not await check_bucket("tenant:invalid:bucket")
+    with pytest.raises(ObjectStorageProtocol.BucketIdValidationError):
+        assert not await check_bucket("tenant-invalid:bucket-valid")
+    with pytest.raises(ObjectStorageProtocol.BucketIdValidationError):
+        assert not await check_bucket("tenant_valid:bucket_invalid")
+
+
 async def test_get_object_etag(s3: S3Fixture, tmp_file: FileObject):  # noqa: F811
     """Test ETag retrieval."""
     await s3.populate_file_objects([tmp_file])