fix issue #56098 cross service copy

modules/invenio-files-rest/invenio_files_rest/config.py

@@ -144,3 +144,17 @@
     'INVENIO_ROLE_COMMUNITY'
 ]
 """The version update roles."""
+
+FILES_REST_STORAGE_SERVICE_PATTERN = {
+    "aws_s3": [
+        r"^s3\.(?P<region>[a-z0-9-]+)\.amazonaws\.com$",
+        r"^(?P<bucket>.+)\.s3\.(?P<region>[a-z0-9-]+)\.amazonaws\.com$",
+        r"^(?P<bucket>.+)\.s3\.amazonaws\.com$",
+        r"^s3\.amazonaws\.com$",
+    ],
+    "wasabi": [
+        r"^s3\.wasabisys\.com$",
+        r"^s3\.(?P<region>[a-z0-9-]+)\.wasabisys\.com$",
+    ],
+}
+"""Storage service patterns for parsing storage host."""

modules/invenio-files-rest/invenio_files_rest/utils.py

@@ -172,6 +172,26 @@ def update_ogp_image(ogp_image, file_uri):
     return src.uri if src else None
 
 
+def parse_storage_host(host):
+    """Parse storage host to get service and parameters.
+    Args:
+        host (str): Storage host.
+    Returns:
+        dict: Dictionary with service and parameters.
+    """
+    service_patterns = current_app.config.get(
+        "FILES_REST_STORAGE_SERVICE_PATTERN", {})
+    for service, patterns in service_patterns.items():
+        for pattern in patterns:
+            result = re.match(pattern, host)
+            if result:
+                return {
+                    "service": service,
+                    "params": result.groupdict()
+                }
+    return {"service": None, "params": {}}
+
+
 def create_boto3_s3_client(access_key, secret_key, region_name=None,
                            endpoint_url=None, client_config={}):
     """Create boto3 S3 client.
@@ -202,20 +222,16 @@ def create_boto3_s3_client(access_key, secret_key, region_name=None,
     # Check endpoint
     if endpoint_url:
         parsed_url = urlparse(endpoint_url)
-        is_aws = re.match(r'.*\.amazonaws\.com$', parsed_url.netloc)
+        service_info = parse_storage_host(parsed_url.netloc)
         current_app.logger.debug(
-            "Endpoint URL '%s' is AWS: %s", endpoint_url, is_aws)
+            "Storage service info: %s", service_info)
         # Only set endpoint_url if it matches the pattern
         # Note: AWS S3 requires no endpoint_url for standard regions
-        if not is_aws:
+        if service_info["service"] != "aws_s3":
             client_kwargs["endpoint_url"] = endpoint_url
-        elif not region_name:
-            # get region from endpoint url for aws s3
-            # ex: s3.us-west-2.amazonaws.com
-            pattern = r'^s3[.-](?P<region>[a-z0-9-]+)\.amazonaws\.com$'
-            result = re.match(pattern, parsed_url.netloc)
-            if result:
-                region_name = result.group('region') or None
+        elif service_info["service"] == "aws_s3" and \
+            "region" in service_info["params"]:
+            region_name = region_name or service_info["params"]["region"]
 
     # Check region
     if region_name:

modules/invenio-files-rest/tests/test_utils.py

@@ -9,8 +9,10 @@
 """Files download/upload REST API similar to S3 for Invenio."""
 
 import botocore.client
-from invenio_files_rest.utils import create_boto3_s3_client
+from invenio_files_rest.utils import create_boto3_s3_client, parse_storage_host
 
+
+# def create_boto3_s3_client(access_key, secret_key, region_name=None, endpoint_url=None, client_config={}):
 # .tox/c1/bin/pytest --cov=invenio_files_rest tests/test_utils.py::test_create_boto3_s3_client -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-files-rest/.tox/c1/tmp
 def test_create_boto3_s3_client(app):
     # Test case (Pos): S3 credentials are not configured
@@ -27,6 +29,17 @@ def test_create_boto3_s3_client(app):
     assert client.meta.region_name == "eu-west-1"
     assert client.meta.endpoint_url == "https://s3.eu-west-1.amazonaws.com"
 
+    # Set app config for testing
+    app.config.update({
+        "FILES_REST_STORAGE_SERVICE_PATTERN": {
+            "aws_s3": [
+                r"^s3\.amazonaws\.com$",
+                r"^s3\.(?P<region>.+)\.amazonaws\.com$",
+                r"^(?P<bucket>.+)\.s3\.(?P<region>.+)\.amazonaws\.com$"
+            ],
+        }
+    })
+
     # Test case (Pos): region_name is set
     client = create_boto3_s3_client(
         "access_key", "secret_key",
@@ -61,3 +74,63 @@ def test_create_boto3_s3_client(app):
     assert isinstance(client, botocore.client.BaseClient)
     assert client.meta.service_model.service_name == "s3"
     assert client.meta.endpoint_url == "https://example.com"
+
+
+# def parse_storage_host(host):
+# .tox/c1/bin/pytest --cov=invenio_files_rest tests/test_utils.py::test_parse_storage_host_match_service -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-files-rest/.tox/c1/tmp
+def test_parse_storage_host_match_service(app):
+    # サービスパターン設定
+    patterns = {
+        "aws_s3": [
+            r"^s3\.amazonaws\.com$",
+            r"^(?P<bucket>.+)\.s3\.amazonaws\.com$"
+        ],
+        "wasabi": [
+            r"^s3\.wasabisys\.com$",
+            r"^s3\.(?P<region>[a-z0-9-]+)\.wasabisys\.com$"
+        ],
+    }
+    mock_config = {"FILES_REST_STORAGE_SERVICE_PATTERN": patterns}
+    app.config.update(mock_config)
+
+    # Test case (Pos): Match aws_s3 pattern
+    result = parse_storage_host("s3.amazonaws.com")
+    assert result["service"] == "aws_s3"
+    assert result["params"] == {}
+    # Test case (Pos): Match aws_s3 pattern with bucket name
+    result2 = parse_storage_host("mybucket.s3.amazonaws.com")
+    assert result2["service"] == "aws_s3"
+    assert result2["params"]["bucket"] == "mybucket"
+    # Test case (Pos): Match wasabi pattern
+    result3 = parse_storage_host("s3.wasabisys.com")
+    assert result3["service"] == "wasabi"
+    assert result3["params"] == {}
+    # Test case (Pos): Match wasabi pattern with region
+    result4 = parse_storage_host("s3.us-west-1.wasabisys.com")
+    assert result4["service"] == "wasabi"
+    assert result4["params"]["region"] == "us-west-1"
+
+
+# def parse_storage_host(host):
+# .tox/c1/bin/pytest --cov=invenio_files_rest tests/test_utils.py::test_parse_storage_host_no_match -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-files-rest/.tox/c1/tmp
+def test_parse_storage_host_no_match(app):
+    # Test case (Pos): No match any pattern
+    patterns = {
+        "aws_s3": [r"^s3\.amazonaws\.com$"],
+    }
+    mock_config = {"FILES_REST_STORAGE_SERVICE_PATTERN": patterns}
+    app.config.update(mock_config)
+    result = parse_storage_host("unknown.host.com")
+    assert result["service"] is None
+    assert result["params"] == {}
+
+
+# def parse_storage_host(host):
+# .tox/c1/bin/pytest --cov=invenio_files_rest tests/test_utils.py::test_parse_storage_host_empty_patterns -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-files-rest/.tox/c1/tmp
+def test_parse_storage_host_empty_patterns(app):
+    # Test case (Pos): Empty patterns
+    mock_config = {"FILES_REST_STORAGE_SERVICE_PATTERN": {}}
+    app.config.update(mock_config)
+    result = parse_storage_host("s3.amazonaws.com")
+    assert result["service"] is None
+    assert result["params"] == {}