selective_apply.py
5.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""Selective, safe application of a change_manifest diff from an isolated
workspace (or a Supervisor candidate checkout) back into its source
workspace.
Adapted from graphloop's selective_apply.py (2026-07-30). Only change from
the original: `import change_manifest as cm` -> relative package import,
since this now lives in the `supervisor.isolation` package alongside
change_manifest.py instead of as a flat top-level module. Logic is
otherwise unchanged. See test_selective_apply.py in graphloop for the
original contract; graphlemon should add an equivalent test module when
this is wired into a real Enabler unit.
"""
import hashlib
import json
import os
import posixpath
import tempfile
from . import change_manifest as cm
class SelectiveApplyError(Exception):
"""Raised when apply_manifest rejects a call before any mutation of
source_root, or refuses to write through a symlink it discovers."""
def _normalize_relative_path(path):
if not isinstance(path, str) or path == "":
raise SelectiveApplyError(f"empty or invalid path: {path!r}")
if posixpath.isabs(path):
raise SelectiveApplyError(f"absolute path not allowed: {path!r}")
normalized = posixpath.normpath(path)
segments = normalized.split("/")
if normalized in (".", "") or any(segment == ".." for segment in segments):
raise SelectiveApplyError(f"path traversal not allowed: {path!r}")
return normalized
def _reject_symlink_components(root, rel_path):
current = root
for segment in rel_path.split("/"):
current = os.path.join(current, segment)
if os.path.islink(current):
raise SelectiveApplyError(f"symlink encountered in path: {rel_path!r}")
def _manifest_digest(manifest):
payload = json.dumps(manifest, sort_keys=True).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _hash_file(path):
with open(path, "rb") as handle:
return hashlib.sha256(handle.read()).hexdigest()
def apply_manifest(manifest, source_root, isolated_root, approved_paths,
expected_source_snapshot=None):
if not isinstance(manifest, dict) or not isinstance(manifest.get("changed_files"), list):
raise SelectiveApplyError("manifest must be a dict with a 'changed_files' list")
if approved_paths is None:
raise SelectiveApplyError("approved_paths is required and must not be None")
if not os.path.isdir(source_root):
raise SelectiveApplyError(f"source_root is not a directory: {source_root!r}")
if not os.path.isdir(isolated_root):
raise SelectiveApplyError(f"isolated_root is not a directory: {isolated_root!r}")
source_root = os.path.realpath(source_root)
isolated_root = os.path.realpath(isolated_root)
entries_by_path = {entry["path"]: entry for entry in manifest["changed_files"]}
normalized_approved = []
for raw_path in approved_paths:
normalized = _normalize_relative_path(raw_path)
if normalized not in entries_by_path:
raise SelectiveApplyError(f"path not present in manifest: {raw_path!r}")
normalized_approved.append(normalized)
if expected_source_snapshot is not None:
current_snapshot = cm.snapshot_workspace(source_root)
if current_snapshot != expected_source_snapshot:
raise SelectiveApplyError(
"source_root has drifted from expected_source_snapshot"
)
approved_entries = [entries_by_path[p] for p in normalized_approved]
# Reject any destination that runs through a symlink (including
# symlinked intermediate directories), and preload isolated bytes for
# add/modify entries, so every possible read/validation failure surfaces
# before source_root is touched.
staged_data = {}
for entry in approved_entries:
rel = entry["path"]
_reject_symlink_components(source_root, rel)
dest = os.path.join(source_root, *rel.split("/"))
if os.path.isdir(dest):
raise SelectiveApplyError(f"destination is a directory: {rel!r}")
if entry["status"] != "deleted":
_reject_symlink_components(isolated_root, rel)
src = os.path.join(isolated_root, *rel.split("/"))
if not os.path.isfile(src):
raise SelectiveApplyError(f"isolated source file missing: {rel!r}")
with open(src, "rb") as handle:
staged_data[rel] = handle.read()
applied = []
already_applied = []
for entry in approved_entries:
rel = entry["path"]
dest = os.path.join(source_root, *rel.split("/"))
if entry["status"] == "deleted":
if os.path.exists(dest):
os.remove(dest)
applied.append(rel)
else:
already_applied.append(rel)
continue
data = staged_data[rel]
if os.path.isfile(dest) and _hash_file(dest) == entry["new_hash"]:
already_applied.append(rel)
continue
dest_dir = os.path.dirname(dest)
os.makedirs(dest_dir, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=dest_dir, prefix=".selective-apply-")
try:
with os.fdopen(fd, "wb") as tmp_handle:
tmp_handle.write(data)
os.replace(tmp_path, dest)
except BaseException:
if os.path.exists(tmp_path):
os.remove(tmp_path)
raise
applied.append(rel)
skipped = sorted(set(entries_by_path) - set(normalized_approved))
return {
"applied": sorted(applied),
"skipped": skipped,
"already_applied": sorted(already_applied),
"manifest_id": _manifest_digest(manifest),
}