fix(security): block RFC6598 shared address space in S3 endpoint (Alibaba/Tencent metadata SSRF)

labelu/internal/application/command/datasource.py

@@ -8,11 +8,22 @@
 from labelu.internal.common.config import settings
 
 
+# Networks that are never a valid S3 endpoint and must be blocked regardless of
+# ALLOW_PRIVATE_S3_ENDPOINT. 100.64.0.0/10 is RFC6598 shared address space
+# (Carrier-Grade NAT) and is NOT flagged private/link-local/reserved by the
+# stdlib, yet it hosts the Alibaba/Tencent cloud metadata service
+# (100.100.100.200) — so it must be rejected explicitly.
+_ALWAYS_BLOCK_NETWORKS = (
+    ipaddress.ip_network("100.64.0.0/10"),
+)
+
+
 def validate_s3_endpoint(endpoint: Union[str, None]) -> Union[str, None]:
     """Reject S3 endpoints that could be abused for SSRF.
 
     Only http(s) URLs are allowed. Link-local (cloud metadata
-    169.254.0.0/16, fe80::/10), multicast, reserved and unspecified
+    169.254.0.0/16, fe80::/10), multicast, reserved, unspecified and RFC6598
+    shared-address-space (100.64.0.0/10, incl. Alibaba/Tencent metadata)
     addresses are always rejected — they are never valid S3 endpoints.
 
     Private (RFC1918 / ULA) and loopback addresses are allowed when
@@ -43,11 +54,16 @@ def validate_s3_endpoint(endpoint: Union[str, None]) -> Union[str, None]:
         # Python, so link-local is checked here and must win over the private
         # carve-out below. Loopback is exempt from the reserved check because
         # ::1 is also flagged is_reserved.
+        in_blocked_net = any(
+            addr.version == net.version and addr in net
+            for net in _ALWAYS_BLOCK_NETWORKS
+        )
         if (
             addr.is_link_local
             or addr.is_multicast
             or addr.is_unspecified
             or (addr.is_reserved and not addr.is_loopback)
+            or in_blocked_net
         ):
             raise ValueError("endpoint host resolves to a disallowed address")
         # Internal networks (RFC1918 / ULA / loopback): allowed only when

labelu/internal/application/service/auto_label.py

@@ -9,6 +9,7 @@
 from loguru import logger
 from sqlalchemy.orm import Session
 
+from labelu.internal.application.service.access import assert_task_access
 from labelu.internal.adapter.persistence import crud_pre_annotation, crud_sample, crud_task
 from labelu.internal.adapter.persistence import crud_auto_label_job
 from labelu.internal.application.command.auto_label import AutoLabelCommand, BatchAutoLabelCommand
@@ -361,6 +362,8 @@ async def create_batch_job(
             status_code=status.HTTP_404_NOT_FOUND,
         )
 
+    assert_task_access(task, current_user)
+
     if task.media_type != MediaType.IMAGE.value:
         raise LabelUException(
             code=ErrorCode.CODE_56001_AUTO_LABEL_UNSUPPORTED_MEDIA,

labelu/tests/internal/adapter/routers/test_sample.py

@@ -353,6 +353,91 @@ def test_sample_list_by_sort_error(
         # check
         assert r.status_code == 422
 
+    def test_batch_auto_label_forbidden_for_non_owner(
+        self, client: TestClient, testuser_token_headers: dict, db: Session, monkeypatch
+    ) -> None:
+        # enable the AI feature so the request reaches the access check
+        monkeypatch.setattr(settings, "AI_AUTO_LABEL_ENABLED", True)
+        monkeypatch.setattr(settings, "AI_MODEL_ENDPOINT", "http://model.local")
+        # task owned by another user
+        with begin_transaction(db):
+            task = crud_task.create(
+                db=db,
+                task=Task(
+                    name="other", description="d", tips="t", media_type="IMAGE",
+                    created_by=999, updated_by=999,
+                ),
+            )
+        r = client.post(
+            f"{settings.API_V1_STR}/tasks/{task.id}/auto_label_job",
+            headers=testuser_token_headers,
+            json={"filter_by_labels": True},
+        )
+        assert r.status_code == 403
+
+    def test_collaborator_can_annotate_shared_task(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # owner = the default test user
+        owner = crud_user.get_user_by_username(db=db, username="test@example.com")
+
+        # a second user, added as collaborator, with a real login token
+        collab_username = "collab_annotator@example.com"
+        client.post(
+            f"{settings.API_V1_STR}/users/signup",
+            json={"username": collab_username, "password": "collab@123"},
+        )
+        login = client.post(
+            f"{settings.API_V1_STR}/users/login",
+            json={"username": collab_username, "password": "collab@123"},
+        )
+        collab_headers = {"Authorization": login.json()["data"]["token"]}
+        collab_user = crud_user.get_user_by_username(db=db, username=collab_username)
+
+        with begin_transaction(db):
+            task = crud_task.create(
+                db=db,
+                task=Task(
+                    name="shared", description="d", tips="t",
+                    created_by=owner.id, updated_by=owner.id,
+                ),
+            )
+        with begin_transaction(db):
+            samples = crud_sample.batch(
+                db=db,
+                samples=[TaskSample(
+                    task_id=task.id, file_id=1,
+                    created_by=owner.id, updated_by=owner.id, data="{}",
+                )],
+            )
+        sid = samples[0].id
+
+        # owner adds the collaborator
+        r = client.post(
+            f"{settings.API_V1_STR}/tasks/{task.id}/collaborators",
+            headers=testuser_token_headers,
+            json={"user_id": collab_user.id},
+        )
+        assert r.status_code == 201
+
+        # the collaborator can list, read and annotate samples in the shared task
+        r = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/samples",
+            headers=collab_headers, params={"page": 0, "size": 10},
+        )
+        assert r.status_code == 200
+        r = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/samples/{sid}",
+            headers=collab_headers,
+        )
+        assert r.status_code == 200
+        r = client.patch(
+            f"{settings.API_V1_STR}/tasks/{task.id}/samples/{sid}",
+            headers=collab_headers,
+            json={"data": {"k": "v"}, "annotated_count": 1},
+        )
+        assert r.status_code == 200
+
     def test_sample_access_forbidden_for_non_member(
         self, client: TestClient, testuser_token_headers: dict, db: Session
     ) -> None:

labelu/tests/internal/application/command/test_datasource.py

@@ -24,8 +24,10 @@ class TestDataSourceEndpointSSRF:
     @pytest.mark.parametrize(
         "endpoint",
         [
-            "http://169.254.169.254",          # cloud metadata (IMDS) - always blocked
+            "http://169.254.169.254",          # AWS/GCP/Azure metadata (IMDS) - link-local
             "http://169.254.169.254/latest/",  # IMDS with path
+            "http://100.100.100.200/latest/",  # Alibaba/Tencent cloud metadata (100.64.0.0/10)
+            "http://100.64.0.1:9000",          # RFC6598 shared address space
             "http://0.0.0.0",                  # unspecified - always blocked
             "file:///etc/passwd",              # non-http scheme
             "ftp://example.com",               # non-http scheme