fix(amass): replace hardcoded /tmp/amass with a private per-task scratch dir

backend/secuscan/executor.py

@@ -5,8 +5,10 @@
 import asyncio
 from asyncio import subprocess
 import os
+import shutil
 import signal
 import base64
+import tempfile
 import uuid
 import json
 import time
@@ -18,6 +20,39 @@
 
 _CANCEL_GRACE_SECONDS = 5
 
+# Reserved command-template token. A plugin that needs a writable working
+# directory puts this literal in its ``command_template`` instead of a
+# hardcoded path (e.g. amass's ``-dir``). At execution time it is replaced with
+# a fresh, private (0700) per-task directory created via ``tempfile.mkdtemp``
+# and removed once the scan finishes. This closes the predictable-world-writable
+# ``/tmp/<tool>`` symlink-hijack / cross-user-collision class of bug: the path is
+# unguessable per run and never shared or left behind. The token deliberately
+# contains no ``{...}`` so it passes plugin validation and template
+# interpolation through untouched as an opaque literal.
+SCRATCH_DIR_PLACEHOLDER = "%SECUSCAN_SCRATCH_DIR%"
+
+
+def _substitute_scratch_dir(
+    command: List[str], plugin_id: str
+) -> Tuple[List[str], Optional[str]]:
+    """Resolve ``SCRATCH_DIR_PLACEHOLDER`` tokens to a fresh private directory.
+
+    Returns ``(command, scratch_dir)``. When the command contains no
+    placeholder, ``command`` is returned unchanged and ``scratch_dir`` is
+    ``None``. Otherwise a new ``tempfile.mkdtemp`` directory is created (mode
+    0700, unpredictable name) and every placeholder token is replaced with its
+    path; the caller is responsible for removing that directory afterwards.
+    """
+    if SCRATCH_DIR_PLACEHOLDER not in command:
+        return command, None
+
+    scratch_dir = tempfile.mkdtemp(prefix=f"secuscan-{plugin_id}-")
+    resolved = [
+        scratch_dir if token == SCRATCH_DIR_PLACEHOLDER else token
+        for token in command
+    ]
+    return resolved, scratch_dir
+
 from .auth import DEFAULT_OWNER_ID
 from .redaction import redact
 from .cache import get_cache
