kevinveenbirkenbach/computer-playbook
1
0 Fork 0
mirror of https://github.com/kevinveenbirkenbach/computer-playbook.git synced 2026-11-06 02:54:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(container): drive exec timeout and CA paths from group_vars SPOTs

Browse Source 
Two wrapper behaviours move onto single sources of truth:

Exec timeout: SYS_TIMEOUT_CONTAINER_EXEC (07_schedule.yml, default 7d so a
hard kill never beats a production migration; the dev/CI inventory
tightens it to 60min) reaches every task host via the play environment as
CONTAINER_EXEC_TIMEOUT. The wrapper prefixes only the exec subcommand with
coreutils timeout --kill-after=15. Ansible task timeouts are SIGALRM-based
and slept through the last wedged docker socket for 4.5 hours; the OS
level kill still fires there and returns 124, which until/failed_when and
the rescue diagnostics react to. Unset or 0 keeps todays behaviour.

CA container paths: the bind-mount targets for the injected root CA and
wrapper script lived as string literals in container.py, compose_ca.py
and the filter plugin. They now live once in the CA_TRUST group_vars SPOT
(inject_cert_container, inject_wrapper_container) and reach container.py
as env vars (play environment plus CA_TRUST_SYSTEMD_ENV) and compose_ca.py
as CLI arguments built by the compose_ca_inject_cmd lookup. A caller that
provides the host CA env but not the container paths is half-migrated and
fails loudly instead of silently dropping CA injection; a host without any
CA env keeps the soft fallback. All tests read the paths from the SPOT,
and a regression test asserts the runtime scripts carry no literal copy.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Fable 5 committed 2026-07-11 10:58:28 +02:00
1 parent 81d52f9d16
commit e2f8d4c304
13 files changed
+373 -159
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

