weichen
b450e51c40
Why: Preserve a single Dev entry and independent repository ownership while allowing active feature branches to migrate without losing source or Git state. Evidence: SELF 29 real Git/Go workspace tests and ordinary component evidence/lock tests passed; detailed RED/GREEN runs are recorded by make evidence in the task card. Impact: Add fixed cloning, selected-version baselines, development overlays, npm source resolution, and independently preserved legacy snapshots. No other active worktree is modified. Refs: Versioned dependencies U2/U5; R15; feature/dependency-workspace
76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import subprocess
|
|
import tempfile
|
|
|
|
|
|
class WorkspaceError(Exception):
|
|
pass
|
|
|
|
|
|
def run(args, cwd, env=None, check=True):
|
|
result = subprocess.run(args, cwd=cwd, env=env, text=True, capture_output=True,
|
|
timeout=180)
|
|
if check and result.returncode:
|
|
raise WorkspaceError(f"{args[0]} failed ({result.returncode}): {result.stderr.strip() or result.stdout.strip()}")
|
|
return result
|
|
|
|
|
|
def relative_path(value, allow_dot=False):
|
|
if not isinstance(value, str) or not value or "\\" in value or any(ord(c) < 32 for c in value):
|
|
raise WorkspaceError("invalid relative path")
|
|
path = PurePosixPath(value)
|
|
if path.is_absolute() or ".." in path.parts or (value == "." and not allow_dot):
|
|
raise WorkspaceError(f"unsafe relative path: {value}")
|
|
return path
|
|
|
|
|
|
def safe_path(root, value, allow_dot=False):
|
|
relative_path(value, allow_dot)
|
|
path = root / value
|
|
# Refuse symlink path components even when a link currently points inside root.
|
|
# This avoids creating a second apparent owner and later retargeting escapes.
|
|
current = root
|
|
for part in PurePosixPath(value).parts:
|
|
current /= part
|
|
if current.is_symlink():
|
|
raise WorkspaceError(f"symlink path is not owned by this workspace: {value}")
|
|
if not path.resolve().is_relative_to(root.resolve()):
|
|
raise WorkspaceError(f"path escapes workspace: {value}")
|
|
return path
|
|
|
|
|
|
def read_json(path):
|
|
if path.is_symlink():
|
|
raise WorkspaceError(f"refusing symlink metadata: {path}")
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except (OSError, ValueError) as error:
|
|
raise WorkspaceError(f"cannot read {path.name}: {error}") from error
|
|
|
|
|
|
def atomic_write(path, content):
|
|
if path.is_symlink():
|
|
raise WorkspaceError(f"refusing symlink output: {path}")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
descriptor, temporary = tempfile.mkstemp(prefix="." + path.name + "-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(descriptor, "w") as stream:
|
|
stream.write(content)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def write_json(path, value):
|
|
atomic_write(path, json.dumps(value, indent=2, sort_keys=True) + "\n")
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|