@@ -562,44 +597,59 @@ async def _execute_standard_scanner(
         if not command:
             raise ValueError("Failed to build command")
 
-        from .validation import validate_command_network_egress
-        cmd_valid, cmd_err = validate_command_network_egress(
-            command, safe_mode, plugin_id, task_id
-        )
-        if not cmd_valid:
-            raise ValueError(f"Command network egress validation failed: {cmd_err}")
-
-        # Apply Docker Sandboxing if enabled
-        if settings.docker_enabled:
-            await self._ensure_docker_network()
-            docker_image = plugin.docker_image or "alpine:latest"
-            docker_cmd = [
-                "docker",
-                "run",
-                "--rm",
-                "--name",
-                f"secuscan_task_{task_id}",
-                "--memory",
-                f"{settings.sandbox_memory_mb}m",
-                "--cpus",
-                str(settings.sandbox_cpu_quota),
-                "--cap-drop", "NET_RAW",
-                "--network", settings.docker_network,
-                docker_image,
-            ]
-            command = docker_cmd + command
-
-        logger.info(f"Executing task {task_id}: {' '.join(command)}")
-        await self._broadcast(task_id, "status", TaskStatus.RUNNING.value)
-        await self._broadcast_phase(task_id, ScanPhase.RUNNING_COMMAND.value)
-
-        # Execute command
-        start_time = time.time()
-        output, exit_code = await self._execute_command(
-            command,
-            task_id,
-            timeout=self._resolve_execution_timeout(inputs),
-        )
+        # Replace any reserved scratch-dir placeholder with a fresh private
+        # per-task directory, and guarantee its removal once execution ends.
+        command, scratch_dir = _substitute_scratch_dir(command, plugin_id)
+        try:
+            from .validation import validate_command_network_egress
+            cmd_valid, cmd_err = validate_command_network_egress(
+                command, safe_mode, plugin_id, task_id
+            )
+            if not cmd_valid:
+                raise ValueError(f"Command network egress validation failed: {cmd_err}")
+
+            # Apply Docker Sandboxing if enabled
+            if settings.docker_enabled:
+                await self._ensure_docker_network()
+                docker_image = plugin.docker_image or "alpine:latest"
+                docker_cmd = [
+                    "docker",
+                    "run",
+                    "--rm",
+                    "--name",
+                    f"secuscan_task_{task_id}",
+                    "--memory",
+                    f"{settings.sandbox_memory_mb}m",
+                    "--cpus",
+                    str(settings.sandbox_cpu_quota),
+                    "--cap-drop", "NET_RAW",
+                    "--network", settings.docker_network,
+                ]
+                # A plugin that resolved SCRATCH_DIR_PLACEHOLDER got a path on
+                # the *host*; the container has its own filesystem, so bind-mount
+                # that private dir into the container at the same path or the
+                # tool cannot write there. The host dir is 0700-private and is
+                # removed in the finally block below; mount it read-write so the
+                # tool can populate it.
+                if scratch_dir:
+                    docker_cmd += ["-v", f"{scratch_dir}:{scratch_dir}:rw"]
+                docker_cmd.append(docker_image)
+                command = docker_cmd + command
+
+            logger.info(f"Executing task {task_id}: {' '.join(command)}")
+            await self._broadcast(task_id, "status", TaskStatus.RUNNING.value)
+            await self._broadcast_phase(task_id, ScanPhase.RUNNING_COMMAND.value)
+
+            # Execute command
+            start_time = time.time()
+            output, exit_code = await self._execute_command(
+                command,
+                task_id,
+                timeout=self._resolve_execution_timeout(inputs),
+            )
+        finally:
+            if scratch_dir:
+                shutil.rmtree(scratch_dir, ignore_errors=True)
         duration = time.time() - start_time
 
         # Save raw output

plugins/amass/metadata.json

@@ -21,7 +21,7 @@
     "-d",
     "{target}",
     "-dir",
-    "/tmp/amass",
+    "%SECUSCAN_SCRATCH_DIR%",
     "-silent"
   ],
   "fields": [
@@ -59,5 +59,5 @@
   "capabilities": [
     "network"
   ],
-  "checksum": "e8dbca61aa046afe6ca08cf3188599f94866ea11ab1d6ded3394c6345d797e10"
+  "checksum": "78c4159360edbefa1afaaa3c6193f6540b71233fcd387333939f1769384d1d58"
 }

testing/backend/test_amass_plugin.py

@@ -114,7 +114,10 @@ def test_amass_command_renders_with_target(setup_test_environment):
     assert "-d" in command
     assert "secuscan.in" in command
     assert "-dir" in command
-    assert "/tmp/amass" in command
+    # The old hardcoded world-writable /tmp/amass path (issue #1803) must be gone,
+    # replaced by the reserved scratch-dir placeholder the executor resolves per task.
+    assert "/tmp/amass" not in command
+    assert "%SECUSCAN_SCRATCH_DIR%" in command
     assert "-silent" in command
 
 
@@ -131,7 +134,7 @@ def test_amass_command_full_token_sequence(setup_test_environment):
         "-d",
         "secuscan.in",
         "-dir",
-        "/tmp/amass",
+        "%SECUSCAN_SCRATCH_DIR%",
         "-silent",
     ], f"Command template drift detected. Got: {command}"
 
@@ -148,7 +151,7 @@ def test_amass_drops_target_token_when_absent(setup_test_environment):
 
     assert rendered is not None
     assert not any("{" in token for token in rendered), "Unresolved placeholder leaked"
-    assert rendered == ["amass", "enum", "-d", "-dir", "/tmp/amass", "-silent"]
+    assert rendered == ["amass", "enum", "-d", "-dir", "%SECUSCAN_SCRATCH_DIR%", "-silent"]
 
     populated = manager.build_command("amass", {"target": "secuscan.in"})
     assert "secuscan.in" in populated