fix(security): fix SQL injection in the sample-listing sort parameter

README_zh-CN.md

@@ -85,6 +85,7 @@ conda activate labelu
 
 4. 安装 LabelU：
 
+安装基础版本（使用SQLite）
 ```bash
 pip install labelu
 ```
@@ -100,7 +101,7 @@ pip install labelu[mysql]
 # pip install labelu mysqlclient
 ```
 
-5. 运行：
+1. 运行：
 
 ```bash
 labelu

labelu/internal/adapter/persistence/crud_sample.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, List, Union
 
 
-from sqlalchemy import case, func, text
+from sqlalchemy import case, func
 from sqlalchemy.orm import Session
 from fastapi.encoders import jsonable_encoder
 
@@ -39,17 +39,32 @@ def list_by(
     whens = {state.value: index for index, state in enumerate(SampleState)}
     sort_logic = case(whens, value=TaskSample.state).label(TaskSample.state.key)
 
+    # Allowlist of sortable columns mapped to real ORM columns. Only these
+    # identifiers may reach ORDER BY; anything else is rejected so request
+    # input can never be interpolated into raw SQL.
+    sortable_columns = {
+        "inner_id": TaskSample.inner_id,
+        "updated_at": TaskSample.updated_at,
+        "annotated_count": TaskSample.annotated_count,
+    }
+
     if sorting:
         sort_strings = sorting.split(",")
         for item in sort_strings:
             sort_key = item.split(":")
-            if sort_key[0] == TaskSample.state.key:
-                if sort_key[1] == "asc":
-                    query = query.order_by(sort_logic.asc())
-                else:
-                    query = query.order_by(sort_logic.desc())
+            column = sort_key[0]
+            direction = sort_key[1] if len(sort_key) > 1 else "asc"
+            if direction not in ("asc", "desc"):
+                continue
+            if column == TaskSample.state.key:
+                order_target = sort_logic
+            elif column in sortable_columns:
+                order_target = sortable_columns[column]
             else:
-                query = query.order_by(text(f"{sort_key[0]} {sort_key[1]}"))
+                continue
+            query = query.order_by(
+                order_target.asc() if direction == "asc" else order_target.desc()
+            )
 
     # default order by id, before need select last items
     if before:

labelu/internal/adapter/routers/datasource.py

@@ -72,7 +72,7 @@ async def get(
     db: Session = Depends(db_module.get_db),
     current_user: User = Depends(get_current_user),
 ):
-    data = await service.get(db=db, ds_id=ds_id)
+    data = await service.get(db=db, ds_id=ds_id, current_user=current_user)
     return OkResp[DataSourceResponse](data=data)
 
 
@@ -123,7 +123,7 @@ async def list_objects(
     current_user: User = Depends(get_current_user),
 ):
     data = await service.list_objects(
-        db=db, ds_id=ds_id, prefix=prefix, extension=extension,
+        db=db, ds_id=ds_id, current_user=current_user, prefix=prefix, extension=extension,
         page_token=page_token, size=size,
     )
     return OkResp[S3ObjectListResponse](data=data)

labelu/internal/adapter/routers/sample.py

@@ -30,6 +30,8 @@
 from labelu.internal.application.response.auto_label import AutoLabelResponse, AutoLabelJobResponse
 from labelu.internal.application.response.export import ExportJobResponse
 from labelu.internal.adapter.persistence import crud_export_job
+from labelu.internal.adapter.persistence import crud_task
+from labelu.internal.application.service.access import assert_task_access
 from labelu.internal.common.storage import get_storage_backend
 
 
@@ -84,7 +86,9 @@ async def list_by(
     page: Union[int, None] = Query(default=None, ge=0),
     size: Union[int, None] = 100,
     sort: Union[str, None] = Query(
-        default=None, pattern="(annotated_count|state|inner_id|updated_at):(desc|asc)"
+        default=None,
+        pattern=r"^(annotated_count|state|inner_id|updated_at):(desc|asc)"
+        r"(,(annotated_count|state|inner_id|updated_at):(desc|asc))*$",
     ),
     authorization: HTTPAuthorizationCredentials = Security(security),
     db: Session = Depends(db_module.get_db),
@@ -109,6 +113,7 @@ async def list_by(
         page=page,
         size=size,
         sorting=sort,
+        current_user=current_user,
     )
 
     # response
@@ -134,7 +139,7 @@ async def get(
 
     # business logic
     data = await service.get(
-        db=db, task_id=task_id, sample_id=sample_id
+        db=db, task_id=task_id, sample_id=sample_id, current_user=current_user
     )
 
     # response
@@ -329,6 +334,10 @@ async def get_export_status(
             status_code=status.HTTP_404_NOT_FOUND,
         )
 
+    task = crud_task.get(db=db, task_id=job.task_id)
+    if task is not None:
+        assert_task_access(task, current_user)
+
     skipped_count, warning_message = _export_progress(
         sample_count=job.sample_count,
         processed_count=job.processed_count,
@@ -373,6 +382,10 @@ async def download_export(
             status_code=status.HTTP_404_NOT_FOUND,
         )
 
+    task = crud_task.get(db=db, task_id=job.task_id)
+    if task is not None:
+        assert_task_access(task, current_user)
+
     storage = get_storage_backend()
     if storage.is_remote:
         download_url = storage.get_read_url(job.file_path)

labelu/internal/adapter/routers/task.py

@@ -220,7 +220,7 @@ async def update(
     """
 
     # business logic
-    data = await service.update(db=db, task_id=task_id, cmd=cmd)
+    data = await service.update(db=db, task_id=task_id, cmd=cmd, current_user=current_user)
 
     # response
     return OkResp[TaskResponse](data=data)

labelu/internal/application/command/datasource.py

@@ -1,12 +1,68 @@
+import ipaddress
+import socket
 from typing import Union
+from urllib.parse import urlparse
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from labelu.internal.common.config import settings
+
+
+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
+    addresses are always rejected — they are never valid S3 endpoints.
+
+    Private (RFC1918 / ULA) and loopback addresses are allowed when
+    ``settings.ALLOW_PRIVATE_S3_ENDPOINT`` is true (the default, for on-prem
+    deployments backed by an internal MinIO/S3-compatible store) and
+    rejected otherwise (strict mode).
+    """
+    if endpoint is None or endpoint == "":
+        return endpoint
+
+    parsed = urlparse(endpoint)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError("endpoint must be an http(s) URL")
+    host = parsed.hostname
+    if not host:
+        raise ValueError("endpoint must include a host")
+
+    try:
+        infos = socket.getaddrinfo(host, None)
+    except socket.gaierror:
+        raise ValueError("endpoint host cannot be resolved")
+
+    allow_private = settings.ALLOW_PRIVATE_S3_ENDPOINT
+    for info in infos:
+        addr = ipaddress.ip_address(info[4][0])
+        # Never a legitimate S3 endpoint; always rejected regardless of config.
+        # Note: the metadata IP (169.254.x) is also flagged is_private on newer
+        # 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.
+        if (
+            addr.is_link_local
+            or addr.is_multicast
+            or addr.is_unspecified
+            or (addr.is_reserved and not addr.is_loopback)
+        ):
+            raise ValueError("endpoint host resolves to a disallowed address")
+        # Internal networks (RFC1918 / ULA / loopback): allowed only when
+        # explicitly permitted via ALLOW_PRIVATE_S3_ENDPOINT.
+        if (addr.is_private or addr.is_loopback) and not allow_private:
+            raise ValueError("endpoint host resolves to a private address")
+    return endpoint
 
 
 class CreateDataSourceCommand(BaseModel):
     name: str = Field(max_length=128, description="Display name")
     type: str = Field(default="S3", max_length=32)
     endpoint: Union[str, None] = Field(default=None, max_length=512)
+
+    _validate_endpoint = field_validator("endpoint")(validate_s3_endpoint)
     region: Union[str, None] = Field(default=None, max_length=64)
     bucket: str = Field(max_length=256)
     prefix: str = Field(default="", max_length=512)
@@ -20,6 +76,8 @@ class CreateDataSourceCommand(BaseModel):
 class UpdateDataSourceCommand(BaseModel):
     name: Union[str, None] = Field(default=None, max_length=128)
     endpoint: Union[str, None] = Field(default=None, max_length=512)
+
+    _validate_endpoint = field_validator("endpoint")(validate_s3_endpoint)
     region: Union[str, None] = Field(default=None, max_length=64)
     bucket: Union[str, None] = Field(default=None, max_length=256)
     prefix: Union[str, None] = Field(default=None, max_length=512)

labelu/internal/application/service/access.py

@@ -0,0 +1,31 @@
+from fastapi import status
+
+from labelu.internal.common.error_code import ErrorCode, LabelUException
+
+
+def assert_task_access(task, current_user) -> None:
+    """Ensure current_user may access the given task's data.
+
+    Access is granted to the task owner or any of its collaborators. Any other
+    authenticated user is rejected with 403, closing IDOR/BOLA on task-scoped
+    resources (samples, pre-annotations, exports, ...).
+    """
+    collaborator_ids = {c.id for c in task.collaborators}
+    if task.created_by != current_user.id and current_user.id not in collaborator_ids:
+        raise LabelUException(
+            code=ErrorCode.CODE_30001_NO_PERMISSION,
+            status_code=status.HTTP_403_FORBIDDEN,
+        )
+
+
+def assert_owner(resource, current_user) -> None:
+    """Ensure current_user owns the resource (created_by).
+
+    Used for user-private resources without a collaborator concept
+    (e.g. data sources).
+    """
+    if resource.created_by != current_user.id:
+        raise LabelUException(
+            code=ErrorCode.CODE_30001_NO_PERMISSION,
+            status_code=status.HTTP_403_FORBIDDEN,
+        )

labelu/internal/application/service/attachment.py

@@ -82,11 +82,21 @@ async def create(
 
         # file relative path
     path_filename = cmd.file.filename.split("/")
-    #  filename = str(uuid.uuid4())[0:8] + "-" + path_filename[-1] NOTE: If you want keep filename safe, you can use uuid as filename 
+    #  filename = str(uuid.uuid4())[0:8] + "-" + path_filename[-1] NOTE: If you want keep filename safe, you can use uuid as filename
     filename = path_filename[-1]
     sanitized = re.sub(r'%', '_pct_', filename)
     sanitized = re.sub(r'[\\/*?:"<>|#]', '_', sanitized)
-    path = "/".join(path_filename[:-1])
+    # Sanitize the directory portion of the filename too: drop traversal
+    # components ("..", ".", empty) and apply the same character sanitization,
+    # so a crafted filename can never escape the task's upload directory.
+    safe_parts = []
+    for part in path_filename[:-1]:
+        part = re.sub(r'%', '_pct_', part)
+        part = re.sub(r'[\\/*?:"<>|#]', '_', part)
+        if part in ("", ".", ".."):
+            continue
+        safe_parts.append(part)
+    path = "/".join(safe_parts)
     attachment_relative_base_dir = Path(settings.UPLOAD_DIR).joinpath(
         str(task_id), path
     )

labelu/internal/application/service/datasource.py

@@ -9,6 +9,7 @@
 from sqlalchemy.orm import Session
 
 from labelu.internal.adapter.persistence import crud_datasource
+from labelu.internal.application.service.access import assert_owner
 from labelu.internal.application.command.datasource import (
     CreateDataSourceCommand,
     UpdateDataSourceCommand,
@@ -110,13 +111,14 @@ async def list_by(
     return [_to_response(ds) for ds in items], total
 
 
-async def get(db: Session, ds_id: int) -> DataSourceResponse:
+async def get(db: Session, ds_id: int, current_user: User) -> DataSourceResponse:
     ds = crud_datasource.get(db=db, ds_id=ds_id)
     if not ds:
         raise LabelUException(
             code=ErrorCode.CODE_61000_NO_DATA,
             status_code=status.HTTP_404_NOT_FOUND,
         )
+    assert_owner(ds, current_user)
     return _to_response(ds)
 
 
@@ -129,6 +131,7 @@ async def update(
             code=ErrorCode.CODE_61000_NO_DATA,
             status_code=status.HTTP_404_NOT_FOUND,
         )
+    assert_owner(ds, current_user)
     obj_in = cmd.model_dump(exclude_unset=True)
     if "access_key_id" in obj_in and obj_in["access_key_id"] is not None:
         obj_in["access_key_id"] = encrypt_value(obj_in["access_key_id"])
@@ -147,6 +150,7 @@ async def delete(db: Session, ds_id: int, current_user: User) -> None:
             code=ErrorCode.CODE_61000_NO_DATA,
             status_code=status.HTTP_404_NOT_FOUND,
         )
+    assert_owner(ds, current_user)
     with begin_transaction(db):
         crud_datasource.soft_delete(db=db, db_obj=ds)
 
@@ -156,6 +160,7 @@ async def delete(db: Session, ds_id: int, current_user: User) -> None:
 async def list_objects(
     db: Session,
     ds_id: int,
+    current_user: User,
     prefix: Optional[str] = None,
     extension: Optional[str] = None,
     page_token: Optional[str] = None,
@@ -167,6 +172,7 @@ async def list_objects(
             code=ErrorCode.CODE_61000_NO_DATA,
             status_code=status.HTTP_404_NOT_FOUND,
         )
+    assert_owner(ds, current_user)
 
     client = _build_s3_client(ds)
     full_prefix = prefix if prefix is not None else (ds.prefix or "")