Fix validation of target_exe blob name (#1371)

src/api-service/__app__/onefuzzlib/tasks/config.py

@@ -4,9 +4,8 @@
 # Licensed under the MIT License.
 
 import logging
-import ntpath
 import os
-import posixpath
+import pathlib
 from typing import Dict, List, Optional
 from uuid import UUID
 
@@ -105,14 +104,12 @@ def check_target_exe(config: TaskConfig, definition: TaskDefinition) -> None:
 
         return
 
-    # Azure Blob Store uses virtualized directory structures.  As such, we need
-    # the paths to already be canonicalized.  As an example, accessing the blob
-    # store path "./foo" generates an exception, but "foo" and "foo/bar" do
-    # not.
-    if (
-        posixpath.relpath(config.task.target_exe) != config.task.target_exe
-        or ntpath.relpath(config.task.target_exe) != config.task.target_exe
-    ):
+    # User-submitted paths must be relative to the setup directory that contains them.
+    # They also must be normalized, and exclude special filesystem path elements.
+    #
+    # For example, accessing the blob store path "./foo" generates an exception, but
+    # "foo" and "foo/bar" do not.
+    if not is_valid_blob_name(config.task.target_exe):
         raise TaskConfigError("target_exe must be a canonicalized relative path")
 
     container = [x for x in config.containers if x.type == ContainerType.setup][0]
@@ -124,6 +121,53 @@ def check_target_exe(config: TaskConfig, definition: TaskDefinition) -> None:
         LOGGER.warning(err)
 
 
+# Azure Blob Storage uses a flat scheme, and has no true directory hierarchy. Forward
+# slashes are used to delimit a _virtual_ directory structure.
+def is_valid_blob_name(blob_name: str) -> bool:
+    # https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#blob-names
+    MIN_LENGTH = 1
+    MAX_LENGTH = 1024  # Inclusive
+    MAX_PATH_SEGMENTS = 254
+
+    length = len(blob_name)
+
+    # No leading/trailing whitespace.
+    if blob_name != blob_name.strip():
+        return False
+
+    if length < MIN_LENGTH:
+        return False
+
+    if length > MAX_LENGTH:
+        return False
+
+    path = pathlib.PurePosixPath(blob_name)
+
+    if len(path.parts) > MAX_PATH_SEGMENTS:
+        return False
+
+    # No path segment should end with a dot (`.`).
+    for part in path.parts:
+        if part.endswith("."):
+            return False
+
+    # Reject absolute paths to avoid confusion.
+    if path.is_absolute():
+        return False
+
+    # Reject paths with special relative filesystem entries.
+    if "." in path.parts:
+        return False
+
+    if ".." in path.parts:
+        return False
+
+    # Will not have a leading `.`, even if `blob_name` does.
+    normalized = path.as_posix()
+
+    return blob_name == normalized
+
+
 def target_uses_input(config: TaskConfig) -> bool:
     if config.task.target_options is not None:
         for option in config.task.target_options:

src/api-service/tests/test_task_config.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+#
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from typing import Tuple
+
+import pytest
+
+from __app__.onefuzzlib.tasks.config import is_valid_blob_name
+
+BlobNameTestCase = Tuple[str, bool]
+
+
+BLOB_NAME_TEST_CASES = [
+    # Valid
+    ("fuzz.exe", True),
+    ("bin/fuzz.exe", True),
+    ("/".join("a" * 254), True),
+    ("a" * 1024, True),
+    # Invalid (absolute)
+    ("/fuzz.exe", False),
+    ("/bin/fuzz.exe", False),
+    # Invalid (special dirs)
+    ("./fuzz.exe", False),
+    ("././fuzz.exe", False),
+    ("./bin/fuzz.exe", False),
+    ("./bin/./fuzz.exe", False),
+    ("../fuzz.exe", False),
+    ("../bin/fuzz.exe", False),
+    (".././fuzz.exe", False),
+    ("../bin/./fuzz.exe", False),
+    # Out of Azure size bounds
+    ("", False),
+    ("  ", False),
+    ("/".join("a" * 255), False),
+    ("a" * 1025, False),
+    # Paths with invalid segments.
+    ("a.", False),
+    ("a..", False),
+    ("a./b", False),
+    ("a/b./c", False),
+    ("a./", False),
+    ("a../", False),
+    ("a./b/", False),
+    ("a/b./c/", False),
+    ("a//", False),
+]
+
+
+@pytest.mark.parametrize("blob_name_test_case", BLOB_NAME_TEST_CASES)
+def test_is_valid_blob_name(blob_name_test_case: BlobNameTestCase) -> None:
+    blob_name, expected = blob_name_test_case
+
+    is_valid = is_valid_blob_name(blob_name)
+
+    assert is_valid == expected