legacy_identifiers.py (9279B)
1 """Exact legacy-name exceptions, separate from current app naming.""" 2 3 from __future__ import annotations 4 5 import collections 6 import json 7 import re 8 import subprocess 9 import tempfile 10 from pathlib import Path 11 from typing import Any 12 13 POLICY = "test-fixtures/legacy-identifiers.v1.json" 14 MAX_BYTES = 2 * 1024 * 1024 15 MAX_FILES = 4096 16 ROOTS = ( 17 "Tera/", 18 "TeraTests/", 19 "TeraUITests/", 20 "TeraPublicAPITests/", 21 "core/crates/", 22 "scripts/", 23 "TeraFFI/scripts/", 24 ) 25 ROOT_FILES = { 26 "AGENTS.md", 27 "README.md", 28 "Cargo.toml", 29 "Package.swift", 30 "project.yml", 31 "Makefile", 32 "TeraFFI/Makefile", 33 "TeraFFI/producer.toml", 34 "radroots.lib.source-lock.v1.toml", 35 } 36 GENERATED_OR_TOOL_CACHE = ( 37 "Tera/Generated/", 38 "Tera/Frameworks/", 39 "scripts/persona-verifier/.venv/", 40 ) 41 TEXT_SUFFIXES = { 42 ".swift", 43 ".rs", 44 ".py", 45 ".sh", 46 ".toml", 47 ".plist", 48 ".xcconfig", 49 ".strings", 50 ".json", 51 ".yml", 52 ".md", 53 ".kt", 54 ".kts", 55 } 56 IDENTIFIER = re.compile(r"(?i)\b[a-z0-9_]*radroots[a-z0-9_]*(?:[.-][a-z0-9_-]+)*") 57 DECLARATION = re.compile( 58 r"(?m)^\s*(?:(?:public|private|internal|fileprivate|final|open|indirect|" 59 r"nonisolated|static|async|unsafe|pub(?:\([^\n)]*\))?)\s+)*" 60 r"(?:class|struct|enum|actor|protocol|typealias|trait|type|fn|fun|func|static\s+(?:var|let))\s+" 61 r"(?i:radroots)[a-zA-Z0-9_]*\b" 62 ) 63 CATEGORY_FIELDS = {"owner", "reason", "reader", "removal_condition"} 64 65 66 class LegacyIdentifierError(ValueError): 67 """A source-free naming contract rejection.""" 68 69 70 def require(condition: bool, message: str) -> None: 71 if not condition: 72 raise LegacyIdentifierError(message) 73 74 75 def safe_path(value: Any) -> str: 76 require(isinstance(value, str) and bool(value), "legacy path is absent") 77 path = Path(value) 78 require( 79 not path.is_absolute() and ".." not in path.parts and str(path) == value, 80 "legacy path is unsafe", 81 ) 82 require( 83 not any(character in value for character in "*?[]\0\n"), 84 "legacy path must be exact", 85 ) 86 return value 87 88 89 def selected(path: str) -> bool: 90 if path.startswith(GENERATED_OR_TOOL_CACHE) or "__pycache__" in Path(path).parts: 91 return False 92 return path in ROOT_FILES or ( 93 path.startswith(ROOTS) and Path(path).suffix in TEXT_SUFFIXES 94 ) 95 96 97 def read_regular(root: Path, relative: str) -> bytes: 98 path = root 99 for part in Path(safe_path(relative)).parts: 100 path /= part 101 require(not path.is_symlink(), "legacy source contains a symlink") 102 require(path.is_file(), "legacy source is not a regular file") 103 size = path.stat().st_size 104 require(size <= MAX_BYTES, "legacy source exceeds its byte bound") 105 value = path.read_bytes() 106 require(len(value) == size, "legacy source changed during inspection") 107 return value 108 109 110 def git_paths(root: Path, *, ignored: bool = False) -> list[str]: 111 flags = ["--others", "--ignored"] if ignored else ["--cached", "--others"] 112 paths = [*ROOTS, *sorted(ROOT_FILES)] 113 paths.extend(f":(exclude,glob){path}**" for path in GENERATED_OR_TOOL_CACHE) 114 with tempfile.TemporaryFile() as output, tempfile.TemporaryFile() as errors: 115 try: 116 result = subprocess.run( 117 ["git", "ls-files", "-z", *flags, "--exclude-standard", "--", *paths], 118 cwd=root, 119 stdin=subprocess.DEVNULL, 120 stdout=output, 121 stderr=errors, 122 timeout=30, 123 check=False, 124 ) 125 except (OSError, subprocess.TimeoutExpired) as error: 126 raise LegacyIdentifierError( 127 "legacy source inventory is unavailable" 128 ) from error 129 output.seek(0) 130 value = output.read(MAX_BYTES + 1) 131 require( 132 result.returncode == 0 and len(value) <= MAX_BYTES, 133 "legacy source inventory failed or exceeded its bound", 134 ) 135 return sorted(set(value.decode("utf-8").split("\0")) - {""}) 136 137 138 def source_texts(root: Path) -> dict[str, str]: 139 require( 140 not any(selected(path) for path in git_paths(root, ignored=True)), 141 "ignored app source cannot bypass the legacy guard", 142 ) 143 paths = [path for path in git_paths(root) if selected(path)] 144 require( 145 0 < len(paths) <= MAX_FILES, "legacy source file inventory exceeds its bound" 146 ) 147 sources = {} 148 total = 0 149 for path in paths: 150 value = read_regular(root, path) 151 total += len(value) 152 require(total <= 32 * MAX_BYTES, "legacy source aggregate exceeds its bound") 153 sources[path] = value.decode("utf-8") 154 return sources 155 156 157 def categories(value: Any) -> set[str]: 158 require(isinstance(value, dict) and bool(value), "legacy categories are absent") 159 for item in value.values(): 160 require( 161 isinstance(item, dict) and set(item) == CATEGORY_FIELDS, 162 "legacy category metadata differs", 163 ) 164 require( 165 all(isinstance(text, str) and bool(text.strip()) for text in item.values()), 166 "legacy category owner, reader or removal condition is absent", 167 ) 168 return set(value) 169 170 171 def occurrences(value: Any, identifier: str) -> dict[tuple[str, str], int]: 172 require(isinstance(value, list) and bool(value), "legacy occurrences are absent") 173 result = {} 174 for item in value: 175 require( 176 isinstance(item, dict) and set(item) == {"path", "count"}, 177 "legacy occurrence fields differ", 178 ) 179 path = safe_path(item["path"]) 180 require( 181 selected(path), "legacy exception is outside the inspected source scope" 182 ) 183 count = item["count"] 184 require( 185 type(count) is int and 0 < count <= 10000, 186 "legacy occurrence count is invalid", 187 ) 188 key = (path, identifier) 189 require(key not in result, "legacy occurrence is duplicated") 190 result[key] = count 191 return result 192 193 194 def policy_index(policy: Any) -> dict[tuple[str, str], int]: 195 require( 196 isinstance(policy, dict) 197 and set(policy) == {"schema", "baseline_commit", "categories", "entries"}, 198 "legacy policy fields differ", 199 ) 200 require( 201 policy["schema"] == "tera.legacy-identifiers.v1", "legacy policy schema differs" 202 ) 203 require( 204 isinstance(policy["baseline_commit"], str) 205 and re.fullmatch(r"[0-9a-f]{40}", policy["baseline_commit"]) is not None, 206 "legacy baseline revision is invalid", 207 ) 208 kinds = categories(policy["categories"]) 209 entries = policy["entries"] 210 require( 211 isinstance(entries, list) and 0 < len(entries) <= 1024, 212 "legacy entry inventory exceeds its bound", 213 ) 214 result = {} 215 seen = set() 216 for item in entries: 217 require( 218 isinstance(item, dict) 219 and set(item) == {"identifier", "category", "occurrences"}, 220 "legacy entry fields differ", 221 ) 222 identifier = item["identifier"] 223 require( 224 isinstance(identifier, str) 225 and IDENTIFIER.fullmatch(identifier) is not None, 226 "legacy identifier must be exact", 227 ) 228 require(identifier not in seen, "legacy identifier is duplicated") 229 require( 230 isinstance(item["category"], str) and item["category"] in kinds, 231 "legacy category is unknown", 232 ) 233 seen.add(identifier) 234 result.update(occurrences(item["occurrences"], identifier)) 235 return result 236 237 238 def inspect_sources(sources: dict[str, str]) -> collections.Counter: 239 observed = collections.Counter() 240 for path, text in sources.items(): 241 if Path(path).suffix in {".swift", ".rs", ".kt", ".kts"}: 242 require( 243 not Path(path).name.lower().startswith("radroots"), 244 "app-owned source filename uses the legacy brand", 245 ) 246 require( 247 DECLARATION.search(text) is None, 248 "app-owned declaration uses the legacy brand", 249 ) 250 observed.update((path, match[0]) for match in IDENTIFIER.finditer(text)) 251 return observed 252 253 254 def validate(policy: Any, sources: dict[str, str]) -> dict[str, int]: 255 expected = policy_index(policy) 256 observed = inspect_sources(sources) 257 require( 258 set(observed) <= set(expected), "unapproved legacy identifier or source path" 259 ) 260 require( 261 observed == expected, 262 "legacy exception is stale or its occurrence count differs", 263 ) 264 return { 265 "identifiers": len(policy["entries"]), 266 "source_identifier_pairs": len(expected), 267 "occurrences": sum(observed.values()), 268 } 269 270 271 def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: 272 result = {} 273 for key, value in pairs: 274 require(key not in result, "legacy JSON field is duplicated") 275 result[key] = value 276 return result 277 278 279 def verify(root: Path) -> dict[str, int]: 280 try: 281 policy = json.loads(read_regular(root, POLICY), object_pairs_hook=unique_object) 282 return validate(policy, source_texts(root)) 283 except (OSError, UnicodeError, json.JSONDecodeError) as error: 284 raise LegacyIdentifierError( 285 "legacy source or policy cannot be decoded" 286 ) from error