Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,14 @@ ENV PYTHONDONTWRITEBYTECODE=1 \

COPY --from=docker-cli /usr/local/bin/docker /usr/local/bin/docker

RUN apk add --no-cache git
# podman: daemonless container runtime for container-mode scans on
# containerd-only nodes (EKS/k3s) where no dockerd/docker.sock exists.
# docker CLI above stays for environments that do have a real daemon --
# aspm_cli.utils.container_runtime picks whichever is actually usable.
RUN apk add --no-cache git podman \
&& mkdir -p /etc/containers \
&& printf '{"default":[{"type":"insecureAcceptAnything"}]}\n' > /etc/containers/policy.json \
&& printf '[storage]\ndriver = "vfs"\n' > /etc/containers/storage.conf

COPY . /CODE

Expand Down
75 changes: 75 additions & 0 deletions aspm_cli/utils/container_runtime.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
import os
import shutil
import subprocess

from aspm_cli.utils.logger import Logger

# Priority order: docker first (preserves existing behavior wherever a real
# docker daemon is reachable), then daemonless/containerd-native tools that
# work in EKS/k3s pods without /var/run/docker.sock, DinD, or --privileged.
_RUNTIME_CANDIDATES = ("docker", "nerdctl", "podman")

# docker/nerdctl are daemon clients -- their binary can be on PATH (e.g.
# baked into the scanner image) while no daemon/socket is reachable, which
# is exactly the EKS/containerd failure this module exists to route around.
# podman is daemonless (runs via runc in the pod's own namespace), so it
# needs no such probe.
_DAEMON_PROBE = {
"docker": ["docker", "info"],
"nerdctl": ["nerdctl", "info"],
}

_cached_runtime = None


def _is_usable(binary: str) -> bool:
probe = _DAEMON_PROBE.get(binary)
if not probe:
return True
try:
return subprocess.run(probe, capture_output=True, timeout=5).returncode == 0
except Exception:
return False


def get_container_runtime() -> str:
"""
Resolve the Docker-CLI-compatible binary to use for `run`/`pull`/`image
inspect` in container-mode scans.

Override with ACCUKNOX_CONTAINER_RUNTIME. Otherwise picks the first of
docker, nerdctl, podman that is both on PATH and actually usable --
docker/nerdctl only count if their daemon responds, so a scanner image
that merely bundles the docker CLI (no daemon behind it) automatically
falls through to podman instead of failing at `docker pull`.
"""
global _cached_runtime
if _cached_runtime:
return _cached_runtime

override = os.getenv("ACCUKNOX_CONTAINER_RUNTIME")
if override:
if not shutil.which(override):
raise RuntimeError(
f"ACCUKNOX_CONTAINER_RUNTIME={override!r} not found on PATH"
)
_cached_runtime = override
return override

for candidate in _RUNTIME_CANDIDATES:
if shutil.which(candidate) and _is_usable(candidate):
Logger.get_logger().debug(f"Using container runtime: {candidate}")
_cached_runtime = candidate
return candidate

raise RuntimeError(
"No usable container runtime found (looked for: "
+ ", ".join(_RUNTIME_CANDIDATES)
+ " -- docker/nerdctl need a reachable daemon, podman needs none). "
"Install podman in the scanner image, or run the scan without --container-mode."
)


def _reset_cache_for_tests():
global _cached_runtime
_cached_runtime = None
11 changes: 6 additions & 5 deletions aspm_cli/utils/docker_pull.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,26 @@
import subprocess

from aspm_cli.utils.container_runtime import get_container_runtime
from aspm_cli.utils.logger import Logger


def _image_exists_locally(image: str) -> bool:
result = subprocess.run(
["docker", "image", "inspect", image],
[get_container_runtime(), "image", "inspect", image],
capture_output=True,
text=True,
)
return result.returncode == 0


def docker_pull(image: str, platform: str = None):
"""Pull a Docker image, or use it if already present locally."""
"""Pull a container image (via docker/nerdctl/podman), or use it if already present locally."""
if _image_exists_locally(image):
Logger.get_logger().debug(f"Using local Docker image: {image}")
Logger.get_logger().debug(f"Using local container image: {image}")
return

Logger.get_logger().debug(f"Pulling Docker image: {image}")
cmd = ["docker", "pull"]
Logger.get_logger().debug(f"Pulling container image: {image}")
cmd = [get_container_runtime(), "pull"]
if platform:
cmd.extend(["--platform", platform])
cmd.append(image)
Expand Down
7 changes: 6 additions & 1 deletion aspm_cli/utils/docker_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
from pathlib import Path
from typing import List, Optional

from aspm_cli.utils.container_runtime import get_container_runtime

# Subcommands that inspect container images or rootfs archives via the Docker API.
_TRIVY_IMAGE_SUBCOMMANDS = frozenset({"image", "rootfs", "container", "i", "vm"})

Expand Down Expand Up @@ -80,7 +82,10 @@ def build_docker_run_prefix(
host_path: Optional[str] = None,
mount_docker_socket: bool = False,
) -> List[str]:
cmd = ["docker", "run", "--rm"]
# get_container_runtime() picks docker/nerdctl/podman by actual usability
# (not mere PATH presence), so this keeps working on a containerd-only
# node (EKS/k3s) with no dockerd -- see aspm_cli.utils.container_runtime.
cmd = [get_container_runtime(), "run", "--rm"]
if mount_docker_socket:
cmd.extend(docker_socket_mount_args())
cmd.extend(docker_workdir_mount(workdir, host_path))
Expand Down
130 changes: 130 additions & 0 deletions tests/test_container_runtime.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
import os
import subprocess

