change_manifest.py
5.97 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
"""Workspace snapshotting and change-manifest/diff generation for isolated
runs (Coder/Validator nodes, or a Supervisor candidate-instance checkout).
Adapted verbatim from graphloop's change_manifest.py (2026-07-30) -- no
functional changes. See test_change_manifest.py in graphloop for the
original contract this was validated against; graphlemon should add an
equivalent test module when this is wired into a real Enabler unit.
"""
import difflib
import hashlib
import os
import posixpath
import re
IGNORED_DIRS = {".git", "__pycache__", ".pytest_cache"}
IGNORED_FILES = {".DS_Store"}
class ChangeManifestError(Exception):
"""Raised when build_change_manifest encounters untrusted input:
a symlink under the isolated workspace, or a workspace-escaping path
in the source snapshot."""
def _iter_files(root, reject_symlinks):
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
keep_dirs = []
for name in dirnames:
if name in IGNORED_DIRS:
continue
full = os.path.join(dirpath, name)
if os.path.islink(full):
if reject_symlinks:
raise ChangeManifestError(f"symlink encountered: {full}")
continue
keep_dirs.append(name)
dirnames[:] = keep_dirs
for name in filenames:
if name in IGNORED_FILES:
continue
full = os.path.join(dirpath, name)
if os.path.islink(full):
if reject_symlinks:
raise ChangeManifestError(f"symlink encountered: {full}")
continue
if not os.path.isfile(full):
continue
rel = os.path.relpath(full, root).replace(os.sep, "/")
yield rel, full
def _analyze_file(full):
with open(full, "rb") as handle:
data = handle.read()
sha256 = hashlib.sha256(data).hexdigest()
if b"\x00" in data:
kind = "binary"
else:
try:
data.decode("utf-8")
kind = "text"
except UnicodeDecodeError:
kind = "binary"
return sha256, len(data), kind
def _snapshot_files(root, reject_symlinks):
files = {}
for rel, full in _iter_files(root, reject_symlinks):
sha256, size, kind = _analyze_file(full)
files[rel] = {"sha256": sha256, "size": size, "kind": kind}
return files
def _validate_relative_path(rel):
if not rel or rel.startswith("/"):
raise ChangeManifestError(f"unsafe path in snapshot: {rel!r}")
normalized = posixpath.normpath(rel)
if normalized == ".." or normalized.startswith("../"):
raise ChangeManifestError(f"unsafe path in snapshot: {rel!r}")
def _safe_join(root, rel):
full = os.path.normpath(os.path.join(root, *rel.split("/")))
if os.path.commonpath([root, full]) != os.path.normpath(root):
raise ChangeManifestError(f"path escapes root: {rel!r}")
return full
def snapshot_workspace(path):
root = os.path.realpath(path)
return {"root": root, "files": _snapshot_files(root, reject_symlinks=False)}
def build_change_manifest(source_snapshot, isolated_path):
source_files = source_snapshot["files"]
for rel in source_files:
_validate_relative_path(rel)
isolated_root = os.path.realpath(isolated_path)
isolated_files = _snapshot_files(isolated_root, reject_symlinks=True)
changed_files = []
for rel in source_files.keys() | isolated_files.keys():
old = source_files.get(rel)
new = isolated_files.get(rel)
if old and new:
if old["sha256"] == new["sha256"]:
continue
changed_files.append({
"path": rel,
"status": "modified",
"old_hash": old["sha256"],
"new_hash": new["sha256"],
"kind": new["kind"],
})
elif new:
changed_files.append({
"path": rel,
"status": "added",
"old_hash": None,
"new_hash": new["sha256"],
"kind": new["kind"],
})
else:
changed_files.append({
"path": rel,
"status": "deleted",
"old_hash": old["sha256"],
"new_hash": None,
"kind": old["kind"],
})
changed_files.sort(key=lambda entry: entry["path"])
return {
"source_root": source_snapshot["root"],
"isolated_root": isolated_root,
"changed_files": changed_files,
}
def _read_text_lines(full):
with open(full, "rb") as handle:
data = handle.read()
return data.decode("utf-8").splitlines(keepends=True)
def unified_diff(manifest):
changed_files = manifest["changed_files"]
if not changed_files:
return ""
source_root = manifest["source_root"]
isolated_root = manifest["isolated_root"]
sections = []
for entry in changed_files:
rel = entry["path"]
status = entry["status"]
old_label = "/dev/null" if status == "added" else rel
new_label = "/dev/null" if status == "deleted" else rel
if entry["kind"] == "binary":
sections.append(f"Binary files {old_label} and {new_label} differ\n")
continue
old_lines = (
_read_text_lines(_safe_join(source_root, rel))
if status != "added" else []
)
new_lines = (
_read_text_lines(_safe_join(isolated_root, rel))
if status != "deleted" else []
)
diff_lines = difflib.unified_diff(
old_lines, new_lines, fromfile=old_label, tofile=new_label
)
# Collapse "@@ -l,s +l,s @@" hunk headers to a single leading "@@"
# marker so callers can count hunks via a plain substring count.
diff_lines = (
re.sub(r" @@\n$", "\n", line) if line.startswith("@@") else line
for line in diff_lines
)
sections.append("".join(diff_lines))
return "".join(sections)