Start a container, detached, interactive. Cleans up any existing container with the same name first (clean slate). ``runtime`` selects an OCI runtime (e.g. ``runsc`` for gVisor). The active runtime is verified via ``docker inspect`` so a typo or missing registration fails loudly ins
(
image_tag: str,
name: str,
network: str = "none",
memory: str = "4g",
shm_size: str | None = None,
shell: str = "/bin/bash",
runtime: str | None = None,
env: dict[str, str] | None = None,
mounts: list[tuple[str, str]] | None = None,
)
| 22 | |
| 23 | |
| 24 | def run( |
| 25 | image_tag: str, |
| 26 | name: str, |
| 27 | network: str = "none", |
| 28 | memory: str = "4g", |
| 29 | shm_size: str | None = None, |
| 30 | shell: str = "/bin/bash", |
| 31 | runtime: str | None = None, |
| 32 | env: dict[str, str] | None = None, |
| 33 | mounts: list[tuple[str, str]] | None = None, |
| 34 | ) -> str: |
| 35 | """Start a container, detached, interactive. Cleans up any existing |
| 36 | container with the same name first (clean slate). |
| 37 | |
| 38 | ``runtime`` selects an OCI runtime (e.g. ``runsc`` for gVisor). The |
| 39 | active runtime is verified via ``docker inspect`` so a typo or missing |
| 40 | registration fails loudly instead of silently falling back to runc.""" |
| 41 | subprocess.run(["docker", "rm", "-f", name], capture_output=True) |
| 42 | runtime = runtime or os.environ.get("VULN_PIPELINE_DOCKER_RUNTIME") |
| 43 | extra: list[str] = [] |
| 44 | if runtime: |
| 45 | extra += ["--runtime", runtime] |
| 46 | if shm_size: |
| 47 | extra += ["--shm-size", shm_size] |
| 48 | for k, v in (env or {}).items(): |
| 49 | extra += ["-e", f"{k}={v}"] |
| 50 | for src, dst in (mounts or []): |
| 51 | extra += ["-v", f"{src}:{dst}:ro"] |
| 52 | r = subprocess.run( |
| 53 | [ |
| 54 | "docker", "run", "-dit", |
| 55 | *extra, |
| 56 | "--name", name, |
| 57 | "--network", network, |
| 58 | "--memory", memory, |
| 59 | image_tag, shell, |
| 60 | ], |
| 61 | check=False, |
| 62 | capture_output=True, |
| 63 | text=True, |
| 64 | ) |
| 65 | if r.returncode != 0: |
| 66 | raise RuntimeError( |
| 67 | f"docker run failed (exit {r.returncode}): {r.stderr.strip()}" |
| 68 | ) |
| 69 | actual_image, actual_runtime = subprocess.run( |
| 70 | ["docker", "inspect", name, "--format", |
| 71 | "{{.Config.Image}}\t{{.HostConfig.Runtime}}"], |
| 72 | capture_output=True, text=True, check=True, |
| 73 | ).stdout.rstrip("\n").split("\t") |
| 74 | if actual_image != image_tag: |
| 75 | raise RuntimeError( |
| 76 | f"container {name} has wrong image: requested {image_tag!r}, got {actual_image!r}" |
| 77 | ) |
| 78 | if runtime and actual_runtime != runtime: |
| 79 | raise RuntimeError( |
| 80 | f"container {name} runtime mismatch: requested {runtime!r}, " |
| 81 | f"docker reports {actual_runtime!r}" |
nothing calls this directly
no outgoing calls
no test coverage detected