235 lines
8.1 KiB
Python
235 lines
8.1 KiB
Python
#!/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())
|