Compare commits

...

2 Commits

Author SHA1 Message Date
didericis-codex 95cbd0e1c8 fix: enforce cleanup and secret integrity
test / image-input-builds (pull_request) Failing after 11m32s
test / unit (pull_request) Has started running
test / coverage (pull_request) Blocked by required conditions
test / integration-docker (pull_request) Blocked by required conditions
tracker-policy-pr / check-pr (pull_request) Failing after 12m21s
2026-07-27 04:46:00 +00:00
didericis-codex f242733d15 fix(cleanup): use authoritative resource identities
test / integration-docker (pull_request) Has been cancelled
test / image-input-builds (pull_request) Successful in 53s
test / unit (pull_request) Successful in 59s
test / coverage (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 12m23s
2026-07-27 04:46:00 +00:00
24 changed files with 391 additions and 152 deletions
+4
View File
@@ -172,6 +172,10 @@ class BottleCleanupPlan(ABC):
"""True iff there is nothing to clean up; the CLI uses this to """True iff there is nothing to clean up; the CLI uses this to
short-circuit before showing the y/N.""" short-circuit before showing the y/N."""
@abstractmethod
def intersect(self, current: "BottleCleanupPlan") -> "BottleCleanupPlan":
"""Resources both displayed to the operator and currently removable."""
@dataclass(frozen=True) @dataclass(frozen=True)
class ExecResult: class ExecResult:
+48
View File
@@ -0,0 +1,48 @@
"""Shared destructive-cleanup execution and failure accounting."""
from __future__ import annotations
import shutil
import subprocess
from collections.abc import Sequence
from pathlib import Path
class CleanupError(RuntimeError):
"""One or more approved cleanup mutations did not complete."""
class CleanupFailures:
"""Attempt every approved mutation, then fail with complete diagnostics."""
def __init__(self) -> None:
self._messages: list[str] = []
def run(self, argv: Sequence[str], description: str) -> None:
try:
result = subprocess.run(
list(argv), capture_output=True, text=True, check=False,
)
except OSError as exc:
self._messages.append(f"{description}: {exc}")
return
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip()
self._messages.append(
f"{description}: {detail or f'exit {result.returncode}'}"
)
def remove_tree(self, path: Path, description: str) -> None:
try:
shutil.rmtree(path)
except FileNotFoundError:
return
except OSError as exc:
self._messages.append(f"{description}: {exc}")
def raise_if_any(self) -> None:
if self._messages:
raise CleanupError("; ".join(self._messages))
__all__ = ["CleanupError", "CleanupFailures"]
@@ -46,6 +46,22 @@ class DockerBottleCleanupPlan(BottleCleanupPlan):
and not self.orphan_state_dirs and not self.orphan_state_dirs
) )
def intersect(self, current: BottleCleanupPlan) -> "DockerBottleCleanupPlan":
if not isinstance(current, DockerBottleCleanupPlan):
raise TypeError("cleanup plans must have the same backend type")
return DockerBottleCleanupPlan(
projects=tuple(x for x in self.projects if x in current.projects),
stray_containers=tuple(
x for x in self.stray_containers if x in current.stray_containers
),
stray_networks=tuple(
x for x in self.stray_networks if x in current.stray_networks
),
orphan_state_dirs=tuple(
x for x in self.orphan_state_dirs if x in current.orphan_state_dirs
),
)
def print(self) -> None: def print(self) -> None:
print(file=sys.stderr) print(file=sys.stderr)
for name in self.projects: for name in self.projects:
+38 -38
View File
@@ -23,11 +23,12 @@ Active-agent enumeration lives in `backend/docker/enumerate.py`.
from __future__ import annotations from __future__ import annotations
import shutil
import subprocess import subprocess
from ...paths import bot_bottle_root from ...paths import bot_bottle_root
from ...log import info, warn from ...log import info
from .. import EnumerationError
from ..cleanup_control import CleanupFailures
from . import util as docker_mod from . import util as docker_mod
from .bottle_cleanup_plan import DockerBottleCleanupPlan from .bottle_cleanup_plan import DockerBottleCleanupPlan
from ...bottle_state import bottle_state_dir, is_preserved from ...bottle_state import bottle_state_dir, is_preserved
@@ -36,15 +37,17 @@ from .compose import COMPOSE_PROJECT_PREFIX, list_compose_projects
def _list_prefixed_containers() -> list[str]: def _list_prefixed_containers() -> list[str]:
"""All bot-bottle-prefixed containers, running or stopped.""" """All bot-bottle-prefixed containers, running or stopped."""
result = subprocess.run( try:
["docker", "ps", "-a", result = subprocess.run(
"--filter", f"name=^{COMPOSE_PROJECT_PREFIX}", ["docker", "ps", "-a",
"--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"], "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}",
capture_output=True, text=True, check=False, "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
) capture_output=True, text=True, check=False,
)
except OSError as exc:
raise EnumerationError(f"docker ps failed: {exc}") from exc
if result.returncode != 0: if result.returncode != 0:
warn(f"docker ps failed: {result.stderr.strip()}") raise EnumerationError(f"docker ps failed: {result.stderr.strip()}")
return []
out: list[str] = [] out: list[str] = []
for line in (result.stdout or "").splitlines(): for line in (result.stdout or "").splitlines():
if not line: if not line:
@@ -63,15 +66,19 @@ def _list_prefixed_networks() -> list[str]:
to a compose project. Compose-managed networks have a to a compose project. Compose-managed networks have a
`com.docker.compose.project` label; bare ones (from pre-compose `com.docker.compose.project` label; bare ones (from pre-compose
code paths) don't.""" code paths) don't."""
result = subprocess.run( try:
["docker", "network", "ls", result = subprocess.run(
"--filter", f"name={COMPOSE_PROJECT_PREFIX}", ["docker", "network", "ls",
"--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"], "--filter", f"name={COMPOSE_PROJECT_PREFIX}",
capture_output=True, text=True, check=False, "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
) capture_output=True, text=True, check=False,
)
except OSError as exc:
raise EnumerationError(f"docker network ls failed: {exc}") from exc
if result.returncode != 0: if result.returncode != 0:
warn(f"docker network ls failed: {result.stderr.strip()}") raise EnumerationError(
return [] f"docker network ls failed: {result.stderr.strip()}"
)
out: list[str] = [] out: list[str] = []
for line in (result.stdout or "").splitlines(): for line in (result.stdout or "").splitlines():
if not line: if not line:
@@ -120,7 +127,10 @@ def prepare_cleanup() -> DockerBottleCleanupPlan:
`enumerate_active_agents()` so the orphan-state-dir bucket `enumerate_active_agents()` so the orphan-state-dir bucket
doesn't include slugs whose non-docker bottle is still up.""" doesn't include slugs whose non-docker bottle is still up."""
docker_mod.require_docker() docker_mod.require_docker()
projects = list_compose_projects() projects = list_compose_projects(
warn_on_error=False,
raise_on_error=True,
)
project_set = set(projects) project_set = set(projects)
# Late import to avoid a circular at module-load time — # Late import to avoid a circular at module-load time —
# the backend package's __init__ imports this module. # the backend package's __init__ imports this module.
@@ -140,40 +150,30 @@ def cleanup(plan: DockerBottleCleanupPlan) -> None:
"""Remove everything in the plan. Projects first (whose `compose """Remove everything in the plan. Projects first (whose `compose
down` reaps their containers + networks atomically), then stray down` reaps their containers + networks atomically), then stray
legacy resources, then orphan state dirs.""" legacy resources, then orphan state dirs."""
failures = CleanupFailures()
for project in plan.projects: for project in plan.projects:
info(f"docker compose down ({project})") info(f"docker compose down ({project})")
result = subprocess.run( failures.run(
["docker", "compose", "-p", project, "down", "--volumes"], ["docker", "compose", "-p", project, "down", "--volumes"],
capture_output=True, text=True, check=False, f"docker compose down failed for {project}",
) )
if result.returncode != 0:
warn(
f"compose down failed for {project}: "
f"{result.stderr.strip()}"
)
for name in plan.stray_containers: for name in plan.stray_containers:
info(f"removing stray container {name}") info(f"removing stray container {name}")
subprocess.run( failures.run(
["docker", "rm", "-f", name], ["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL, f"removing stray container {name}",
stderr=subprocess.DEVNULL,
check=False,
) )
for name in plan.stray_networks: for name in plan.stray_networks:
info(f"removing stray network {name}") info(f"removing stray network {name}")
subprocess.run( failures.run(
["docker", "network", "rm", name], ["docker", "network", "rm", name],
stdout=subprocess.DEVNULL, f"removing stray network {name}",
stderr=subprocess.DEVNULL,
check=False,
) )
for identity in plan.orphan_state_dirs: for identity in plan.orphan_state_dirs:
path = bottle_state_dir(identity) path = bottle_state_dir(identity)
info(f"removing orphan state dir {path}") info(f"removing orphan state dir {path}")
try: failures.remove_tree(path, f"removing orphan state dir {path}")
shutil.rmtree(path, ignore_errors=True) failures.raise_if_any()
except OSError as e:
warn(f"failed to remove {path}: {e}")
@@ -27,3 +27,11 @@ class FirecrackerBottleCleanupPlan(BottleCleanupPlan):
@property @property
def empty(self) -> bool: def empty(self) -> bool:
return not (self.vm_pids or self.run_dirs) return not (self.vm_pids or self.run_dirs)
def intersect(self, current: BottleCleanupPlan) -> "FirecrackerBottleCleanupPlan":
if not isinstance(current, FirecrackerBottleCleanupPlan):
raise TypeError("cleanup plans must have the same backend type")
return FirecrackerBottleCleanupPlan(
vm_pids=tuple(x for x in self.vm_pids if x in current.vm_pids),
run_dirs=tuple(x for x in self.run_dirs if x in current.run_dirs),
)
+41 -19
View File
@@ -11,10 +11,9 @@ Reaps *orphans* only — resources with no live VM behind them:
— a VMM left lingering after its dir was removed. — a VMM left lingering after its dir was removed.
A run dir with a *live* firecracker process is a running bottle and is A run dir with a *live* firecracker process is a running bottle and is
left strictly alone: it is neither killed nor removed. (The backend's left strictly alone: it is neither killed nor removed. Active-agent
`enumerate_active` registry is still a stub — #354 — so a live process enumeration uses this same process snapshot, so cleanup and generic
is the only reliable "this bottle is in use" signal we have. Once the backend consumers agree about which bottles are running.
registry lands, registry-orphaned-but-running VMs can be reaped too.)
TAP slots free themselves (the flock drops when the launcher exits), so TAP slots free themselves (the flock drops when the launcher exits), so
there is nothing to reclaim there. there is nothing to reclaim there.
@@ -22,14 +21,15 @@ there is nothing to reclaim there.
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence
import os import os
import shutil
import signal import signal
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from ...log import info from ...log import info
from .. import EnumerationError from .. import EnumerationError
from ..cleanup_control import CleanupFailures
from . import lifecycle_lock, util from . import lifecycle_lock, util
from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan
@@ -38,7 +38,7 @@ def _run_root() -> Path:
return util.cache_dir() / "run" return util.cache_dir() / "run"
def _run_dir_of(cmd: str, run_root: Path) -> Path | None: def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None:
"""The bottle run dir a firecracker cmdline belongs to, or None. """The bottle run dir a firecracker cmdline belongs to, or None.
A bottle VM is launched with `--config-file <run_root>/<slug>/config.json`, A bottle VM is launched with `--config-file <run_root>/<slug>/config.json`,
@@ -46,15 +46,35 @@ def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
the run root. Anything else (a builder VM, the infra VM elsewhere) is the run root. Anything else (a builder VM, the infra VM elsewhere) is
not ours to reap here. not ours to reap here.
""" """
toks = cmd.split() for i, arg in enumerate(args):
for i, tok in enumerate(toks): if arg == "--config-file" and i + 1 < len(args):
if tok == "--config-file" and i + 1 < len(toks): parent = Path(args[i + 1]).parent
parent = Path(toks[i + 1]).parent
if parent.parent == run_root: if parent.parent == run_root:
return parent return parent
return None return None
def _decode_cmdline(raw: bytes) -> tuple[str, ...]:
"""Decode Linux's NUL-delimited argv without losing embedded spaces."""
return tuple(
value.decode(errors="surrogateescape")
for value in raw.split(b"\0") if value
)
def _process_args(pid: int) -> tuple[str, ...] | None:
"""Read one process's lossless argv, or None when it exited meanwhile."""
try:
raw = Path(f"/proc/{pid}/cmdline").read_bytes()
except FileNotFoundError:
return None
except OSError as exc:
raise EnumerationError(
f"could not inspect Firecracker pid {pid}: {exc}"
) from exc
return _decode_cmdline(raw)
def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]: def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
"""Inspect running firecracker VMs under ``run_root``. """Inspect running firecracker VMs under ``run_root``.
@@ -65,7 +85,7 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
""" """
try: try:
result = subprocess.run( result = subprocess.run(
["pgrep", "-a", "firecracker"], ["pgrep", "firecracker"],
capture_output=True, text=True, check=False, capture_output=True, text=True, check=False,
) )
except OSError as exc: except OSError as exc:
@@ -83,14 +103,14 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
live: set[str] = set() live: set[str] = set()
orphan_pids: list[int] = [] orphan_pids: list[int] = []
for line in result.stdout.splitlines(): for line in result.stdout.splitlines():
parts = line.split(None, 1)
if len(parts) != 2:
continue
try: try:
pid = int(parts[0]) pid = int(line.strip())
except ValueError: except ValueError:
continue continue
run_dir = _run_dir_of(parts[1], run_root) args = _process_args(pid)
if args is None:
continue
run_dir = _run_dir_of(args, run_root)
if run_dir is None: if run_dir is None:
continue continue
if run_dir.is_dir(): if run_dir.is_dir():
@@ -131,11 +151,13 @@ def cleanup(plan: FirecrackerBottleCleanupPlan) -> None:
fresh = prepare_cleanup() fresh = prepare_cleanup()
approved_pids = set(plan.vm_pids).intersection(fresh.vm_pids) approved_pids = set(plan.vm_pids).intersection(fresh.vm_pids)
approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs) approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs)
failures = CleanupFailures()
for pid in sorted(approved_pids): for pid in sorted(approved_pids):
_terminate_orphan(pid, _run_root()) _terminate_orphan(pid, _run_root())
for path in sorted(approved_dirs): for path in sorted(approved_dirs):
info(f"rm -rf {path}") info(f"rm -rf {path}")
shutil.rmtree(path, ignore_errors=True) failures.remove_tree(Path(path), f"removing Firecracker run dir {path}")
failures.raise_if_any()
def _terminate_orphan(pid: int, run_root: Path) -> None: def _terminate_orphan(pid: int, run_root: Path) -> None:
@@ -157,8 +179,8 @@ def _terminate_orphan(pid: int, run_root: Path) -> None:
raise EnumerationError( raise EnumerationError(
f"could not revalidate Firecracker pid {pid}: {exc}" f"could not revalidate Firecracker pid {pid}: {exc}"
) from exc ) from exc
command = raw.replace(b"\0", b" ").decode(errors="replace") args = _decode_cmdline(raw)
run_dir = _run_dir_of(command, run_root) run_dir = _run_dir_of(args, run_root)
if run_dir is None or run_dir.is_dir(): if run_dir is None or run_dir.is_dir():
return return
info(f"kill firecracker VM pid {pid}") info(f"kill firecracker VM pid {pid}")
@@ -41,6 +41,8 @@ from ... import resources
from ...log import die, info from ...log import die, info
from . import util from . import util
ARTIFACT_HTTP_TIMEOUT_SECONDS = 30.0
# Bump if the on-disk artifact *format* changes (compression, layout) so a new # Bump if the on-disk artifact *format* changes (compression, layout) so a new
# scheme can't collide with a cached/published artifact of the old one. # scheme can't collide with a cached/published artifact of the old one.
_ARTIFACT_FORMAT = "1" _ARTIFACT_FORMAT = "1"
@@ -164,7 +166,9 @@ def _download(url: str, dest: Path) -> None:
"""Stream `url` to `dest` (atomic via a `.part` sibling).""" """Stream `url` to `dest` (atomic via a `.part` sibling)."""
tmp = dest.with_suffix(dest.suffix + ".part") tmp = dest.with_suffix(dest.suffix + ".part")
try: try:
with urllib.request.urlopen(_open(url)) as resp, open(tmp, "wb") as out: with urllib.request.urlopen(
_open(url), timeout=ARTIFACT_HTTP_TIMEOUT_SECONDS,
) as resp, open(tmp, "wb") as out:
shutil.copyfileobj(resp, out, _CHUNK) shutil.copyfileobj(resp, out, _CHUNK)
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
tmp.unlink(missing_ok=True) tmp.unlink(missing_ok=True)
@@ -34,6 +34,7 @@ from pathlib import Path
from . import infra_artifact, infra_vm, util from . import infra_artifact, infra_vm, util
_CHUNK = 1 << 20 _CHUNK = 1 << 20
_REGISTRY_HTTP_TIMEOUT_SECONDS = 30.0
_GZ_NAME = "rootfs.ext4.gz" _GZ_NAME = "rootfs.ext4.gz"
_SHA_NAME = "rootfs.ext4.gz.sha256" _SHA_NAME = "rootfs.ext4.gz.sha256"
@@ -91,7 +92,9 @@ def _put(url: str, body: "bytes | Path", token: str) -> None:
req.add_header("Authorization", f"token {token}") req.add_header("Authorization", f"token {token}")
req.add_header("Content-Type", "application/octet-stream") req.add_header("Content-Type", "application/octet-stream")
try: try:
with urllib.request.urlopen(req) as resp: with urllib.request.urlopen(
req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
) as resp:
print(f" uploaded {url} (HTTP {resp.status})") print(f" uploaded {url} (HTTP {resp.status})")
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
if e.code == 409: if e.code == 409:
@@ -112,7 +115,9 @@ def _delete(url: str, token: str) -> None:
if token: if token:
req.add_header("Authorization", f"token {token}") req.add_header("Authorization", f"token {token}")
try: try:
with urllib.request.urlopen(req): with urllib.request.urlopen(
req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
):
pass pass
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
if e.code != 404: if e.code != 404:
@@ -151,7 +156,10 @@ def _try_download_published(role: str, role_dir: Path) -> str | None:
version = _role_version(role) version = _role_version(role)
sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role) sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role)
try: try:
with urllib.request.urlopen(infra_artifact._open(sha_url)): with urllib.request.urlopen(
infra_artifact._open(sha_url),
timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
):
pass pass
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
if e.code == 404: if e.code == 404:
@@ -195,7 +203,10 @@ def _publish_bundle(role: str, role_dir: Path, token: str) -> str:
# present, a re-publish is a no-op. Otherwise clear any partial upload left # present, a re-publish is a no-op. Otherwise clear any partial upload left
# by an interrupted prior attempt and upload the complete set. # by an interrupted prior attempt and upload the complete set.
try: try:
with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp: with urllib.request.urlopen(
infra_artifact._open(sha_url),
timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
) as resp:
remote_sha = resp.read().decode("utf-8").split()[0].strip().lower() remote_sha = resp.read().decode("utf-8").split()[0].strip().lower()
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
if e.code != 404: if e.code != 404:
@@ -25,3 +25,11 @@ class MacosContainerBottleCleanupPlan(BottleCleanupPlan):
@property @property
def empty(self) -> bool: def empty(self) -> bool:
return not self.containers and not self.networks return not self.containers and not self.networks
def intersect(self, current: BottleCleanupPlan) -> "MacosContainerBottleCleanupPlan":
if not isinstance(current, MacosContainerBottleCleanupPlan):
raise TypeError("cleanup plans must have the same backend type")
return MacosContainerBottleCleanupPlan(
containers=tuple(x for x in self.containers if x in current.containers),
networks=tuple(x for x in self.networks if x in current.networks),
)
@@ -5,6 +5,7 @@ from __future__ import annotations
import subprocess import subprocess
from .. import EnumerationError from .. import EnumerationError
from ..cleanup_control import CleanupFailures
from ...log import info from ...log import info
from . import util as container_mod from . import util as container_mod
from .bottle_cleanup_plan import MacosContainerBottleCleanupPlan from .bottle_cleanup_plan import MacosContainerBottleCleanupPlan
@@ -53,19 +54,17 @@ def prepare_cleanup() -> MacosContainerBottleCleanupPlan:
def cleanup(plan: MacosContainerBottleCleanupPlan) -> None: def cleanup(plan: MacosContainerBottleCleanupPlan) -> None:
failures = CleanupFailures()
for name in plan.containers: for name in plan.containers:
info(f"container delete --force {name}") info(f"container delete --force {name}")
subprocess.run( failures.run(
["container", "delete", "--force", name], ["container", "delete", "--force", name],
stdout=subprocess.DEVNULL, f"deleting container {name}",
stderr=subprocess.DEVNULL,
check=False,
) )
for name in plan.networks: for name in plan.networks:
info(f"container network delete {name}") info(f"container network delete {name}")
subprocess.run( failures.run(
["container", "network", "delete", name], ["container", "network", "delete", name],
stdout=subprocess.DEVNULL, f"deleting network {name}",
stderr=subprocess.DEVNULL,
check=False,
) )
failures.raise_if_any()
+12 -5
View File
@@ -22,6 +22,7 @@ from __future__ import annotations
import sys import sys
from ...backend import get_bottle_backend, has_backend, known_backend_names from ...backend import get_bottle_backend, has_backend, known_backend_names
from ...backend.cleanup_control import CleanupError
from ...log import info from ...log import info
from ...util import read_tty_line from ...util import read_tty_line
@@ -54,12 +55,18 @@ def cmd_cleanup(_argv: list[str]) -> int:
# Confirmation authorizes a fresh authoritative snapshot, not blind use of # Confirmation authorizes a fresh authoritative snapshot, not blind use of
# identities that may have changed while the operator reviewed the preview. # identities that may have changed while the operator reviewed the preview.
refreshed = [(name, backend, backend.prepare_cleanup()) failures: list[str] = []
for name, backend, _plan in prepared] for name, backend, displayed in prepared:
for name, backend, plan in refreshed: current = backend.prepare_cleanup()
if plan.empty: approved = displayed.intersect(current)
if approved.empty:
continue continue
backend.cleanup(plan) try:
backend.cleanup(approved)
except CleanupError as exc:
failures.append(f"{name}: {exc}")
if failures:
raise CleanupError("cleanup incomplete: " + "; ".join(failures))
info("cleanup: done") info("cleanup: done")
return 0 return 0
+13 -8
View File
@@ -203,14 +203,19 @@ class GitHttpHandler(BaseHTTPRequestHandler):
except BodyReadError as exc: except BodyReadError as exc:
self.send_error(exc.status, exc.message) self.send_error(exc.status, exc.message)
return return
proc = subprocess.run( try:
["git", "http-backend"], proc = subprocess.run(
input=body, ["git", "http-backend"],
env=env, input=body,
capture_output=True, env=env,
check=False, capture_output=True,
timeout=GIT_GATE_TIMEOUT_SECS, check=False,
) timeout=GIT_GATE_TIMEOUT_SECS,
)
except (OSError, subprocess.SubprocessError) as exc:
self.log_message("git http-backend unavailable: %s", exc)
self.send_error(503, "git backend unavailable")
return
self._write_cgi_response(proc.stdout) self._write_cgi_response(proc.stdout)
def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None: def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:
+1 -7
View File
@@ -366,7 +366,7 @@ class OrchestratorCore:
value with *env_var_secret*, and restores ``_tokens[bottle_id]``. value with *env_var_secret*, and restores ``_tokens[bottle_id]``.
Returns True on success, False when no stored secrets exist for this Returns True on success, False when no stored secrets exist for this
bottle or decryption fails (wrong key / corrupt data).""" bottle or decryption fails (wrong key / corrupt data)."""
from .store.secret_store import decrypt_value, encrypt_value, is_legacy_blob from .store.secret_store import decrypt_value
encrypted = self.registry.get_agent_secrets(bottle_id) encrypted = self.registry.get_agent_secrets(bottle_id)
if not encrypted: if not encrypted:
return False return False
@@ -377,12 +377,6 @@ class OrchestratorCore:
except ValueError: except ValueError:
return False return False
self._tokens[bottle_id] = decrypted self._tokens[bottle_id] = decrypted
if any(is_legacy_blob(value) for value in encrypted.values()):
migrated = {
key: encrypt_value(env_var_secret, value)
for key, value in decrypted.items()
}
self.registry.store_agent_secrets(bottle_id, migrated)
return True return True
# --- consolidated gateway ---------------------------------------------- # --- consolidated gateway ----------------------------------------------
@@ -129,6 +129,10 @@ _MIGRATIONS = TableMigrations(
# v5 — index for fast per-bottle lookups and bulk DELETE on teardown. # v5 — index for fast per-bottle lookups and bulk DELETE on teardown.
"CREATE INDEX IF NOT EXISTS idx_bottled_agent_secrets_id " "CREATE INDEX IF NOT EXISTS idx_bottled_agent_secrets_id "
"ON bottled_agent_secrets (bottled_agent_id, type)", "ON bottled_agent_secrets (bottled_agent_id, type)",
# v6 — unauthenticated legacy ciphertext must never be selected by
# attacker-controlled blob contents. Existing local agents are
# intentionally reprovisioned instead of retaining downgrade support.
"DELETE FROM bottled_agent_secrets",
], ],
) )
+5 -32
View File
@@ -18,9 +18,9 @@ value is encrypted independently. New output blobs are:
``version || nonce (16 bytes) || ciphertext || tag (32 bytes)`` ``version || nonce (16 bytes) || ciphertext || tag (32 bytes)``
encoded as URL-safe base64 (no padding). The version marker lets the reader encoded as URL-safe base64 (no padding). Unversioned legacy ciphertext is
accept legacy ``nonce || ciphertext`` rows long enough to rewrite them in the rejected; the registry migration clears those rows rather than allowing blob
authenticated format after a successful reprovision. contents to select an unauthenticated decoder.
keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big")) keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big"))
ciphertext_i = plaintext_i XOR keystream_block_i[:len(plaintext_i)] ciphertext_i = plaintext_i XOR keystream_block_i[:len(plaintext_i)]
@@ -84,44 +84,18 @@ def encrypt_value(secret_b64: str, plaintext: str) -> str:
return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode() return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode()
def is_legacy_blob(blob_b64: str) -> bool:
"""Whether *blob_b64* uses the pre-authentication storage format."""
try:
return not _b64dec(blob_b64).startswith(_VERSION)
except (ValueError, TypeError):
return False
def _decrypt_legacy(key: bytes, blob: bytes) -> str:
"""Read the original ``nonce || ciphertext`` format for migration only."""
if len(blob) < _NONCE_BYTES:
raise ValueError("ciphertext blob too short")
nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]
pt = bytearray()
# The legacy format used the byte offset as the PRF counter.
for i in range(0, len(ciphertext), _BLOCK):
chunk = ciphertext[i : i + _BLOCK]
ks = _keystream(key, nonce, i)[: len(chunk)]
pt.extend(c ^ k for c, k in zip(chunk, ks))
try:
return bytes(pt).decode()
except UnicodeDecodeError as exc:
raise ValueError(f"decryption produced non-UTF-8 output: {exc}") from exc
def decrypt_value(secret_b64: str, blob_b64: str) -> str: def decrypt_value(secret_b64: str, blob_b64: str) -> str:
"""Decrypt a blob produced by :func:`encrypt_value`. """Decrypt a blob produced by :func:`encrypt_value`.
Returns the original plaintext string. Raises ``ValueError`` for malformed Returns the original plaintext string. Raises ``ValueError`` for malformed
input, authentication failure, or a key mismatch. Legacy unauthenticated input, authentication failure, or a key mismatch."""
rows remain readable so callers can migrate them immediately."""
key = _b64dec(secret_b64) key = _b64dec(secret_b64)
try: try:
blob = _b64dec(blob_b64) blob = _b64dec(blob_b64)
except (ValueError, TypeError) as exc: except (ValueError, TypeError) as exc:
raise ValueError(f"invalid ciphertext blob: {exc}") from exc raise ValueError(f"invalid ciphertext blob: {exc}") from exc
if not blob.startswith(_VERSION): if not blob.startswith(_VERSION):
return _decrypt_legacy(key, blob) raise ValueError("unsupported ciphertext format")
minimum = len(_VERSION) + _NONCE_BYTES + _TAG_BYTES minimum = len(_VERSION) + _NONCE_BYTES + _TAG_BYTES
if len(blob) < minimum: if len(blob) < minimum:
raise ValueError("ciphertext blob too short") raise ValueError("ciphertext blob too short")
@@ -152,5 +126,4 @@ __all__ = [
"new_env_var_secret", "new_env_var_secret",
"encrypt_value", "encrypt_value",
"decrypt_value", "decrypt_value",
"is_legacy_blob",
] ]
+6 -2
View File
@@ -17,6 +17,7 @@ from bot_bottle.cli.commands import cleanup as cmd
def _make_backend(empty: bool = True): def _make_backend(empty: bool = True):
backend = MagicMock() backend = MagicMock()
plan = MagicMock(empty=empty) plan = MagicMock(empty=empty)
plan.intersect.return_value = plan
backend.prepare_cleanup.return_value = plan backend.prepare_cleanup.return_value = plan
backend.cleanup = MagicMock() backend.cleanup = MagicMock()
return backend, plan return backend, plan
@@ -135,10 +136,12 @@ class TestCmdCleanup(unittest.TestCase):
docker.cleanup.assert_called_once_with(docker_plan) docker.cleanup.assert_called_once_with(docker_plan)
fc.cleanup.assert_not_called() fc.cleanup.assert_not_called()
def test_executes_refreshed_plan_after_confirmation(self): def test_executes_only_displayed_resources_still_current(self):
backend = MagicMock() backend = MagicMock()
preview = MagicMock(empty=False) preview = MagicMock(empty=False)
refreshed = MagicMock(empty=False) refreshed = MagicMock(empty=False)
approved = MagicMock(empty=False)
preview.intersect.return_value = approved
backend.prepare_cleanup.side_effect = [preview, refreshed] backend.prepare_cleanup.side_effect = [preview, refreshed]
with patch.object( with patch.object(
@@ -152,7 +155,8 @@ class TestCmdCleanup(unittest.TestCase):
): ):
self.assertEqual(0, cmd.cmd_cleanup([])) self.assertEqual(0, cmd.cmd_cleanup([]))
backend.cleanup.assert_called_once_with(refreshed) preview.intersect.assert_called_once_with(refreshed)
backend.cleanup.assert_called_once_with(approved)
if __name__ == "__main__": if __name__ == "__main__":
+35
View File
@@ -14,9 +14,12 @@ from __future__ import annotations
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch
from tests.unit import use_bottle_root from tests.unit import use_bottle_root
from bot_bottle import bottle_state from bot_bottle import bottle_state
from bot_bottle.backend import EnumerationError
from bot_bottle.backend.docker import cleanup
from bot_bottle.backend.docker.cleanup import _list_orphan_state_dirs from bot_bottle.backend.docker.cleanup import _list_orphan_state_dirs
@@ -116,5 +119,37 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
) )
class TestAuthoritativeDiscovery(unittest.TestCase):
def test_prepare_requires_authoritative_compose_projects(self):
with patch.object(cleanup.docker_mod, "require_docker"), \
patch.object(
cleanup, "list_compose_projects",
side_effect=EnumerationError("compose unavailable"),
) as projects, self.assertRaisesRegex(
EnumerationError, "compose unavailable",
):
cleanup.prepare_cleanup()
projects.assert_called_once_with(
warn_on_error=False,
raise_on_error=True,
)
def test_container_query_failure_raises(self):
failed = cleanup.subprocess.CompletedProcess(
[], 1, stdout="", stderr="daemon unavailable",
)
with patch.object(cleanup.subprocess, "run", return_value=failed), \
self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
cleanup._list_prefixed_containers()
def test_network_query_failure_raises(self):
failed = cleanup.subprocess.CompletedProcess(
[], 1, stdout="", stderr="daemon unavailable",
)
with patch.object(cleanup.subprocess, "run", return_value=failed), \
self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
cleanup._list_prefixed_networks()
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+47 -15
View File
@@ -32,15 +32,19 @@ class TestProcessScan(unittest.TestCase):
self.assertEqual( self.assertEqual(
Path("/cache/run/dev-a"), Path("/cache/run/dev-a"),
fc_cleanup._run_dir_of( fc_cleanup._run_dir_of(
f"firecracker --config-file {run_root}/dev-a/config.json", run_root ("firecracker", "--config-file",
f"{run_root}/dev-a/config.json"), run_root
), ),
) )
# infra/builder VMs elsewhere, or nested paths, are not ours. # infra/builder VMs elsewhere, or nested paths, are not ours.
self.assertIsNone( self.assertIsNone(
fc_cleanup._run_dir_of("firecracker --config-file /elsewhere/config.json", run_root) fc_cleanup._run_dir_of(
("firecracker", "--config-file", "/elsewhere/config.json"),
run_root,
)
) )
self.assertIsNone( self.assertIsNone(
fc_cleanup._run_dir_of("firecracker --no-config", run_root) fc_cleanup._run_dir_of(("firecracker", "--no-config"), run_root)
) )
def test_scan_splits_live_dirs_from_orphan_pids(self): def test_scan_splits_live_dirs_from_orphan_pids(self):
@@ -48,17 +52,40 @@ class TestProcessScan(unittest.TestCase):
run_root = Path(tmp) run_root = Path(tmp)
(run_root / "live-a").mkdir() # dir present -> live VM, protected (run_root / "live-a").mkdir() # dir present -> live VM, protected
# "gone-b" dir intentionally absent -> lingering VMM, orphan pid # "gone-b" dir intentionally absent -> lingering VMM, orphan pid
out = ( args = {
f"111 firecracker --config-file {run_root}/live-a/config.json\n" 111: ("firecracker", "--config-file",
f"222 firecracker --config-file {run_root}/gone-b/config.json\n" f"{run_root}/live-a/config.json"),
"333 firecracker --config-file /elsewhere/config.json\n" 222: ("firecracker", "--config-file",
"notanint firecracker --config-file x\n" f"{run_root}/gone-b/config.json"),
) 333: ("firecracker", "--config-file", "/elsewhere/config.json"),
with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)): }
with patch.object(
fc_cleanup.subprocess, "run",
return_value=_proc("111\n222\n333\nnotanint\n"),
), patch.object(
fc_cleanup, "_process_args", side_effect=args.get,
):
live, orphan_pids = fc_cleanup._scan_processes(run_root) live, orphan_pids = fc_cleanup._scan_processes(run_root)
self.assertEqual({str(run_root / "live-a")}, live) self.assertEqual({str(run_root / "live-a")}, live)
self.assertEqual([222], orphan_pids) self.assertEqual([222], orphan_pids)
def test_scan_preserves_spaces_in_config_path(self):
with tempfile.TemporaryDirectory(prefix="fc cache ") as tmp:
run_root = Path(tmp)
live = run_root / "live bottle"
live.mkdir()
with patch.object(
fc_cleanup.subprocess, "run", return_value=_proc("111\n"),
), patch.object(
fc_cleanup, "_process_args",
return_value=("firecracker", "--config-file",
str(live / "config.json")),
):
self.assertEqual(
({str(live)}, []),
fc_cleanup._scan_processes(run_root),
)
def test_scan_empty_when_pgrep_finds_no_processes(self): def test_scan_empty_when_pgrep_finds_no_processes(self):
with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)): with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)):
self.assertEqual((set(), []), fc_cleanup._scan_processes(Path("/x"))) self.assertEqual((set(), []), fc_cleanup._scan_processes(Path("/x")))
@@ -117,9 +144,14 @@ class TestProcessScan(unittest.TestCase):
run_root = Path(tmp) run_root = Path(tmp)
(run_root / "live-a").mkdir() (run_root / "live-a").mkdir()
(run_root / "dead-b").mkdir() (run_root / "dead-b").mkdir()
out = f"111 firecracker --config-file {run_root}/live-a/config.json\n"
with patch.object(fc_cleanup, "_run_root", return_value=run_root), \ with patch.object(fc_cleanup, "_run_root", return_value=run_root), \
patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)): patch.object(fc_cleanup.subprocess, "run",
return_value=_proc("111\n")), \
patch.object(
fc_cleanup, "_process_args",
return_value=("firecracker", "--config-file",
str(run_root / "live-a/config.json")),
):
plan = fc_cleanup.prepare_cleanup() plan = fc_cleanup.prepare_cleanup()
self.assertEqual((), plan.vm_pids) self.assertEqual((), plan.vm_pids)
self.assertEqual((str(run_root / "dead-b"),), plan.run_dirs) self.assertEqual((str(run_root / "dead-b"),), plan.run_dirs)
@@ -135,11 +167,11 @@ class TestCleanupRemoval(unittest.TestCase):
with patch.object(fc_cleanup, "prepare_cleanup", return_value=plan), \ with patch.object(fc_cleanup, "prepare_cleanup", return_value=plan), \
patch.object(fc_cleanup, "_run_root", return_value=Path("/run")), \ patch.object(fc_cleanup, "_run_root", return_value=Path("/run")), \
patch.object(fc_cleanup, "_terminate_orphan") as terminate, \ patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
patch.object(fc_cleanup.shutil, "rmtree") as rmtree, \ patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree, \
patch.object(fc_cleanup, "info"): patch.object(fc_cleanup, "info"):
fc_cleanup.cleanup(plan) fc_cleanup.cleanup(plan)
terminate.assert_called_once_with(101, Path("/run")) terminate.assert_called_once_with(101, Path("/run"))
rmtree.assert_called_once_with("/run/dev-x", ignore_errors=True) rmtree.assert_called_once_with(Path("/run/dev-x"))
def test_cleanup_skips_resources_no_longer_in_refreshed_plan(self): def test_cleanup_skips_resources_no_longer_in_refreshed_plan(self):
preview = FirecrackerBottleCleanupPlan( preview = FirecrackerBottleCleanupPlan(
@@ -149,7 +181,7 @@ class TestCleanupRemoval(unittest.TestCase):
fc_cleanup, "prepare_cleanup", fc_cleanup, "prepare_cleanup",
return_value=FirecrackerBottleCleanupPlan(), return_value=FirecrackerBottleCleanupPlan(),
), patch.object(fc_cleanup, "_terminate_orphan") as terminate, \ ), patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
patch.object(fc_cleanup.shutil, "rmtree") as rmtree: patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree:
fc_cleanup.cleanup(preview) fc_cleanup.cleanup(preview)
terminate.assert_not_called() terminate.assert_not_called()
rmtree.assert_not_called() rmtree.assert_not_called()
+13
View File
@@ -486,6 +486,19 @@ class TestMalformedStatusHeader(unittest.TestCase):
) )
self.assertEqual(500, status) self.assertEqual(500, status)
def test_backend_timeout_returns_503(self):
with mock.patch(
"bot_bottle.gateway.git_gate.http_backend.subprocess.run",
side_effect=subprocess.TimeoutExpired(["git", "http-backend"], 1),
):
req = urllib.request.Request(
f"http://127.0.0.1:{self._port}/repo.git/info/refs",
method="GET",
)
with self.assertRaises(urllib.error.HTTPError) as raised:
urllib.request.urlopen(req, timeout=3)
self.assertEqual(503, raised.exception.code)
class TestContentLengthBounds(unittest.TestCase): class TestContentLengthBounds(unittest.TestCase):
"""PRD 0041: malformed or oversized Content-Length is rejected before """PRD 0041: malformed or oversized Content-Length is rejected before
+12
View File
@@ -284,6 +284,18 @@ class TestEnsureArtifact(_CacheMixin):
ia.ensure_artifact_gz(version, role=_ROLE) ia.ensure_artifact_gz(version, role=_ROLE)
self.assertEqual(first_calls, len(net.calls)) self.assertEqual(first_calls, len(net.calls))
def test_download_uses_network_deadline(self) -> None:
response = mock.MagicMock()
response.__enter__.return_value = io.BytesIO(b"payload")
with tempfile.TemporaryDirectory() as d, mock.patch.object(
ia.urllib.request, "urlopen", return_value=response,
) as urlopen:
ia._download("https://registry/artifact", Path(d) / "artifact")
self.assertEqual(
ia.ARTIFACT_HTTP_TIMEOUT_SECONDS,
urlopen.call_args.kwargs["timeout"],
)
def test_checksum_mismatch_fails_closed(self) -> None: def test_checksum_mismatch_fails_closed(self) -> None:
version = "beefbeefbeefbeef" version = "beefbeefbeefbeef"
gz = _gz(b"payload") gz = _gz(b"payload")
+17 -1
View File
@@ -6,6 +6,7 @@ import unittest
from unittest.mock import patch from unittest.mock import patch
from bot_bottle.backend import EnumerationError from bot_bottle.backend import EnumerationError
from bot_bottle.backend.cleanup_control import CleanupError
from bot_bottle.backend.macos_container import cleanup, enumerate as enum_mod from bot_bottle.backend.macos_container import cleanup, enumerate as enum_mod
from bot_bottle.backend.macos_container.bottle_cleanup_plan import ( from bot_bottle.backend.macos_container.bottle_cleanup_plan import (
MacosContainerBottleCleanupPlan, MacosContainerBottleCleanupPlan,
@@ -31,7 +32,10 @@ class TestMacosContainerCleanup(unittest.TestCase):
containers=("bot-bottle-a",), containers=("bot-bottle-a",),
networks=("bot-bottle-net-a",), networks=("bot-bottle-net-a",),
) )
with patch.object(cleanup.subprocess, "run") as run: completed = cleanup.subprocess.CompletedProcess(
args=[], returncode=0, stdout="", stderr="",
)
with patch.object(cleanup.subprocess, "run", return_value=completed) as run:
cleanup.cleanup(plan) cleanup.cleanup(plan)
self.assertEqual( self.assertEqual(
["container", "delete", "--force", "bot-bottle-a"], ["container", "delete", "--force", "bot-bottle-a"],
@@ -42,6 +46,18 @@ class TestMacosContainerCleanup(unittest.TestCase):
run.call_args_list[1].args[0], run.call_args_list[1].args[0],
) )
def test_cleanup_attempts_all_resources_then_raises(self):
plan = MacosContainerBottleCleanupPlan(
containers=("bot-bottle-a",), networks=("bot-bottle-net-a",),
)
failed = cleanup.subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr="unavailable",
)
with patch.object(cleanup.subprocess, "run", return_value=failed) as run, \
self.assertRaisesRegex(CleanupError, "unavailable"):
cleanup.cleanup(plan)
self.assertEqual(2, run.call_count)
def test_container_enumeration_failure_aborts(self): def test_container_enumeration_failure_aborts(self):
completed = cleanup.subprocess.CompletedProcess( completed = cleanup.subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr="service unavailable", args=[], returncode=1, stdout="", stderr="service unavailable",
@@ -224,6 +224,17 @@ class TestAgentSecrets(unittest.TestCase):
reopened = RegistryStore(self.db) reopened = RegistryStore(self.db)
self.assertEqual({"K": "v"}, reopened.get_agent_secrets("bottle-1")) self.assertEqual({"K": "v"}, reopened.get_agent_secrets("bottle-1"))
def test_v6_migration_clears_legacy_secret_rows(self) -> None:
self.store.store_agent_secrets("bottle-1", {"K": "legacy"})
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"UPDATE schema_versions SET version = 5 "
"WHERE module = 'orchestrator_registry'"
)
conn.commit()
self.store.migrate()
self.assertEqual({}, self.store.get_agent_secrets("bottle-1"))
class TestReapAbsent(unittest.TestCase): class TestReapAbsent(unittest.TestCase):
"""`reap_absent` — the self-heal for rows whose bottle is gone. """`reap_absent` — the self-heal for rows whose bottle is gone.
+15 -12
View File
@@ -3,8 +3,6 @@
from __future__ import annotations from __future__ import annotations
import base64 import base64
import hashlib
import hmac
import unittest import unittest
from bot_bottle.orchestrator.store.secret_store import ( from bot_bottle.orchestrator.store.secret_store import (
@@ -83,16 +81,21 @@ class TestDecryptErrors(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "authentication failed"): with self.assertRaisesRegex(ValueError, "authentication failed"):
decrypt_value(self.secret, tampered) decrypt_value(self.secret, tampered)
def test_reads_legacy_ciphertext_for_migration(self) -> None: def test_rejects_legacy_ciphertext(self) -> None:
key = base64.urlsafe_b64decode(self.secret + "==") legacy = base64.urlsafe_b64encode(
nonce = b"0123456789abcdef" b"0123456789abcdeflegacy-token",
plaintext = b"legacy-token" ).rstrip(b"=").decode()
stream = hmac.new( with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
key, nonce + (0).to_bytes(4, "big"), hashlib.sha256, decrypt_value(self.secret, legacy)
).digest()
ciphertext = bytes(p ^ k for p, k in zip(plaintext, stream)) def test_rejects_authenticated_blob_with_changed_version(self) -> None:
legacy = base64.urlsafe_b64encode(nonce + ciphertext).rstrip(b"=").decode() raw = bytearray(base64.urlsafe_b64decode(
self.assertEqual("legacy-token", decrypt_value(self.secret, legacy)) encrypt_value(self.secret, "secret-token") + "=="
))
raw[0] ^= 1
downgraded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
decrypt_value(self.secret, downgraded)
def test_truncated_blob_raises_value_error(self) -> None: def test_truncated_blob_raises_value_error(self) -> None:
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
+10
View File
@@ -49,6 +49,16 @@ class TestPut(unittest.TestCase):
self.assertNotIsInstance(req.data, (bytes, bytearray)) self.assertNotIsInstance(req.data, (bytes, bytearray))
self.assertEqual(str(len(payload)), req.get_header("Content-length")) self.assertEqual(str(len(payload)), req.get_header("Content-length"))
def test_put_uses_network_deadline(self) -> None:
with mock.patch.object(
pub.urllib.request, "urlopen", return_value=_Resp(),
) as urlopen:
pub._put("https://reg/pkg", b"payload", token="")
self.assertEqual(
pub._REGISTRY_HTTP_TIMEOUT_SECONDS,
urlopen.call_args.kwargs["timeout"],
)
def test_small_bytes_body_still_works(self) -> None: def test_small_bytes_body_still_works(self) -> None:
captured: list[urllib.request.Request] = [] captured: list[urllib.request.Request] = []