#!/usr/bin/env python3 """Reject production environment readers outside the crank-config leaf crate.""" from __future__ import annotations import argparse import re import sys from pathlib import Path MAX_FILE_BYTES = 1024 * 1024 IDENTIFIER = re.compile(r"(?:r#)?[A-Za-z_][A-Za-z0-9_]*") ALLOWED_ENV_MACRO = ("env", "!", "(", '"CARGO_PKG_VERSION"', ")") class BoundaryError(Exception): pass def resolve_file(root: Path, supplied: str) -> Path: logical = Path(supplied) if logical.is_absolute() or ".." in logical.parts: raise BoundaryError("explicit path must be repository-relative") path = root / logical if path.is_symlink() or not path.is_file(): raise BoundaryError("explicit path must be a regular non-symlink file") try: path.resolve().relative_to(root) except ValueError as error: raise BoundaryError("explicit path escapes repository") from error return path def default_files(root: Path) -> list[Path]: files: list[Path] = [] for base in (root / "apps", root / "crates"): if not base.exists(): continue for path in base.rglob("*.rs"): relative_parts = path.relative_to(root).parts if "crank-config" in relative_parts or "tests" in relative_parts: continue if path.is_symlink(): raise BoundaryError("production Rust source must not be a symlink") if not path.is_file(): raise BoundaryError("production Rust source must be a regular file") files.append(path) return sorted(files) def tokenize(text: str) -> list[tuple[str, int]]: """Return Rust-like tokens while discarding comments and preserving literals.""" tokens: list[tuple[str, int]] = [] index = 0 line = 1 block_depth = 0 while index < len(text): if block_depth: if text.startswith("/*", index): block_depth += 1 index += 2 elif text.startswith("*/", index): block_depth -= 1 index += 2 else: line += text[index] == "\n" index += 1 continue if text.startswith("//", index): newline = text.find("\n", index + 2) if newline < 0: break index = newline continue if text.startswith("/*", index): block_depth = 1 index += 2 continue character = text[index] if character.isspace(): line += character == "\n" index += 1 continue token_line = line char_match = re.match(r"'(?:\\.|[^\\'\n])'", text[index:]) if char_match: tokens.append((char_match.group(), token_line)) index += char_match.end() continue raw_match = re.match(r'r(#{0,255})"', text[index:]) if raw_match: terminator = '"' + raw_match.group(1) end = text.find(terminator, index + raw_match.end()) if end < 0: raise BoundaryError("unterminated raw string literal") end += len(terminator) token = text[index:end] tokens.append((token, token_line)) line += token.count("\n") index = end continue if character == '"': end = index + 1 escaped = False while end < len(text): current = text[end] if current == '"' and not escaped: end += 1 break escaped = current == "\\" and not escaped if current != "\\": escaped = False line += current == "\n" end += 1 else: raise BoundaryError("unterminated string literal") tokens.append((text[index:end], token_line)) index = end continue match = IDENTIFIER.match(text, index) if match: tokens.append((match.group(), token_line)) index = match.end() continue if text.startswith("::", index): tokens.append(("::", token_line)) index += 2 continue tokens.append((character, token_line)) index += 1 if block_depth: raise BoundaryError("unterminated block comment") return tokens def scan(path: Path) -> list[tuple[int, int]]: data = path.read_bytes() if len(data) > MAX_FILE_BYTES: raise BoundaryError("Rust source exceeds boundary scanner size limit") try: text = data.decode("utf-8") except UnicodeDecodeError as error: raise BoundaryError("Rust source is not valid UTF-8") from error tokens = tokenize(text) values = [token for token, _ in tokens] findings: set[tuple[int, int]] = set() std_aliases: set[str] = set() for index in range(len(tokens)): window = values[index : index + 5] line = tokens[index][1] if ( len(window) >= 5 and window[:4] == ["std", "::", "env", "::"] and window[4] in {"var", "vars", "var_os", "vars_os"} ): findings.add((line, 1)) if len(window) >= 3 and window[0] == "env" and window[1] == "::" and window[2] in { "var", "vars", "var_os", "vars_os", }: findings.add((line, 2)) if values[index].removeprefix("r#").endswith("_from_env"): findings.add((line, 3)) if len(window) >= 2 and window[0] in {"dotenv", "dotenvy"} and window[1] == "::": findings.add((line, 4)) if len(window) >= 5 and window[0] in {"env", "option_env"} and window[1] == "!": if tuple(window) != ALLOWED_ENV_MACRO: findings.add((line, 5)) if ( len(window) >= 4 and window[0] in {"use", "crate"} and window[1] == "std" and window[2] == "as" ): std_aliases.add(window[3]) grouped = values[index : index + 8] if ( len(grouped) == 8 and grouped[:6] == ["use", "std", "::", "{", "self", "as"] and IDENTIFIER.fullmatch(grouped[6]) and grouped[7] in {"}", ","} ): std_aliases.add(grouped[6]) if len(window) >= 5 and window[:3] == ["extern", "crate", "std"] and window[3] == "as": std_aliases.add(window[4]) if window[:3] == ["use", "std", "::"]: end = index + 3 while end < len(tokens) and values[end] != ";": if values[end] == "env": findings.add((line, 1)) break end += 1 for index, (token, line) in enumerate(tokens): if token in std_aliases and values[index : index + 3] == [token, "::", "env"]: findings.add((line, 6)) return sorted(findings) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, default=Path(".")) parser.add_argument("--files", nargs="*") args = parser.parse_args() root = args.root.resolve() try: files = ( [resolve_file(root, item) for item in args.files] if args.files is not None else default_files(root) ) violations: list[tuple[str, int, int]] = [] for path in files: if "crank-config" in path.relative_to(root).parts: continue for line, pattern in scan(path): violations.append((path.relative_to(root).as_posix(), line, pattern)) except (BoundaryError, OSError) as error: print(f"config boundary check failed: {error}", file=sys.stderr) return 1 if violations: for path, line, pattern in sorted(violations): print( f"config boundary violation: {path}:{line} pattern={pattern}", file=sys.stderr, ) return 1 print(f"config boundary check passed: files={len(files)}") return 0 if __name__ == "__main__": raise SystemExit(main())