Skip to content

Commit a3ecec6

Browse files
Improve virtual container path handling
1 parent 677cfef commit a3ecec6

2 files changed

Lines changed: 461 additions & 0 deletions

File tree

scripts/virtual_container.py

Lines changed: 289 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,289 @@
1+
#!/usr/bin/env python3
2+
"""Utility to spin up a lightweight virtual container for running commands.
3+
4+
The virtual container concept implemented here provides an isolated working
5+
directory (with environment settings) for executing shell commands. It is
6+
aimed at cases where a throwaway sandbox is preferred to running commands
7+
directly in the repository tree. The tool supports both interactive and
8+
one-off command execution modes.
9+
"""
10+
11+
from __future__ import annotations
12+
13+
import argparse
14+
import os
15+
import shutil
16+
import subprocess
17+
import sys
18+
import tempfile
19+
from dataclasses import dataclass
20+
from pathlib import Path
21+
from typing import Dict, Iterable, Mapping, MutableMapping, Sequence
22+
23+
24+
@dataclass
25+
class VirtualContainer:
26+
"""Manage a throwaway workspace for executing shell commands.
27+
28+
Parameters
29+
----------
30+
root:
31+
The root directory that represents ``/`` within the virtual container.
32+
cwd:
33+
The current working directory inside the container. Defaults to the
34+
``root``.
35+
env:
36+
A copy of the environment variables that should be supplied to spawned
37+
commands.
38+
"""
39+
40+
root: Path
41+
cwd: Path
42+
env: MutableMapping[str, str]
43+
44+
@classmethod
45+
def create(
46+
cls,
47+
*,
48+
base_dir: str | None = None,
49+
env: Mapping[str, str] | None = None,
50+
) -> "VirtualContainer":
51+
"""Instantiate a new virtual container."""
52+
53+
if base_dir is not None:
54+
Path(base_dir).mkdir(parents=True, exist_ok=True)
55+
root_dir = Path(
56+
tempfile.mkdtemp(prefix="virtual_container_", dir=base_dir)
57+
).resolve()
58+
env_vars: Dict[str, str] = dict(os.environ)
59+
if env:
60+
env_vars.update(env)
61+
return cls(root=root_dir, cwd=root_dir, env=env_vars)
62+
63+
def run(
64+
self,
65+
command: Sequence[str] | str,
66+
*,
67+
check: bool = False,
68+
) -> subprocess.CompletedProcess[str]:
69+
"""Run a command inside the container and return the completed process."""
70+
71+
if isinstance(command, str):
72+
exec_command = command
73+
shell = True
74+
else:
75+
exec_command = list(command)
76+
shell = False
77+
completed = subprocess.run(
78+
exec_command,
79+
cwd=str(self.cwd),
80+
env=self.env,
81+
capture_output=True,
82+
text=True,
83+
shell=shell,
84+
)
85+
if check and completed.returncode != 0:
86+
raise subprocess.CalledProcessError(
87+
completed.returncode,
88+
exec_command,
89+
output=completed.stdout,
90+
stderr=completed.stderr,
91+
)
92+
return completed
93+
94+
def cleanup(self) -> None:
95+
"""Remove the virtual container directory tree."""
96+
97+
if self.root.exists():
98+
shutil.rmtree(self.root)
99+
100+
def format_path(self, path: Path | None = None) -> str:
101+
"""Return a container-relative representation of *path*."""
102+
103+
target = path or self.cwd
104+
if target == self.root:
105+
return "/"
106+
return f"/{target.relative_to(self.root)}"
107+
108+
def resolve_path(self, raw_path: str) -> Path:
109+
"""Resolve *raw_path* to a directory within the container.
110+
111+
Absolute paths are interpreted relative to the container root in the
112+
same way they would be inside a real container. Attempts to traverse
113+
outside ``/`` are clamped to the root directory so that commands like
114+
``cd ..`` behave the way they do on POSIX systems.
115+
"""
116+
117+
if not raw_path:
118+
raise ValueError("path cannot be empty")
119+
120+
path_obj = Path(raw_path)
121+
if path_obj.is_absolute():
122+
candidate = self.root
123+
parts = path_obj.parts[1:]
124+
else:
125+
candidate = self.cwd
126+
parts = path_obj.parts
127+
128+
for part in parts:
129+
if part in {"", "."}:
130+
continue
131+
if part == "..":
132+
if candidate == self.root:
133+
continue
134+
candidate = candidate.parent
135+
continue
136+
candidate = candidate / part
137+
138+
resolved = candidate.resolve()
139+
try:
140+
resolved.relative_to(self.root)
141+
except ValueError as exc: # pragma: no cover - safety guard
142+
raise ValueError("cannot leave the container root") from exc
143+
if not resolved.exists():
144+
raise FileNotFoundError(raw_path)
145+
if not resolved.is_dir():
146+
raise NotADirectoryError(raw_path)
147+
return resolved
148+
149+
def change_directory(self, raw_path: str) -> Path:
150+
"""Change the container's working directory."""
151+
152+
target = self.resolve_path(raw_path)
153+
self.cwd = target
154+
return self.cwd
155+
156+
157+
def parse_env_assignments(assignments: Iterable[str]) -> Dict[str, str]:
158+
"""Parse ``KEY=VALUE`` assignments supplied on the command line."""
159+
160+
parsed: Dict[str, str] = {}
161+
for item in assignments:
162+
if "=" not in item:
163+
raise ValueError(f"invalid environment assignment: '{item}'")
164+
key, value = item.split("=", 1)
165+
key = key.strip()
166+
if not key:
167+
raise ValueError("environment variable name cannot be empty")
168+
parsed[key] = value
169+
return parsed
170+
171+
172+
def build_parser() -> argparse.ArgumentParser:
173+
"""Create the command-line argument parser for the script."""
174+
175+
parser = argparse.ArgumentParser(
176+
description=(
177+
"Create a lightweight virtual container directory and execute commands "
178+
"inside it."
179+
)
180+
)
181+
parser.add_argument(
182+
"command",
183+
nargs=argparse.REMAINDER,
184+
help=(
185+
"Command to execute inside the container. When omitted an interactive "
186+
"prompt is started."
187+
),
188+
)
189+
parser.add_argument(
190+
"-e",
191+
"--env",
192+
action="append",
193+
default=[],
194+
metavar="KEY=VALUE",
195+
help="Environment variables to inject into the container.",
196+
)
197+
parser.add_argument(
198+
"--base-dir",
199+
type=str,
200+
default=None,
201+
help="Directory where container instances should be created.",
202+
)
203+
parser.add_argument(
204+
"--persist",
205+
action="store_true",
206+
help="Do not delete the container directory when the program exits.",
207+
)
208+
return parser
209+
210+
211+
def container_prompt(container: VirtualContainer) -> str:
212+
"""Return a shell-like prompt for the interactive session."""
213+
214+
return f"{container.format_path()}$ "
215+
216+
217+
def interactive_loop(container: VirtualContainer) -> int:
218+
"""Run an interactive terminal loop inside the container."""
219+
220+
print("Starting virtual container interactive session.")
221+
print("Type 'exit' or 'quit' (or press Ctrl-D) to leave.")
222+
while True:
223+
try:
224+
command = input(container_prompt(container)).strip()
225+
except EOFError:
226+
print()
227+
break
228+
if not command:
229+
continue
230+
if command in {"exit", "quit"}:
231+
break
232+
if command == "cd":
233+
container.change_directory("/")
234+
continue
235+
if command.startswith("cd "):
236+
target = command[3:].strip() or "/"
237+
try:
238+
container.change_directory(target)
239+
except OSError as exc:
240+
print(f"cd: {exc}")
241+
continue
242+
result = container.run(command)
243+
if result.stdout:
244+
sys.stdout.write(result.stdout)
245+
if result.stderr:
246+
sys.stderr.write(result.stderr)
247+
print(f"[exit {result.returncode}]")
248+
return 0
249+
250+
251+
def main(argv: Sequence[str] | None = None) -> int:
252+
parser = build_parser()
253+
args = parser.parse_args(argv)
254+
try:
255+
env_vars = parse_env_assignments(args.env)
256+
except ValueError as exc:
257+
parser.error(str(exc))
258+
container = VirtualContainer.create(base_dir=args.base_dir, env=env_vars)
259+
exit_code = 0
260+
try:
261+
if args.command:
262+
command = args.command
263+
if command and command[0] == "--":
264+
command = command[1:]
265+
if not command:
266+
parser.error("a command must follow '--'")
267+
if len(command) == 1:
268+
command = command[0]
269+
result = container.run(command)
270+
if result.stdout:
271+
sys.stdout.write(result.stdout)
272+
if result.stderr:
273+
sys.stderr.write(result.stderr)
274+
exit_code = result.returncode
275+
else:
276+
exit_code = interactive_loop(container)
277+
finally:
278+
if args.persist:
279+
print(
280+
"Container preserved at",
281+
container.root,
282+
)
283+
else:
284+
container.cleanup()
285+
return exit_code
286+
287+
288+
if __name__ == "__main__":
289+
raise SystemExit(main())

0 commit comments

Comments
 (0)