Code / public_research_code/check_projection.py
public_research_code/check_projection.py
1
"""Fail closed when a public-projection candidate exceeds HuleEdu's boundary."""
2
3
from __future__ import annotations
4
5
import hashlib
6
import json
7
import os
8
import re
9
import stat
10
import sys
11
from pathlib import Path
12
from typing import TypeAlias
13
14
JsonValue: TypeAlias = str | int | float | bool | None | list["JsonValue"] | dict[str, "JsonValue"]
15
16
INVENTORY_PATH = Path("public_research_code/inventory.json")
17
MANIFEST_PATH = Path("public_research_code/manifest.json")
18
EXPECTED_CHECK = ["/usr/bin/env", "python3", "public_research_code/check_projection.py"]
19
PROHIBITED_COMPONENTS = frozenset(
20
{
21
".git",
22
".orchestration",
23
"artifact",
24
"artifacts",
25
"cache",
26
"caches",
27
"credential",
28
"credentials",
29
"data",
30
"dataset",
31
"datasets",
32
"evidence",
33
"fixture",
34
"fixtures",
35
"governance",
36
"identity",
37
"identities",
38
"operation",
39
"operations",
40
"private",
41
"secret",
42
"secrets",
43
"weight",
44
"weights",
45
}
46
)
47
CREDENTIAL_PATTERNS = (
48
re.compile(r"AKIA[0-9A-Z]{16}"),
49
re.compile(r"gh[pousr]_[A-Za-z0-9]{36,255}"),
50
re.compile(
51
r"(?i)(?:api[_-]?key|password|secret|token)\s*[:=]\s*['\"][A-Za-z0-9_./+=-]{16,}['\"]"
52
),
53
)
54
KNOWN_HARMLESS_CREDENTIAL_MATCHES = frozenset(
55
{
56
# This identity enum member is a public error code, not a credential assignment.
57
(
58
"libs/common_core/src/common_core/identity_enums.py",
59
"PASSWORD = " + '"invalid_password"',
60
),
61
}
62
)
63
64
65
def _failure(message: str) -> int:
66
print(f"public projection check failed: {message[:240]}", file=sys.stderr)
67
return 1
68
69
70
def _load_json(path: Path) -> tuple[JsonValue | None, str | None]:
71
try:
72
return json.loads(path.read_text(encoding="utf-8")), None
73
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
74
return None, f"cannot read UTF-8 JSON {path.as_posix()}: {error}"
75
76
77
def _safe_relative_path(value: str) -> bool:
78
parts = value.split("/")
79
return (
80
bool(value)
81
and not value.startswith("/")
82
and "\\" not in value
83
and all(part not in {"", ".", ".."} for part in parts)
84
)
85
86
87
def _inventory_files(data: JsonValue) -> tuple[set[str] | None, str | None]:
88
if not isinstance(data, dict) or set(data) != {"schema_version", "files"}:
89
return None, "inventory must contain exactly schema_version and files"
90
if data["schema_version"] != 1:
91
return None, "inventory schema_version must be 1"
92
files = data["files"]
93
if not isinstance(files, list) or not files:
94
return None, "inventory files must be a non-empty list"
95
if not all(isinstance(path, str) and _safe_relative_path(path) for path in files):
96
return None, "inventory files must contain safe POSIX-relative paths"
97
if files != sorted(files) or len(files) != len(set(files)):
98
return None, "inventory files must be sorted and unique"
99
return set(files), None
100
101
102
def _manifest_limits(
103
data: JsonValue, inventory_digest: str
104
) -> tuple[dict[str, int] | None, str | None]:
105
if not isinstance(data, dict) or set(data) != {
106
"schema_version",
107
"inventory_sha256",
108
"consumer_check",
109
"limits",
110
}:
111
return None, "manifest has an unexpected shape"
112
if data["schema_version"] != 1:
113
return None, "manifest schema_version must be 1"
114
if data["inventory_sha256"] != inventory_digest:
115
return None, "manifest inventory_sha256 does not match inventory.json"
116
if data["consumer_check"] != EXPECTED_CHECK:
117
return None, "manifest consumer_check is not the required public check"
118
limits = data["limits"]
119
if not isinstance(limits, dict) or set(limits) != {
120
"max_files",
121
"max_file_bytes",
122
"max_total_bytes",
123
}:
124
return None, "manifest limits have an unexpected shape"
125
if not all(type(value) is int and value > 0 for value in limits.values()):
126
return None, "manifest limits must be positive integers"
127
return {name: value for name, value in limits.items() if type(value) is int}, None
128
129
130
def _candidate_files(root: Path) -> tuple[set[str] | None, str | None]:
131
paths: set[str] = set()
132
for directory, directory_names, file_names in os.walk(root, followlinks=False):
133
directory_path = Path(directory)
134
for name in directory_names:
135
child = directory_path / name
136
if child.is_symlink() or not child.is_dir():
137
return None, f"candidate contains a non-directory entry: {child.relative_to(root)}"
138
for name in file_names:
139
child = directory_path / name
140
try:
141
mode = child.lstat().st_mode
142
except OSError as error:
143
return None, f"cannot inspect candidate file {child.relative_to(root)}: {error}"
144
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode):
145
return None, f"candidate contains a non-regular file: {child.relative_to(root)}"
146
paths.add(child.relative_to(root).as_posix())
147
return paths, None
148
149
150
def _prohibited_path(path: str) -> bool:
151
parts = path.split("/")
152
if any(part.lower() in PROHIBITED_COMPONENTS or part.startswith(".env") for part in parts):
153
return True
154
return any(part.lower() in {"model", "models"} for part in parts[:-1]) and not path.endswith(
155
".py"
156
)
157
158
159
def _contains_credential_like_content(relative_path: str, text: str) -> bool:
160
for pattern in CREDENTIAL_PATTERNS:
161
for match in pattern.finditer(text):
162
if (relative_path, match.group()) not in KNOWN_HARMLESS_CREDENTIAL_MATCHES:
163
return True
164
return False
165
166
167
def _validate_content(root: Path, files: set[str], limits: dict[str, int]) -> str | None:
168
if len(files) > limits["max_files"]:
169
return "candidate file count exceeds manifest max_files"
170
total_bytes = 0
171
for relative_path in sorted(files):
172
if _prohibited_path(relative_path):
173
return f"candidate contains prohibited path category: {relative_path}"
174
candidate_path = root.joinpath(*relative_path.split("/"))
175
try:
176
candidate_path.relative_to(root)
177
mode = candidate_path.lstat().st_mode
178
content = candidate_path.read_bytes()
179
except OSError as error:
180
return f"cannot read candidate file {relative_path}: {error}"
181
if stat.S_ISLNK(mode) or not stat.S_ISREG(mode):
182
return f"candidate contains a non-regular file: {relative_path}"
183
if len(content) > limits["max_file_bytes"]:
184
return f"candidate file exceeds max_file_bytes: {relative_path}"
185
total_bytes += len(content)
186
if total_bytes > limits["max_total_bytes"]:
187
return "candidate files exceed max_total_bytes"
188
if b"\x00" in content:
189
return f"candidate contains NUL content: {relative_path}"
190
try:
191
text = content.decode("utf-8")
192
except UnicodeDecodeError:
193
return f"candidate is not UTF-8: {relative_path}"
194
if _contains_credential_like_content(relative_path, text):
195
return f"candidate contains credential-like content: {relative_path}"
196
return None
197
198
199
def validate_candidate(root: Path) -> str | None:
200
"""Return a bounded refusal reason, or None when the candidate is safe."""
201
202
inventory_path = root / INVENTORY_PATH
203
manifest_path = root / MANIFEST_PATH
204
inventory, error = _load_json(inventory_path)
205
if error is not None:
206
return error
207
files, error = _inventory_files(inventory)
208
if error is not None or files is None:
209
return error
210
try:
211
inventory_digest = hashlib.sha256(inventory_path.read_bytes()).hexdigest()
212
except OSError as read_error:
213
return f"cannot read inventory.json bytes: {read_error}"
214
manifest, error = _load_json(manifest_path)
215
if error is not None:
216
return error
217
limits, error = _manifest_limits(manifest, inventory_digest)
218
if error is not None or limits is None:
219
return error
220
candidate_files, error = _candidate_files(root)
221
if error is not None or candidate_files is None:
222
return error
223
if candidate_files != files:
224
return "candidate file set does not equal inventory files"
225
return _validate_content(root, files, limits)
226
227
228
def main(argv: list[str]) -> int:
229
if len(argv) != 2:
230
return _failure("expected exactly one projector-appended candidate root")
231
candidate_argument = Path(argv[1])
232
if candidate_argument.is_symlink():
233
return _failure("candidate root must not be a symlink")
234
try:
235
root = candidate_argument.resolve(strict=True)
236
except OSError as error:
237
return _failure(f"cannot resolve candidate root: {error}")
238
if not root.is_dir():
239
return _failure("candidate root must be a directory")
240
error = validate_candidate(root)
241
if error is not None:
242
return _failure(error)
243
return 0
244
245
246
if __name__ == "__main__":
247
raise SystemExit(main(sys.argv))