group_vars/all/02_tls.yml
+4
View File
@@ -30,10 +30,14 @@ CA_TRUST:
trust_name: "{{ DOMAIN_PRIMARY | lower }}"
wrapper_host: "{{ BIN_CA_TRUST }}"
inject_script: "{{ BIN_CA_INJECT }}"
inject_cert_container: "/tmp/infinito/ca/root-ca.crt"
inject_wrapper_container: "/tmp/infinito/bin/with-ca-trust.sh"
CA_TRUST_SYSTEMD_ENV:
- "CA_TRUST_CERT_HOST={{ CA_TRUST.cert_host }}"
- "CA_TRUST_WRAPPER_HOST={{ CA_TRUST.wrapper_host }}"
- "CA_TRUST_NAME={{ CA_TRUST.trust_name }}"
- "CA_TRUST_CERT_CONTAINER={{ CA_TRUST.inject_cert_container }}"
- "CA_TRUST_WRAPPER_CONTAINER={{ CA_TRUST.inject_wrapper_container }}"
CA_TRUST_SYSTEMD_ENV_SELFSIGNED: >-
{{ CA_TRUST_SYSTEMD_ENV if (TLS_MODE | lower) == 'self_signed' else [] }}
CA_CONTAINER_ENABLED: true # When falsy the container wrapper skips CA injection on `container run`
group_vars/all/07_schedule.yml
+1
View File
@@ -13,6 +13,7 @@ SYS_TIMEOUT_CLEANUP_SERVICES: "15min"
SYS_TIMEOUT_DOCKER_UPDATE: "20min"
SYS_TIMEOUT_STORAGE_OPTIMIZER: "{{ SYS_TIMEOUT_DOCKER_UPDATE }}"
SYS_TIMEOUT_BACKUP_SERVICES: "60min"
SYS_TIMEOUT_CONTAINER_EXEC: "7d" # Hard OS-level kill for `container exec`; a week so it never beats a production migration
SYS_SCHEDULE_HEALTH_BTRFS: "*-*-* 00:00:00" # Check once per day the btrfs for errors
SYS_SCHEDULE_HEALTH_JOURNALCTL: "*-*-* 00:00:00" # Check once per day the journalctl for errors
inventories/development/default.yml
+1
View File
@@ -23,6 +23,7 @@ INFINITO_IMAGE_TAG: "{{ lookup('env', 'INFINITO_IMAGE_TAG') | default('latest',
INFINITO_GHCR_MIRROR_PREFIX: "{{ lookup('env', 'INFINITO_GHCR_MIRROR_PREFIX') | default('', true) }}"
CA_CONTAINER_ENABLED: "{{ lookup('env', 'INFINITO_CONTAINER_CA_ENABLED') | bool }}"
test_e2e_playwright_keep: "{{ lookup('env', 'INFINITO_PLAYWRIGHT_KEEP') | bool }}"
SYS_TIMEOUT_CONTAINER_EXEC: "60min"
networks:
internet:
ip4: "{{ lookup('env', 'INFINITO_IP4') }}"
playbook.yml
+3
View File
@@ -6,12 +6,15 @@
CA_TRUST_CERT_HOST: "{{ CA_TRUST.cert_host }}"
CA_TRUST_WRAPPER_HOST: "{{ CA_TRUST.wrapper_host }}"
CA_TRUST_NAME: "{{ CA_TRUST.trust_name }}"
CA_TRUST_CERT_CONTAINER: "{{ CA_TRUST.inject_cert_container }}"
CA_TRUST_WRAPPER_CONTAINER: "{{ CA_TRUST.inject_wrapper_container }}"
CA_CONTAINER_ENABLED: "{{ CA_CONTAINER_ENABLED | int }}"
COMPOSE_HTTP_TIMEOUT: "600"
DOCKER_CLIENT_TIMEOUT: "600"
DOCKER_BUILDKIT: "{{ BUILDKIT_ENABLED | int }}"
COMPOSE_DOCKER_CLI_BUILD: "{{ BUILDKIT_ENABLED | int }}"
INFINITO_RESCUE_DIAGNOSTICS_DIR: "{{ DIR_RESCUE_DIAGNOSTICS }}"
CONTAINER_EXEC_TIMEOUT: "{{ SYS_TIMEOUT_CONTAINER_EXEC }}"
tasks:
- name: "Load 'constructor' tasks"
include_tasks: "tasks/stages/01_constructor.yml"
plugins/filter/ca_trust_paths.py
+4 -8
View File
@@ -1,19 +1,15 @@
"""Filter ``ca_cert_host`` plus the container-side CA constant: SPOT for the root-CA cert locations.
"""Filter ``ca_cert_host``: SPOT for the host-side root-CA cert location.
{{ SOFTWARE_DOMAIN | ca_cert_host }} -> /etc/<software_domain>/ca/root-ca.crt
``ca_cert_host`` is where sys-ca-selfsigned provisions the root CA on the host.
``CA_CONTAINER_CERT`` is the bind-mount target sys-svc-compose-ca/sys-svc-container
inject into every container; the standalone runtime scripts
(roles/sys-svc-compose-ca/files/compose_ca.py, roles/sys-svc-container/files/container.py)
cannot import this module and carry the literal themselves; the unit test pins
them to this SPOT.
The container-side bind-mount targets live in the ``CA_TRUST`` group_vars SPOT
(``inject_cert_container`` / ``inject_wrapper_container``) and reach the
standalone runtime scripts as env vars and CLI arguments.
"""
from __future__ import annotations
CA_CONTAINER_CERT = "/tmp/infinito/ca/root-ca.crt" # noqa: S108 - fixed bind-mount target inside containers, not a writable temp file
def ca_cert_host(software_domain: str) -> str:
return f"/etc/{software_domain}/ca/root-ca.crt"
roles/sys-svc-compose-ca/files/compose_ca.py
+20 -16
View File
@@ -17,12 +17,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any
# This script is deployed to the target host via `ansible.builtin.copy`
# (see roles/sys-svc-compose-ca/tasks/01_core.yml). The deploy target
# does NOT have the project's `utils/` package on PYTHONPATH, so direct
# yaml.safe_load / yaml.safe_dump are the only option here. Each call
# below carries an explicit `# nocheck: direct-yaml` marker that the lint
# `tests/lint/repository/yaml/test_no_direct_calls.py` honours.
import yaml
@@ -39,8 +34,6 @@ def run(
timeout: int = 600,
capture: bool = True,
) -> tuple[int, str, str]:
# killpg the whole group on timeout: a daemon-side docker/buildx grandchild
# can keep the captured pipe open and make subprocess's own timeout inert.
pipe = subprocess.PIPE if capture else subprocess.DEVNULL
proc = subprocess.Popen(
cmd,
@@ -488,7 +481,6 @@ def _compose_cmd_with_profile(base_cmd: list[str], profile: str) -> list[str]:
if len(base_cmd) < 2 or base_cmd[0] != "docker" or base_cmd[1] != "compose":
die(f"Invalid compose base cmd: {base_cmd}")
# Insert after the compose wrapper prefix
return [*base_cmd[:2], "--profile", profile, *base_cmd[2:]]
@@ -519,6 +511,8 @@ def render_override(
ca_host: str,
wrapper_host: str,
trust_name: str,
ca_container: str,
wrapper_container: str,
wrap: bool = True,
) -> dict[str, Any]:
"""
@@ -540,11 +534,6 @@ def render_override(
Docker Compose interpolates $VARS in YAML strings on the host.
We escape any $ in the command argv with $$ so container-side expansion works.
"""
# Container-internal CA-injection paths bind-mounted from the host.
# Not user-controllable; well-known by the role's compose template.
ca_container = "/tmp/infinito/ca/root-ca.crt" # noqa: S108
wrapper_container = "/tmp/infinito/bin/with-ca-trust.sh" # noqa: S108
out_services: dict[str, Any] = {}
image_meta = gather_image_meta(
@@ -589,7 +578,6 @@ def render_override(
image = svc.get("image")
if not isinstance(image, str) or not image.strip():
# If there is no image, we can only wrap if effective command is explicitly defined
if not svc_ep and not svc_cmd:
die(
f"Service '{name}' has no image and no entrypoint/command in composed config"
@@ -646,6 +634,16 @@ def main() -> int:
required=True,
help="Host path to wrapper script (bind-mounted)",
)
ap.add_argument(
"--ca-container",
required=True,
help="Container path the CA cert is bind-mounted to",
)
ap.add_argument(
"--wrapper-container",
required=True,
help="Container path the wrapper script is bind-mounted to",
)
ap.add_argument(
"--trust-name",
required=True,
@@ -669,6 +667,8 @@ def main() -> int:
ca_host = str(args.ca_host).strip()
wrapper_host = str(args.wrapper_host).strip()
trust_name = str(args.trust_name).strip()
ca_container = str(args.ca_container).strip()
wrapper_container = str(args.wrapper_container).strip()
if not ca_host:
die("--ca-host must be non-empty")
@@ -676,6 +676,10 @@ def main() -> int:
die("--wrapper-host must be non-empty")
if not trust_name:
die("--trust-name must be non-empty")
if not ca_container:
die("--ca-container must be non-empty")
if not wrapper_container:
die("--wrapper-container must be non-empty")
env = dict(os.environ)
@@ -698,8 +702,6 @@ def main() -> int:
service_to_compose_cmd: dict[str, list[str]] = {}
if args.no_wrapper:
# Avoid `docker compose config` here: its dotenv parser aborts on swarm
# env files with verbatim special-char secrets. wrap=False needs only names.
for cf in compose_files:
doc = parse_yaml(
cf.read_text(encoding="utf-8"), # nocheck: direct-yaml
@@ -749,6 +751,8 @@ def main() -> int:
ca_host=ca_host,
wrapper_host=wrapper_host,
trust_name=trust_name,
ca_container=ca_container,
wrapper_container=wrapper_container,
wrap=not args.no_wrapper,
)
else:
roles/sys-svc-compose/lookup_plugins/compose_ca_inject_cmd.py
+18 -16
View File
@@ -31,7 +31,6 @@ def _require(d: Any, key: str, expected_type: Any, *, label: str) -> Any:
def _shell_quote(s: str) -> str:
# Safe single-quote shell quoting: ' -> '"'"'
return "'" + s.replace("'", "'\"'\"'") + "'"
@@ -77,9 +76,6 @@ class LookupModule(LookupBase):
def run(self, terms, variables: dict | None = None, **kwargs):
variables = variables or {}
# ---------------------------------------------------------------------
# Validate input
# ---------------------------------------------------------------------
if not terms or len(terms) != 1:
raise AnsibleError(
"compose_ca_inject_cmd: exactly one term required (application_id)"
@@ -99,9 +95,6 @@ class LookupModule(LookupBase):
templar = getattr(self, "_templar", None)
# ---------------------------------------------------------------------
# Resolve compose paths via lookup('container', ...)
# ---------------------------------------------------------------------
instance_dir = _docker_lookup(
self,
application_id=application_id,
@@ -136,9 +129,6 @@ class LookupModule(LookupBase):
"compose_ca_inject_cmd: output basename resolved to empty"
)
# ---------------------------------------------------------------------
# CA_TRUST variables
# ---------------------------------------------------------------------
ca_trust = _require(variables, "CA_TRUST", dict, label="variable")
script_path = render_ansible_strict(
@@ -169,10 +159,21 @@ class LookupModule(LookupBase):
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
ca_container = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "inject_cert_container", str, label="CA_TRUST"),
var_name="CA_TRUST.inject_cert_container",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
wrapper_container = render_ansible_strict(
templar=templar,
raw=_require(ca_trust, "inject_wrapper_container", str, label="CA_TRUST"),
var_name="CA_TRUST.inject_wrapper_container",
err_prefix="compose_ca_inject_cmd",
variables=variables,
)
# ---------------------------------------------------------------------
# Resolve compose files (without CA override!)
# ---------------------------------------------------------------------
compose_file_args_lkp = lookup_loader.get(
"compose_file_args", self._loader, self._templar
)
@@ -187,9 +188,6 @@ class LookupModule(LookupBase):
"compose_ca_inject_cmd: compose_file_args returned empty"
)
# ---------------------------------------------------------------------
# Build command
# ---------------------------------------------------------------------
cmd = [
"python3",
_shell_quote(script_path),
@@ -212,6 +210,10 @@ class LookupModule(LookupBase):
_shell_quote(ca_host),
"--wrapper-host",
_shell_quote(wrapper_host),
"--ca-container",
_shell_quote(ca_container),
"--wrapper-container",
_shell_quote(wrapper_container),
"--trust-name",
_shell_quote(trust_name),
]
roles/sys-svc-container/files/container.py
+73 -53
View File
@@ -9,10 +9,6 @@ import subprocess
import sys
from pathlib import Path
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def die(msg: str, code: int = 2) -> None:
print(f"[container] {msg}", file=sys.stderr)
@@ -30,7 +26,6 @@ def must_exist(path: str, label: str) -> str:
return str(p)
# docker run flags that take a value
FLAGS_TAKE_VALUE = {
"-e",
"--env",
@@ -74,11 +69,6 @@ FLAGS_TAKE_VALUE = {
}
# ---------------------------------------------------------------------------
# docker run argument parsing
# ---------------------------------------------------------------------------
def split_docker_run_argv(argv: list[str]) -> tuple[list[str], list[str]]:
"""
Split argv into:
@@ -181,11 +171,6 @@ def extract_pull_policy(run_opts: list[str]) -> str:
return policy
# ---------------------------------------------------------------------------
# Docker helpers
# ---------------------------------------------------------------------------
def docker_pull(image: str) -> None:
try:
p = subprocess.run(
@@ -253,36 +238,49 @@ def try_inspect_entrypoint_with_pull(image: str, pull_policy: str) -> list[str]:
raise
# ---------------------------------------------------------------------------
# CA handling (SOFT)
# ---------------------------------------------------------------------------
def require_ca_env_soft() -> tuple[str, str, str] | None:
def require_ca_env_soft() -> tuple[str, str, str, str, str] | None:
"""
Return (ca_cert_host, wrapper_host, trust_name)
or None if CA injection is not available.
Return (ca_cert_host, wrapper_host, trust_name, ca_container,
wrapper_container) or None if CA injection is not available. The two
container-side paths come from the CA_TRUST group_vars SPOT, forwarded
as env like the host-side trio.
"""
ca_host = os.environ.get("CA_TRUST_CERT_HOST", "").strip()
wrapper_host = os.environ.get("CA_TRUST_WRAPPER_HOST", "").strip()
trust_name = os.environ.get("CA_TRUST_NAME", "").strip()
ca_container = os.environ.get("CA_TRUST_CERT_CONTAINER", "").strip()
wrapper_container = os.environ.get("CA_TRUST_WRAPPER_CONTAINER", "").strip()
missing = []
host_missing = []
if not ca_host:
missing.append("CA_TRUST_CERT_HOST")
host_missing.append("CA_TRUST_CERT_HOST")
if not wrapper_host:
missing.append("CA_TRUST_WRAPPER_HOST")
host_missing.append("CA_TRUST_WRAPPER_HOST")
if not trust_name:
missing.append("CA_TRUST_NAME")
host_missing.append("CA_TRUST_NAME")
if missing:
container_missing = []
if not ca_container:
container_missing.append("CA_TRUST_CERT_CONTAINER")
if not wrapper_container:
container_missing.append("CA_TRUST_WRAPPER_CONTAINER")
if host_missing:
warn(
"CA injection disabled (missing env: "
+ ", ".join(missing)
+ ", ".join(host_missing + container_missing)
+ "). Falling back to plain 'docker run'."
)
return None
if container_missing:
die(
"CA injection misconfigured: "
+ ", ".join(container_missing)
+ " not set while the CA_TRUST host env is present. Redeploy the "
"systemd units / update the caller env to the CA_TRUST group_vars SPOT."
)
try:
ca_host = must_exist(ca_host, "CA trust certificate")
wrapper_host = must_exist(wrapper_host, "CA trust wrapper script")
@@ -292,12 +290,7 @@ def require_ca_env_soft() -> tuple[str, str, str] | None:
)
return None
return ca_host, wrapper_host, trust_name
# ---------------------------------------------------------------------------
# Execution
# ---------------------------------------------------------------------------
return ca_host, wrapper_host, trust_name, ca_container, wrapper_container
def exec_docker(cmd: list[str], debug: bool) -> int:
@@ -327,7 +320,7 @@ def container_run(argv: list[str], debug: bool, with_ca: bool) -> int:
if not ca_env:
return exec_docker(["docker", "run", *argv], debug=debug)
ca_host, wrapper_host, trust_name = ca_env
ca_host, wrapper_host, trust_name, ca_container, wrapper_container = ca_env
run_opts, image_and_args = split_docker_run_argv(argv)
pull_policy = extract_pull_policy(run_opts)
@@ -336,11 +329,6 @@ def container_run(argv: list[str], debug: bool, with_ca: bool) -> int:
image = image_and_args[0]
user_args = image_and_args[1:]
# Container-internal CA-injection paths bind-mounted from the host.
# Not user-controllable; matched on the host side by sys-svc-compose-ca.
ca_container = "/tmp/infinito/ca/root-ca.crt" # noqa: S108
wrapper_container = "/tmp/infinito/bin/with-ca-trust.sh" # noqa: S108
inject_opts: list[str] = [
"-v",
f"{ca_host}:{ca_container}:ro",
@@ -369,9 +357,6 @@ def container_run(argv: list[str], debug: bool, with_ca: bool) -> int:
"Image has no ENTRYPOINT and none was provided. "
"Injecting CA env vars without entrypoint wrapper."
)
# Inject CA cert as a volume + env vars so Node.js (NODE_EXTRA_CA_CERTS),
# curl (CURL_CA_BUNDLE), Python requests (REQUESTS_CA_BUNDLE), and other
# tools can trust the internal CA even without a wrapper entrypoint.
ca_inject_opts: list[str] = [
"-v",
f"{ca_host}:{ca_container}:ro",
@@ -399,9 +384,7 @@ def container_run(argv: list[str], debug: bool, with_ca: bool) -> int:
if debug:
print(">>> " + " ".join(shlex.quote(x) for x in final_cmd), file=sys.stderr)
# List-form exec: no shell parses the argv. CMD comes from this
# script's own argv (the wrapper), not from external user input.
os.execvp(final_cmd[0], final_cmd) # noqa: S606
os.execvp(final_cmd[0], final_cmd) # noqa: S606 list-form exec, argv comes from the wrapper itself
return 0
@@ -414,9 +397,44 @@ def passthrough_any(subcmd: str, argv: list[str], debug: bool) -> int:
return exec_docker(["docker", subcmd, *argv], debug=debug)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
_EXEC_TIMEOUT_UNITS = {"s": 1, "min": 60, "h": 3600, "d": 86400}
def parse_exec_timeout_seconds(value: str) -> int:
"""Parse a systemd-style duration ('60min', '7d', '45s', '2h') to seconds.
Args:
value: duration string; a bare number counts as seconds.
Returns:
The duration in whole seconds.
"""
raw = value.strip()
for suffix, factor in sorted(_EXEC_TIMEOUT_UNITS.items(), key=lambda i: -len(i[0])):
if raw.endswith(suffix):
return int(float(raw[: -len(suffix)])) * factor
return int(float(raw))
def exec_timeout_prefix() -> list[str]:
"""coreutils timeout prefix for `container exec` from CONTAINER_EXEC_TIMEOUT.
Returns:
['timeout', '--kill-after=15', '<seconds>'] when the env var holds a
positive duration; [] when it is unset, empty or 0 (no limit). Ansible
task timeouts are SIGALRM-based and sleep through a wedged docker
socket; this OS-level kill is the enforcement that still fires there.
"""
raw = os.environ.get("CONTAINER_EXEC_TIMEOUT", "").strip()
if not raw or raw == "0":
return []
try:
seconds = parse_exec_timeout_seconds(raw)
except ValueError:
die(f"Invalid CONTAINER_EXEC_TIMEOUT duration: {raw!r}")
if seconds <= 0:
return []
return ["timeout", "--kill-after=15", str(seconds)]
def main() -> int:
@@ -456,9 +474,11 @@ def main() -> int:
if cmd == "docker":
return exec_docker(["docker", *args], debug=debug)
# Default: passthrough ANY docker subcommand (info/version/system/network/volume/buildx/...)
# Example: `container info` -> `docker info`
# `container system prune -af` -> `docker system prune -af`
if cmd == "exec":
return exec_docker(
[*exec_timeout_prefix(), "docker", "exec", *args], debug=debug
)
return passthrough_any(cmd, args, debug=debug)
tests/unit/plugins/filter/test_ca_trust_paths.py
+21 -11
View File
@@ -1,14 +1,19 @@
import unittest
from plugins.filter.ca_trust_paths import CA_CONTAINER_CERT, ca_cert_host
from plugins.filter.ca_trust_paths import ca_cert_host
from utils import PROJECT_ROOT
from utils.cache.files import read_text
from utils.cache.yaml import load_yaml
RUNTIME_SCRIPTS = (
PROJECT_ROOT / "roles/sys-svc-compose-ca/files/compose_ca.py",
PROJECT_ROOT / "roles/sys-svc-container/files/container.py",
)
_CA_TRUST = load_yaml(str(PROJECT_ROOT / "group_vars" / "all" / "02_tls.yml"))[
"CA_TRUST"
]
class TestCaTrustPaths(unittest.TestCase):
def test_host_path_derives_from_software_domain(self):
@@ -17,18 +22,23 @@ class TestCaTrustPaths(unittest.TestCase):
)
self.assertEqual(ca_cert_host("example.org"), "/etc/example.org/ca/root-ca.crt")
def test_container_path_is_stable(self):
self.assertEqual(CA_CONTAINER_CERT, "/tmp/infinito/ca/root-ca.crt")
def test_group_vars_spot_holds_plain_container_paths(self):
for key in ("inject_cert_container", "inject_wrapper_container"):
value = _CA_TRUST[key]
self.assertIsInstance(value, str)
self.assertTrue(value.startswith("/"), f"{key} must be an absolute path")
self.assertNotIn("{{", value, f"{key} must be a plain string")
def test_runtime_scripts_pin_the_container_spot(self):
expected = f'"{CA_CONTAINER_CERT}"'
def test_runtime_scripts_carry_no_container_path_literal(self):
for script in RUNTIME_SCRIPTS:
self.assertIn(
expected,
read_text(str(script)),
f"{script} no longer carries the SPOT container CA path; "
"update plugins/filter/ca_trust_paths.py and the script together",
)
content = read_text(str(script))
for key in ("inject_cert_container", "inject_wrapper_container"):
self.assertNotIn(
_CA_TRUST[key],
content,
f"{script} hardcodes {_CA_TRUST[key]}; it must take the value "
"from env/CLI so group_vars stays the only SPOT",
)
def test_env_handler_exports_the_host_spot(self):
from utils.env.builder import BuildContext, EnvBuilder
tests/unit/roles/sys-svc-compose-ca/files/test_compose_ca.py
+63 -40
View File
@@ -5,10 +5,16 @@ import unittest
from pathlib import Path
from unittest.mock import patch
from utils.cache.yaml import load_yaml_str
from utils.cache.yaml import load_yaml, load_yaml_str
from . import PROJECT_ROOT
_CA_TRUST = load_yaml(str(PROJECT_ROOT / "group_vars" / "all" / "02_tls.yml"))[
"CA_TRUST"
]
CA_CERT_CONTAINER = _CA_TRUST["inject_cert_container"]
CA_WRAPPER_CONTAINER = _CA_TRUST["inject_wrapper_container"]
def _load_module(rel_path: str, name: str):
path = PROJECT_ROOT / rel_path
@@ -125,7 +131,6 @@ class TestComposeCaInject(unittest.TestCase):
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
calls.append(cmd)
# container image inspect <image>: first missing, second exists
if cmd[:3] == ["docker", "image", "inspect"]:
inspect_calls = [
c for c in calls if c[:3] == ["docker", "image", "inspect"]
@@ -134,11 +139,9 @@ class TestComposeCaInject(unittest.TestCase):
return 1, "", "no such image"
return 0, json.dumps([{"Config": {"Entrypoint": [], "Cmd": []}}]), ""
# compose ... build app
if cmd[:2] == ["docker", "compose"] and cmd[-2:] == ["build", "app"]:
return 0, "built", ""
# anything else
return 1, "", "unexpected"
base_cmd = ["docker", "compose", "-p", "p", "-f", "compose.yml"]
@@ -178,7 +181,6 @@ class TestComposeCaInject(unittest.TestCase):
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
calls.append(cmd)
# container image inspect: first missing, second exists
if cmd[:3] == ["docker", "image", "inspect"]:
inspect_calls = [
c for c in calls if c[:3] == ["docker", "image", "inspect"]
@@ -187,11 +189,9 @@ class TestComposeCaInject(unittest.TestCase):
return 1, "", "no such image"
return 0, json.dumps([{"Config": {"Entrypoint": [], "Cmd": []}}]), ""
# build builder (app)
if cmd[:2] == ["docker", "compose"] and cmd[-2:] == ["build", "app"]:
return 0, "built", ""
# if pull(worker) happens, that's wrong for this test; still return success
if cmd[:2] == ["docker", "compose"] and cmd[-2:] == ["pull", "worker"]:
return 0, "pulled", ""
@@ -225,7 +225,6 @@ class TestComposeCaInject(unittest.TestCase):
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
calls.append(cmd)
# image inspect: first missing, second exists after pull
if cmd[:3] == ["docker", "image", "inspect"]:
inspect_calls = [
c for c in calls if c[:3] == ["docker", "image", "inspect"]
@@ -234,7 +233,6 @@ class TestComposeCaInject(unittest.TestCase):
return 1, "", "no such image"
return 0, json.dumps([{"Config": {"Entrypoint": [], "Cmd": []}}]), ""
# pull succeeds
if cmd[:2] == ["docker", "compose"] and cmd[-2:] == ["pull", "worker"]:
return 0, "pulled", ""
@@ -272,7 +270,6 @@ class TestComposeCaInject(unittest.TestCase):
_read_text.return_value = "services:\n app:\n image: myimage:latest\n"
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
# compose ... config
if (
len(cmd) >= 3
and cmd[0:2] == ["docker", "compose"]
@@ -281,7 +278,6 @@ class TestComposeCaInject(unittest.TestCase):
yml = "services:\n app:\n image: myimage:latest\n"
return 0, yml, ""
# container image inspect <image>
if cmd[:3] == ["docker", "image", "inspect"]:
json_out = json.dumps(
[{"Config": {"Entrypoint": ["/entry"], "Cmd": ["run"]}}]
@@ -305,6 +301,10 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
"--ca-container",
CA_CERT_CONTAINER,
"--wrapper-container",
CA_WRAPPER_CONTAINER,
"--trust-name",
"infinito.local",
]
@@ -314,7 +314,6 @@ class TestComposeCaInject(unittest.TestCase):
self.assertEqual(rc, 0)
self.assertTrue(_write_text.called)
# Optional sanity: ensure the written YAML contains CA_TRUST_NAME and trust-name value
args, _kwargs = _write_text.call_args
written = args[1] if len(args) > 1 else ""
self.assertIn("CA_TRUST_NAME", written)
@@ -357,6 +356,10 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
"--ca-container",
CA_CERT_CONTAINER,
"--wrapper-container",
CA_WRAPPER_CONTAINER,
"--trust-name",
"infinito.local",
]
@@ -387,7 +390,10 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
# missing: --trust-name
"--ca-container",
CA_CERT_CONTAINER,
"--wrapper-container",
CA_WRAPPER_CONTAINER,
]
with patch("sys.argv", argv), self.assertRaises(SystemExit):
self.m.main()
@@ -445,11 +451,13 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
out = doc["services"]["svc"]
self.assertEqual(out.get("entrypoint"), ["/tmp/infinito/bin/with-ca-trust.sh"])
self.assertEqual(out.get("entrypoint"), [CA_WRAPPER_CONTAINER])
self.assertEqual(out.get("command"), ["/entry", "run"])
def test_render_override_wraps_when_inspect_ambiguous(self):
@@ -474,11 +482,13 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
out = doc["services"]["svc"]
self.assertEqual(out.get("entrypoint"), ["/tmp/infinito/bin/with-ca-trust.sh"])
self.assertEqual(out.get("entrypoint"), [CA_WRAPPER_CONTAINER])
self.assertEqual(out.get("command"), ["/svc-entry", "svc-run"])
def test_render_override_env_only_when_probe_clean_negative(self):
@@ -497,6 +507,8 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
@@ -548,6 +560,8 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
@@ -555,24 +569,21 @@ class TestComposeCaInject(unittest.TestCase):
self.assertIn("svc", doc["services"])
out = doc["services"]["svc"]
# Always mounts both CA and wrapper
self.assertIn("volumes", out)
self.assertIn("/host/ca.crt:/tmp/infinito/ca/root-ca.crt:ro", out["volumes"])
self.assertIn(f"/host/ca.crt:{CA_CERT_CONTAINER}:ro", out["volumes"])
self.assertIn(
"/host/with-ca-trust.sh:/tmp/infinito/bin/with-ca-trust.sh:ro",
f"/host/with-ca-trust.sh:{CA_WRAPPER_CONTAINER}:ro",
out["volumes"],
)
# Always sets env vars
env = out.get("environment", {})
self.assertEqual(env.get("CA_TRUST_CERT"), "/tmp/infinito/ca/root-ca.crt")
self.assertEqual(env.get("CA_TRUST_CERT"), CA_CERT_CONTAINER)
self.assertEqual(env.get("CA_TRUST_NAME"), "infinito.local")
self.assertEqual(env.get("SSL_CERT_FILE"), "/tmp/infinito/ca/root-ca.crt")
self.assertEqual(env.get("REQUESTS_CA_BUNDLE"), "/tmp/infinito/ca/root-ca.crt")
self.assertEqual(env.get("CURL_CA_BUNDLE"), "/tmp/infinito/ca/root-ca.crt")
self.assertEqual(env.get("NODE_EXTRA_CA_CERTS"), "/tmp/infinito/ca/root-ca.crt")
self.assertEqual(env.get("SSL_CERT_FILE"), CA_CERT_CONTAINER)
self.assertEqual(env.get("REQUESTS_CA_BUNDLE"), CA_CERT_CONTAINER)
self.assertEqual(env.get("CURL_CA_BUNDLE"), CA_CERT_CONTAINER)
self.assertEqual(env.get("NODE_EXTRA_CA_CERTS"), CA_CERT_CONTAINER)
# For distroless-like images: do NOT override entrypoint/command
self.assertNotIn("entrypoint", out)
self.assertNotIn("command", out)
@@ -596,11 +607,13 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
out = doc["services"]["svc"]
self.assertEqual(out.get("entrypoint"), ["/tmp/infinito/bin/with-ca-trust.sh"])
self.assertEqual(out.get("entrypoint"), [CA_WRAPPER_CONTAINER])
self.assertEqual(out.get("command"), ["/entry", "run", "--flag"])
def test_render_override_uses_service_entrypoint_command_over_image(self):
@@ -629,6 +642,8 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
@@ -680,11 +695,13 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
out = doc["services"]["svc"]
self.assertEqual(out.get("entrypoint"), ["/tmp/infinito/bin/with-ca-trust.sh"])
self.assertEqual(out.get("entrypoint"), [CA_WRAPPER_CONTAINER])
self.assertEqual(out.get("command")[:2], ["/bin/sh", "-c"])
self.assertEqual(
out.get("command")[2],
@@ -714,6 +731,8 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
@@ -724,7 +743,6 @@ class TestComposeCaInject(unittest.TestCase):
"""
main(): when /bin/sh exists for the image, the written override should include entrypoint wrapper.
"""
# Compose file content used for profile discovery only
with (
patch.object(Path, "exists", autospec=True, return_value=True),
patch.object(Path, "is_dir", autospec=True, return_value=True),
@@ -739,7 +757,6 @@ class TestComposeCaInject(unittest.TestCase):
):
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
# compose ... config
if (
len(cmd) >= 3
and cmd[0:2] == ["docker", "compose"]
@@ -748,14 +765,12 @@ class TestComposeCaInject(unittest.TestCase):
yml = "services:\n app:\n image: myimage:latest\n"
return 0, yml, ""
# container image inspect <image>
if cmd[:3] == ["docker", "image", "inspect"]:
json_out = json.dumps(
[{"Config": {"Entrypoint": ["/entry"], "Cmd": ["run"]}}]
)
return 0, json_out, ""
# container run --entrypoint /bin/sh ... -c exit 0 (bin/sh probe)
if cmd[:6] == [
"docker",
"run",
@@ -783,6 +798,10 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
"--ca-container",
CA_CERT_CONTAINER,
"--wrapper-container",
CA_WRAPPER_CONTAINER,
"--trust-name",
"infinito.local",
]
@@ -799,7 +818,7 @@ class TestComposeCaInject(unittest.TestCase):
self.assertIn("app", parsed["services"])
self.assertEqual(
parsed["services"]["app"].get("entrypoint"),
["/tmp/infinito/bin/with-ca-trust.sh"],
[CA_WRAPPER_CONTAINER],
)
def test_escape_compose_vars_replaces_single_dollar(self):
@@ -844,11 +863,13 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
out = doc["services"]["svc"]
self.assertEqual(out.get("entrypoint"), ["/tmp/infinito/bin/with-ca-trust.sh"])
self.assertEqual(out.get("entrypoint"), [CA_WRAPPER_CONTAINER])
self.assertEqual(
out.get("command"),
["sh", "-lc", 'exec "$$CHESS_ENTRYPOINT_INT"'],
@@ -875,15 +896,15 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
wrap=False,
)
out = doc["services"]["svc"]
self.assertIn("volumes", out)
self.assertEqual(
out["environment"].get("SSL_CERT_FILE"), "/tmp/infinito/ca/root-ca.crt"
)
self.assertEqual(out["environment"].get("SSL_CERT_FILE"), CA_CERT_CONTAINER)
self.assertNotIn("entrypoint", out)
self.assertNotIn("command", out)
p_ensure.assert_not_called()
@@ -917,6 +938,8 @@ class TestComposeCaInject(unittest.TestCase):
env={},
ca_host="/host/ca.crt",
wrapper_host="/host/with-ca-trust.sh",
ca_container=CA_CERT_CONTAINER,
wrapper_container=CA_WRAPPER_CONTAINER,
trust_name="infinito.local",
)
@@ -945,7 +968,6 @@ class TestComposeCaInject(unittest.TestCase):
):
def fake_run(cmd, *, cwd, env, timeout=None, capture=True):
# compose ... config
if (
len(cmd) >= 3
and cmd[0:2] == ["docker", "compose"]
@@ -954,7 +976,6 @@ class TestComposeCaInject(unittest.TestCase):
yml = "services:\n app:\n image: myimage:latest\n"
return 0, yml, ""
# container image inspect <image> -> Cmd contains '$'
if cmd[:3] == ["docker", "image", "inspect"]:
json_out = json.dumps(
[
@@ -972,7 +993,6 @@ class TestComposeCaInject(unittest.TestCase):
)
return 0, json_out, ""
# container run --entrypoint /bin/sh ... (bin/sh probe)
if cmd[:6] == [
"docker",
"run",
@@ -1000,6 +1020,10 @@ class TestComposeCaInject(unittest.TestCase):
"/etc/infinito/ca/root-ca.crt",
"--wrapper-host",
"/etc/infinito/bin/with-ca-trust.sh",
"--ca-container",
CA_CERT_CONTAINER,
"--wrapper-container",
CA_WRAPPER_CONTAINER,
"--trust-name",
"infinito.local",
]
@@ -1015,7 +1039,6 @@ class TestComposeCaInject(unittest.TestCase):
self.assertIn("services", parsed)
self.assertIn("app", parsed["services"])
# This is the crucial assertion: '$' must be doubled in YAML.
self.assertEqual(
parsed["services"]["app"].get("command"),
["sh", "-lc", 'exec "$$CHESS_ENTRYPOINT_INT"'],
tests/unit/roles/sys-svc-compose/lookup_plugins/test_compose_ca_inject_cmd.py
+13
View File
@@ -7,9 +7,16 @@ from unittest.mock import patch
from ansible.errors import AnsibleError
from plugins.filter.ca_trust_paths import ca_cert_host
from utils.cache.yaml import load_yaml
from . import PROJECT_ROOT
_CA_TRUST = load_yaml(str(PROJECT_ROOT / "group_vars" / "all" / "02_tls.yml"))[
"CA_TRUST"
]
CA_CERT_CONTAINER = _CA_TRUST["inject_cert_container"]
CA_WRAPPER_CONTAINER = _CA_TRUST["inject_wrapper_container"]
def _load_module(rel_path: str, name: str):
path = PROJECT_ROOT / rel_path
@@ -94,6 +101,8 @@ class ComposeCaInjectCmdLookupTests(unittest.TestCase):
"inject_script": "/usr/local/bin/compose_ca.py",
"cert_host": ca_cert_host("infinito.nexus"),
"wrapper_host": "/usr/local/bin/with-ca-trust.sh",
"inject_cert_container": CA_CERT_CONTAINER,
"inject_wrapper_container": CA_WRAPPER_CONTAINER,
"trust_name": "infinito-root-ca",
}
}
@@ -166,6 +175,8 @@ class ComposeCaInjectCmdLookupTests(unittest.TestCase):
"inject_script": "/usr/local/bin/compose_ca.py",
"cert_host": ca_cert_host("infinito.nexus"),
"wrapper_host": "/usr/local/bin/with-ca-trust.sh",
"inject_cert_container": CA_CERT_CONTAINER,
"inject_wrapper_container": CA_WRAPPER_CONTAINER,
"trust_name": "infinito-root-ca",
}
}
@@ -246,6 +257,8 @@ class ComposeCaInjectCmdLookupTests(unittest.TestCase):
"inject_script": "/usr/local/bin/compose_ca.py",
"cert_host": ca_cert_host("infinito.nexus"),
"wrapper_host": "/usr/local/bin/with-ca-trust.sh",
"inject_cert_container": CA_CERT_CONTAINER,
"inject_wrapper_container": CA_WRAPPER_CONTAINER,
"trust_name": "infinito-root-ca",
}
}
tests/unit/roles/sys-svc-container/files/test_container.py
+56 -15
View File
@@ -8,16 +8,20 @@ import tempfile
import unittest
from pathlib import Path
from . import PROJECT_ROOT
from utils.cache.yaml import load_yaml
# ---------------------------------------------------------------------------
# Manual import of roles/sys-svc-container/files/container.py
# ---------------------------------------------------------------------------
from . import PROJECT_ROOT
THIS_FILE = Path(__file__).resolve()
CONTAINER_PY = PROJECT_ROOT / "roles" / "sys-svc-container" / "files" / "container.py"
_CA_TRUST = load_yaml(str(PROJECT_ROOT / "group_vars" / "all" / "02_tls.yml"))[
"CA_TRUST"
]
CA_CERT_CONTAINER = _CA_TRUST["inject_cert_container"]
CA_WRAPPER_CONTAINER = _CA_TRUST["inject_wrapper_container"]
if not CONTAINER_PY.exists():
raise FileNotFoundError(f"container.py not found at {CONTAINER_PY}")
@@ -154,7 +158,6 @@ class TestContainerWrapper(unittest.TestCase):
def fake_pull(image: str) -> None:
calls.append(("pull", image))
# after pull, inspect should succeed
container.inspect_image_entrypoint = inspect_ok
orig_pull = container.docker_pull
@@ -204,7 +207,7 @@ class TestContainerWrapper(unittest.TestCase):
def test_require_ca_env_soft_missing_env_returns_none_and_warns(self):
"""
If any CA env var is missing, require_ca_env_soft() should:
If the host CA env trio is missing, require_ca_env_soft() should:
- return None
- emit a warning on stderr
"""
@@ -213,6 +216,8 @@ class TestContainerWrapper(unittest.TestCase):
os.environ.pop("CA_TRUST_CERT_HOST", None)
os.environ.pop("CA_TRUST_WRAPPER_HOST", None)
os.environ.pop("CA_TRUST_NAME", None)
os.environ.pop("CA_TRUST_CERT_CONTAINER", None)
os.environ.pop("CA_TRUST_WRAPPER_CONTAINER", None)
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
@@ -226,6 +231,39 @@ class TestContainerWrapper(unittest.TestCase):
os.environ.clear()
os.environ.update(old_env)
def test_require_ca_env_soft_half_migrated_dies_loudly(self):
"""
Host trio present but container paths absent = half-migrated caller:
require_ca_env_soft() must fail hard instead of silently stripping
CA trust from a host that clearly wants it.
"""
old_env = dict(os.environ)
try:
with tempfile.TemporaryDirectory() as td:
td_path = Path(td)
ca_file = td_path / "root-ca.crt"
wrapper_file = td_path / "with-ca-trust.sh"
ca_file.write_text("dummy-ca\n", encoding="utf-8")
wrapper_file.write_text("#!/bin/sh\necho ok\n", encoding="utf-8")
os.environ["CA_TRUST_CERT_HOST"] = str(ca_file)
os.environ["CA_TRUST_WRAPPER_HOST"] = str(wrapper_file)
os.environ["CA_TRUST_NAME"] = "test-ca"
os.environ.pop("CA_TRUST_CERT_CONTAINER", None)
os.environ.pop("CA_TRUST_WRAPPER_CONTAINER", None)
buf = io.StringIO()
with (
contextlib.redirect_stderr(buf),
self.assertRaises(SystemExit),
):
container.require_ca_env_soft()
self.assertIn("CA injection misconfigured", buf.getvalue())
finally:
os.environ.clear()
os.environ.update(old_env)
def test_require_ca_env_soft_missing_files_returns_none_and_warns(self):
"""
If env vars exist but point to non-existing files, require_ca_env_soft() should:
@@ -237,6 +275,8 @@ class TestContainerWrapper(unittest.TestCase):
os.environ["CA_TRUST_CERT_HOST"] = "/does/not/exist/ca.crt"
os.environ["CA_TRUST_WRAPPER_HOST"] = "/does/not/exist/with-ca-trust.sh"
os.environ["CA_TRUST_NAME"] = "test-ca"
os.environ["CA_TRUST_CERT_CONTAINER"] = CA_CERT_CONTAINER
os.environ["CA_TRUST_WRAPPER_CONTAINER"] = CA_WRAPPER_CONTAINER
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
@@ -267,6 +307,8 @@ class TestContainerWrapper(unittest.TestCase):
os.environ["CA_TRUST_CERT_HOST"] = str(ca_file)
os.environ["CA_TRUST_WRAPPER_HOST"] = str(wrapper_file)
os.environ["CA_TRUST_NAME"] = "test-ca"
os.environ["CA_TRUST_CERT_CONTAINER"] = CA_CERT_CONTAINER
os.environ["CA_TRUST_WRAPPER_CONTAINER"] = CA_WRAPPER_CONTAINER
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
@@ -274,11 +316,12 @@ class TestContainerWrapper(unittest.TestCase):
self.assertIsNotNone(res)
assert res is not None
ca_host, wrapper_host, trust_name = res
ca_host, wrapper_host, trust_name, ca_container, wrapper_container = res
self.assertEqual(ca_host, str(ca_file))
self.assertEqual(wrapper_host, str(wrapper_file))
self.assertEqual(trust_name, "test-ca")
# should not warn in the OK case
self.assertEqual(ca_container, CA_CERT_CONTAINER)
self.assertEqual(wrapper_container, CA_WRAPPER_CONTAINER)
self.assertEqual(buf.getvalue().strip(), "")
finally:
os.environ.clear()
@@ -317,7 +360,7 @@ class TestContainerWrapper(unittest.TestCase):
calls[0][1],
["docker", "run", *argv],
)
self.assertTrue(calls[0][2]) # debug=True
self.assertTrue(calls[0][2])
finally:
container.exec_docker = orig_exec_docker
os.environ.clear()
@@ -336,7 +379,7 @@ class TestContainerWrapper(unittest.TestCase):
return 0
def fake_try_inspect_no_ep(image, pull_policy):
return [] # no entrypoint
return []
orig_exec_docker = container.exec_docker
orig_try_inspect = container.try_inspect_entrypoint_with_pull
@@ -351,6 +394,8 @@ class TestContainerWrapper(unittest.TestCase):
os.environ["CA_TRUST_CERT_HOST"] = str(ca_file)
os.environ["CA_TRUST_WRAPPER_HOST"] = str(wrapper_file)
os.environ["CA_TRUST_NAME"] = "test-ca"
os.environ["CA_TRUST_CERT_CONTAINER"] = CA_CERT_CONTAINER
os.environ["CA_TRUST_WRAPPER_CONTAINER"] = CA_WRAPPER_CONTAINER
container.exec_docker = fake_exec_docker
container.try_inspect_entrypoint_with_pull = fake_try_inspect_no_ep
@@ -361,16 +406,12 @@ class TestContainerWrapper(unittest.TestCase):
self.assertEqual(rc, 0)
self.assertEqual(len(calls), 1)
cmd = calls[0][1]
# Must be a docker run command
self.assertEqual(cmd[0], "docker")
self.assertEqual(cmd[1], "run")
# Must contain NODE_EXTRA_CA_CERTS injection
cmd_str = " ".join(cmd)
self.assertIn("NODE_EXTRA_CA_CERTS", cmd_str)
self.assertIn("/tmp/infinito/ca/root-ca.crt", cmd_str)
# Must NOT use --entrypoint (no wrapper)
self.assertIn(CA_CERT_CONTAINER, cmd_str)
self.assertNotIn("--entrypoint", cmd_str)
# Image and original command must appear
self.assertIn("playwright:v1.0", cmd_str)
self.assertIn("npm install", cmd_str)
finally:
tests/unit/roles/sys-svc-container/files/test_container_exec_timeout.py
+96
View File
@@ -0,0 +1,96 @@
from __future__ import annotations
import importlib.util
import unittest
from unittest.mock import patch
from . import PROJECT_ROOT
CONTAINER_PY = PROJECT_ROOT / "roles" / "sys-svc-container" / "files" / "container.py"
spec = importlib.util.spec_from_file_location("container_exec_timeout", CONTAINER_PY)
container = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(container)
class TestParseExecTimeoutSeconds(unittest.TestCase):
def test_units(self):
self.assertEqual(container.parse_exec_timeout_seconds("45s"), 45)
self.assertEqual(container.parse_exec_timeout_seconds("60min"), 3600)
self.assertEqual(container.parse_exec_timeout_seconds("2h"), 7200)
self.assertEqual(container.parse_exec_timeout_seconds("7d"), 604800)
def test_bare_number_is_seconds(self):
self.assertEqual(container.parse_exec_timeout_seconds("90"), 90)
def test_invalid_raises(self):
with self.assertRaises(ValueError):
container.parse_exec_timeout_seconds("soon")
class TestExecTimeoutPrefix(unittest.TestCase):
def test_unset_means_no_prefix(self):
with patch.dict(container.os.environ, {}, clear=True):
self.assertEqual(container.exec_timeout_prefix(), [])
def test_zero_means_no_prefix(self):
env = {"CONTAINER_EXEC_TIMEOUT": "0"}
with patch.dict(container.os.environ, env, clear=True):
self.assertEqual(container.exec_timeout_prefix(), [])
def test_duration_builds_kill_prefix(self):
env = {"CONTAINER_EXEC_TIMEOUT": "60min"}
with patch.dict(container.os.environ, env, clear=True):
self.assertEqual(
container.exec_timeout_prefix(),
["timeout", "--kill-after=15", "3600"],
)
def test_week_default_builds_prefix(self):
env = {"CONTAINER_EXEC_TIMEOUT": "7d"}
with patch.dict(container.os.environ, env, clear=True):
self.assertEqual(
container.exec_timeout_prefix(),
["timeout", "--kill-after=15", "604800"],
)
class TestMainExecDispatch(unittest.TestCase):
def _run_main(self, argv, env):
calls = []
def _capture(cmd, debug):
calls.append(cmd)
return 0
with (
patch.object(container.sys, "argv", ["container", *argv]),
patch.dict(container.os.environ, env, clear=True),
patch.object(container, "exec_docker", side_effect=_capture),
):
rc = container.main()
self.assertEqual(rc, 0)
self.assertEqual(len(calls), 1)
return calls[0]
def test_exec_gets_timeout_prefix(self):
cmd = self._run_main(
["exec", "-i", "c1", "ls"], {"CONTAINER_EXEC_TIMEOUT": "60min"}
)
self.assertEqual(
cmd,
["timeout", "--kill-after=15", "3600", "docker", "exec", "-i", "c1", "ls"],
)
def test_exec_without_env_is_plain_passthrough(self):
cmd = self._run_main(["exec", "c1", "ls"], {})
self.assertEqual(cmd, ["docker", "exec", "c1", "ls"])
def test_other_subcommands_never_get_the_prefix(self):
cmd = self._run_main(["ps", "-a"], {"CONTAINER_EXEC_TIMEOUT": "60min"})
self.assertEqual(cmd, ["docker", "ps", "-a"])
if __name__ == "__main__":
unittest.main()