Prevent ORAS execution from the current directory

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: takuro-sato

src/confcom/azext_confcom/command/fragment_attach.py

@@ -19,7 +19,6 @@ def oras_attach(
     platform: Optional[str] = None,
 ) -> None:
     cmd = [
-        "oras",
         "attach",
         "--artifact-type", "application/x-ms-ccepolicy-frag",
     ]
@@ -51,7 +50,7 @@ def oras_attach(
     ])
 
     subprocess.run(
-        cmd,
+        [oras_proxy.resolve_oras_path(), *cmd],
         check=True,
         timeout=120,
     )

src/confcom/azext_confcom/command/fragment_push.py

@@ -9,6 +9,7 @@
 import tempfile
 from typing import Optional
 
+from azext_confcom import oras_proxy
 from azext_confcom.errors import eprint
 
 
@@ -18,7 +19,7 @@ def oras_push(
 ) -> None:
     subprocess.run(
         [
-            "oras",
+            oras_proxy.resolve_oras_path(),
             "push",
             "--artifact-type", "application/x-ms-ccepolicy-frag",
             manifest_tag,

src/confcom/azext_confcom/oras_proxy.py

@@ -6,6 +6,7 @@
 import json
 import os
 import re
+import shutil
 import subprocess
 from tempfile import mkdtemp
 from typing import List, Optional
@@ -24,6 +25,43 @@
 logger = get_logger(__name__)
 
 
+def resolve_oras_path() -> str:
+    """Resolve ORAS outside CWD to prevent executable search path hijacking."""
+    current_directory = os.path.normcase(os.path.realpath(os.getcwd()))
+    executable_extensions = [""]
+    if os.name == "nt":
+        executable_extensions = [
+            extension if extension.startswith(".") else f".{extension}"
+            for extension in os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").split(os.pathsep)
+            if extension
+        ]
+
+    for path_entry in os.environ.get("PATH", "").split(os.pathsep):
+        # Normalize quoted entries before rejecting empty or relative paths.
+        path_entry = path_entry.strip().strip('"')
+        if not path_entry or not os.path.isabs(path_entry):
+            continue
+        if os.name == "nt" and not os.path.splitdrive(path_entry)[0]:
+            continue
+
+        resolved_directory = os.path.realpath(path_entry)
+        # Exclude CWD as recommended for preventing binary planting attacks:
+        # https://learn.microsoft.com/en-us/windows/win32/dlls/dynamic-link-library-security
+        if os.path.normcase(resolved_directory) == current_directory:
+            continue
+
+        for extension in executable_extensions:
+            oras_path = shutil.which(os.path.join(resolved_directory, f"oras{extension}"))
+            if oras_path:
+                resolved_oras_path = os.path.realpath(os.path.abspath(oras_path))
+                if os.path.normcase(os.path.dirname(resolved_oras_path)) != current_directory:
+                    return resolved_oras_path
+
+    eprint("ORAS CLI not found in absolute PATH entries. Please install ORAS CLI: "
+           "https://oras.land/docs/installation")
+    raise RuntimeError("ORAS path resolution failed")
+
+
 def prepend_docker_registry(image_name: str) -> str:
     """
     Normalize a Docker image reference by adding `docker.io/library` if necessary.
@@ -54,7 +92,12 @@ def prepend_docker_registry(image_name: str) -> str:
 
 
 def call_oras_cli(args, check=False):
-    return subprocess.run(args, check=check, capture_output=True, timeout=120)
+    return subprocess.run(
+        [resolve_oras_path(), *args],
+        check=check,
+        capture_output=True,
+        timeout=120,
+    )
 
 
 def manifest_fetch(image_tag: str) -> Optional[dict]:
@@ -64,7 +107,7 @@ def manifest_fetch(image_tag: str) -> Optional[dict]:
     """
     try:
         result = subprocess.run(
-            ["oras", "manifest", "fetch", "--format", "json", image_tag],
+            [resolve_oras_path(), "manifest", "fetch", "--format", "json", image_tag],
             capture_output=True,
             text=True,
             timeout=30,
@@ -86,7 +129,7 @@ def manifest_fetch_config(image_tag: str) -> Optional[dict]:
     """
     try:
         result = subprocess.run(
-            ["oras", "manifest", "fetch-config", image_tag],
+            [resolve_oras_path(), "manifest", "fetch-config", image_tag],
             capture_output=True,
             text=True,
             timeout=30,
@@ -179,7 +222,7 @@ def discover(
     # normalize the name in case the docker registry is implied
     image = prepend_docker_registry(image)
 
-    arg_list = ["oras", "discover", image, "-o", "json", "--artifact-type", ARTIFACT_TYPE]
+    arg_list = ["discover", image, "-o", "json", "--artifact-type", ARTIFACT_TYPE]
     if platform:
         arg_list.extend(["--platform", platform])
 
@@ -242,7 +285,7 @@ def pull(
     logger.info("Pulling fragment: %s", full_path)
 
     temp_folder = mkdtemp()
-    arg_list = ["oras", "pull", full_path, "-o", temp_folder]
+    arg_list = ["pull", full_path, "-o", temp_folder]
 
     item = call_oras_cli(arg_list, check=False)
 
@@ -340,13 +383,10 @@ def pull_all_standalone_fragments(fragment_imports):
 
 
 def check_oras_cli():
-    text = "ORAS CLI not installed. Please install ORAS CLI: https://oras.land/docs/installation"
-    try:
-        item = call_oras_cli(["oras", "version"], check=False)
-        if item.returncode != 0:
-            eprint(text)
-    except FileNotFoundError:
-        eprint(text)
+    item = call_oras_cli(["version"], check=False)
+    if item.returncode != 0:
+        eprint("ORAS CLI could not be executed. Please install ORAS CLI: "
+               "https://oras.land/docs/installation")
 
 
 # used for image-attached fragments
@@ -355,7 +395,6 @@ def attach_fragment_to_image(image_name: str, filename: str, platform: Optional[
         image_name += ":latest"
     # attach the fragment to the image
     arg_list = [
-        "oras",
         "attach",
         "--artifact-type",
         ARTIFACT_TYPE,
@@ -406,7 +445,6 @@ def generate_imports_from_image_name(image_name: str, minimum_svn: str) -> List[
 def push_fragment_to_registry(feed_name: str, filename: str) -> None:
     # push the fragment to the registry
     arg_list = [
-        "oras",
         "push",
         feed_name,
         "--artifact-type",

src/confcom/azext_confcom/tests/latest/test_confcom_acifragmentgen.py

@@ -205,6 +205,83 @@ def test_acifragmentgen_fragment_push(docker_image, cert_chain, capsysbinary):
         assert f.read() == signed_fragment
 
 
+# Unit tests for ORAS path resolution
+
+
+def test_resolve_oras_path_uses_absolute_path_entries(monkeypatch, tmp_path):
+    """A planted CWD executable is ignored in favor of another absolute PATH entry."""
+    from azext_confcom.oras_proxy import resolve_oras_path
+
+    current_directory = tmp_path / "work"
+    trusted_directory = tmp_path / "tools"
+    current_directory.mkdir()
+    trusted_directory.mkdir()
+    oras_name = "oras.exe" if os.name == "nt" else "oras"
+    planted_oras = current_directory / oras_name
+    trusted_oras = trusted_directory / oras_name
+    planted_oras.touch()
+    trusted_oras.touch()
+    planted_oras.chmod(0o755)
+    trusted_oras.chmod(0o755)
+
+    monkeypatch.chdir(current_directory)
+    if os.name == "nt":
+        monkeypatch.setenv("PATHEXT", ".EXE")
+    monkeypatch.setenv(
+        "PATH",
+        os.pathsep.join(["", ".", str(current_directory), "relative-tools", str(trusted_directory)]),
+    )
+
+    assert resolve_oras_path() == os.path.realpath(trusted_oras)
+
+
+def test_resolve_oras_path_errors_when_no_safe_candidate(monkeypatch, tmp_path):
+    """Resolution fails instead of executing ORAS from the current directory."""
+    from azext_confcom.oras_proxy import resolve_oras_path
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("PATH", os.pathsep.join(["", ".", str(tmp_path), "relative-tools"]))
+
+    with patch("azext_confcom.oras_proxy.shutil.which") as mock_which:
+        with pytest.raises(SystemExit):
+            resolve_oras_path()
+
+    mock_which.assert_not_called()
+
+
+def test_call_oras_cli_executes_resolved_absolute_path():
+    """The shared ORAS runner never passes a bare executable name to subprocess."""
+    from azext_confcom.oras_proxy import call_oras_cli
+
+    with patch(
+        "azext_confcom.oras_proxy.resolve_oras_path",
+        return_value=os.path.abspath(os.path.join("trusted-tools", "oras.exe")),
+    ) as mock_resolve, patch("azext_confcom.oras_proxy.subprocess.run") as mock_run:
+        call_oras_cli(["discover", "example.azurecr.io/image:tag"])
+
+    mock_resolve.assert_called_once_with()
+    command = mock_run.call_args.args[0]
+    assert os.path.isabs(command[0])
+    assert command[1:] == ["discover", "example.azurecr.io/image:tag"]
+
+
+@patch("azext_confcom.command.fragment_push.subprocess.run")
+def test_oras_push_executes_resolved_absolute_path(mock_run):
+    """The fragment push command executes the resolved ORAS path."""
+    from azext_confcom.command.fragment_push import oras_push
+
+    resolved_path = os.path.abspath(os.path.join("trusted-tools", "oras.exe"))
+    with patch(
+        "azext_confcom.command.fragment_push.oras_proxy.resolve_oras_path",
+        return_value=resolved_path,
+    ):
+        oras_push("fragment.rego.cose", "example.azurecr.io/fragments:test")
+
+    command = mock_run.call_args.args[0]
+    assert command[0] == resolved_path
+    assert command[1] == "push"
+
+
 # ── Unit tests for prepend_docker_registry ─────────────────────────────────
 
 
@@ -361,14 +438,19 @@ def test_oras_attach_explicit_platform(mock_run):
 
     mock_run.return_value = MagicMock(returncode=0)
 
-    oras_attach(
-        signed_fragment_path="/tmp/fragment.cose",
-        manifest_tag="myregistry.io/myimage:latest",
-        platform="linux/amd64",
-    )
+    with patch(
+        "azext_confcom.command.fragment_attach.oras_proxy.resolve_oras_path",
+        return_value=os.path.abspath(os.path.join("trusted-tools", "oras.exe")),
+    ):
+        oras_attach(
+            signed_fragment_path="/tmp/fragment.cose",
+            manifest_tag="myregistry.io/myimage:latest",
+            platform="linux/amd64",
+        )
 
     mock_run.assert_called_once()
     cmd = mock_run.call_args[0][0]
+    assert os.path.isabs(cmd[0])
     assert "--platform" in cmd
     assert "linux/amd64" in cmd
     assert "application/cose-x509+rego" in " ".join(cmd)
@@ -385,13 +467,18 @@ def test_oras_attach_auto_detected_platform(mock_platforms, mock_run):
 
     mock_run.return_value = MagicMock(returncode=0)
 
-    oras_attach(
-        signed_fragment_path="/tmp/fragment.cose",
-        manifest_tag="myregistry.io/myimage:latest",
-    )
+    with patch(
+        "azext_confcom.command.fragment_attach.oras_proxy.resolve_oras_path",
+        return_value=os.path.abspath(os.path.join("trusted-tools", "oras.exe")),
+    ):
+        oras_attach(
+            signed_fragment_path="/tmp/fragment.cose",
+            manifest_tag="myregistry.io/myimage:latest",
+        )
 
     mock_run.assert_called_once()
     cmd = mock_run.call_args[0][0]
+    assert os.path.isabs(cmd[0])
     assert "--platform" in cmd
     assert "linux/amd64" in cmd
     assert "application/cose-x509+rego" in " ".join(cmd)