import pytest

import sys

from aspm_cli.utils import container_runtime
import aspm_cli.utils.docker_pull # noqa: F401 - ensures the real submodule is in sys.modules
from aspm_cli.scan.sast import SASTScanner

docker_pull_module = sys.modules["aspm_cli.utils.docker_pull"]


@pytest.fixture(autouse=True)
def _reset_runtime_cache(monkeypatch):
container_runtime._reset_cache_for_tests()
monkeypatch.delenv("ACCUKNOX_CONTAINER_RUNTIME", raising=False)
yield
container_runtime._reset_cache_for_tests()


def test_prefers_docker_when_daemon_reachable(monkeypatch):
monkeypatch.setattr(
container_runtime.shutil, "which", lambda name: f"/usr/bin/{name}" if name == "docker" else None
)
monkeypatch.setattr(container_runtime, "_is_usable", lambda name: True)
assert container_runtime.get_container_runtime() == "docker"


def test_falls_back_to_nerdctl_without_docker(monkeypatch):
monkeypatch.setattr(
container_runtime.shutil, "which", lambda name: f"/usr/bin/{name}" if name == "nerdctl" else None
)
monkeypatch.setattr(container_runtime, "_is_usable", lambda name: True)
assert container_runtime.get_container_runtime() == "nerdctl"


def test_falls_back_to_podman_without_docker_or_nerdctl(monkeypatch):
monkeypatch.setattr(
container_runtime.shutil, "which", lambda name: f"/usr/bin/{name}" if name == "podman" else None
)
assert container_runtime.get_container_runtime() == "podman"


def test_raises_when_no_runtime_found(monkeypatch):
monkeypatch.setattr(container_runtime.shutil, "which", lambda name: None)
with pytest.raises(RuntimeError, match="No usable container runtime found"):
container_runtime.get_container_runtime()


def test_docker_binary_present_but_daemon_unreachable_falls_back_to_podman(monkeypatch):
"""Reproduces the EKS failure: scanner image bundles the docker CLI but
there is no dockerd behind /var/run/docker.sock. `docker` must not win
just because the binary exists on PATH."""
monkeypatch.setattr(
container_runtime.shutil,
"which",
lambda name: f"/usr/bin/{name}" if name in ("docker", "podman") else None,
)

def fake_run(cmd, **kwargs):
if cmd[0] == "docker":
raise FileNotFoundError("dial unix /var/run/docker.sock: no such file or directory")
raise AssertionError(f"unexpected probe: {cmd}")

monkeypatch.setattr(container_runtime.subprocess, "run", fake_run)
assert container_runtime.get_container_runtime() == "podman"


def test_env_override_wins(monkeypatch):
monkeypatch.setenv("ACCUKNOX_CONTAINER_RUNTIME", "podman")
monkeypatch.setattr(
container_runtime.shutil, "which", lambda name: f"/usr/bin/{name}" if name == "podman" else None
)
assert container_runtime.get_container_runtime() == "podman"


def test_env_override_missing_binary_raises(monkeypatch):
monkeypatch.setenv("ACCUKNOX_CONTAINER_RUNTIME", "nerdctl")
monkeypatch.setattr(container_runtime.shutil, "which", lambda name: None)
with pytest.raises(RuntimeError, match="ACCUKNOX_CONTAINER_RUNTIME"):
container_runtime.get_container_runtime()


def test_result_cached_across_calls(monkeypatch):
calls = []

def fake_which(name):
calls.append(name)
return "/usr/bin/nerdctl" if name == "nerdctl" else None

monkeypatch.setattr(container_runtime.shutil, "which", fake_which)
monkeypatch.setattr(container_runtime, "_is_usable", lambda name: True)
container_runtime.get_container_runtime()
container_runtime.get_container_runtime()
# docker (miss) + nerdctl (hit) probed once; second call served from cache
assert calls == ["docker", "nerdctl"]


def test_docker_pull_uses_resolved_runtime(monkeypatch):
monkeypatch.setattr(container_runtime, "_cached_runtime", "nerdctl")
seen_cmds = []

def fake_run(cmd, **kwargs):
seen_cmds.append(cmd)
return subprocess.CompletedProcess(cmd, returncode=0, stdout="", stderr="")

monkeypatch.setattr(subprocess, "run", fake_run)
docker_pull_module.docker_pull("some/image:tag")

assert seen_cmds[0][0] == "nerdctl"
assert seen_cmds[0][1:3] == ["image", "inspect"]


def test_sast_container_mode_command_uses_resolved_runtime(monkeypatch):
monkeypatch.setattr(container_runtime, "_cached_runtime", "podman")
scanner = SASTScanner(command="scan .", container_mode=True)
cmd = scanner._build_sast_command(["scan", "--json"])
assert cmd[0] == "podman"
assert "run" in cmd


def test_sast_non_container_mode_unaffected(monkeypatch):
monkeypatch.setattr(
"aspm_cli.scan.sast.ToolManager.get_path", lambda name: "/opt/opengrep/opengrep"
)
scanner = SASTScanner(command="scan .", container_mode=False)
cmd = scanner._build_sast_command(["scan", "--json"])
assert cmd[0] == "/opt/opengrep/opengrep"