diff --git a/.agents/skills/autoreview/SKILL.md b/.agents/skills/autoreview/SKILL.md index 1745b9be816f..775f10335eb5 100644 --- a/.agents/skills/autoreview/SKILL.md +++ b/.agents/skills/autoreview/SKILL.md @@ -17,6 +17,8 @@ Use when: - after non-trivial code edits, before final/commit/ship - reviewing a local branch or PR branch after fixes +Do not require autoreview for a change whose entire diff is prose-only internal notes or `SKILL.md` documentation. Still inspect the diff directly and run the repository's lightweight documentation validation, if any. This exception does not cover user-facing documentation, executable examples, configuration, scripts, generated files, or behavior changes. + ## Contract - Treat review output as advisory. Never blindly apply it. @@ -36,7 +38,7 @@ Use when: - Tools are useful in review mode. Codex receives the validated bundle in an empty workspace so ignored files and linked-worktree metadata remain unreadable; web search stays available for dependency contracts and upstream docs. - Security perspective is always included, but it should not cripple legitimate functionality. Report security findings only when the change creates a concrete, actionable risk or removes an important safety check. - Reviewer subprocesses preserve engine authentication and non-credentialed proxy variables needed by headless or restricted-network environments while stripping process-injection, Git override, and credentialed proxy values. -- Review bundles fail closed before engine invocation when tracked or untracked paths look sensitive or patch text looks secret-like. Obvious synthetic values shaped like `-` remain reviewable, such as `token: "test-token"`, without one-off allowlists. Safe large diffs are scanned in full, sent as one pass while they fit the aggregate prompt limit, then partitioned into complete bounded passes without truncation. +- Before engine invocation, autoreview runs TruffleHog over temporary snapshots of the exact added or modified content under review. It intentionally matches TruffleHog's low-false-positive pre-commit policy (`verified,unknown`); it does not classify arbitrary password-like strings or rescan unchanged history. Install TruffleHog using its official platform-neutral instructions; autoreview fails with that link when the binary is unavailable and never auto-installs it. Repositories should also run TruffleHog in pull-request CI as a backup outside autoreview; repository-local Git hooks are optional. Review bundles still omit security-sensitive paths or files, and explicit prompt and dataset inputs remain checked before engine invocation. Safe large diffs are sent as one pass while they fit the aggregate prompt limit, then partitioned into complete bounded passes without truncation. - For regression provenance, keep roles separate: blamed code author, blamed PR author, PR merger/committer, current PR author, and PR/date. If no blamed PR is traceable, use the blamed commit as the provenance: commit SHA, date, and author username. Do not guess a merger or frame missing PR metadata as a separate finding. - If the blamed PR was merged by `clawsweeper[bot]` or another automation, identify the human trigger when practical. Check timeline/comments first; if rate-limited, use gitcrawl/cache or public PR HTML. Look for maintainer commands such as `@clawsweeper automerge`, `/landpr`, or labels/status comments that armed automerge. Report `automerge triggered by @login`; if not found, say trigger unknown. - Do not invoke built-in `codex review`, nested reviewers, or reviewer panels from inside the review. The helper builds one validated bundle, calls the selected engine once for normal inputs or once per complete bounded chunk for oversized inputs, validates the structured results, and stops. diff --git a/.agents/skills/autoreview/scripts/autoreview b/.agents/skills/autoreview/scripts/autoreview index 9d611cda2e04..4aae6601bbbe 100755 --- a/.agents/skills/autoreview/scripts/autoreview +++ b/.agents/skills/autoreview/scripts/autoreview @@ -12,6 +12,7 @@ import functools import hashlib import io import json +import math import os import queue import re @@ -50,6 +51,7 @@ SAFE_GIT_CONFIG_ARGS = ( "pager.show=cat", ) SAFE_DIFF_FLAGS = ("--no-ext-diff", "--no-textconv", "--no-renames") +DIFF_HUNK_CONTENT_BOUNDARY = "\0autoreview-diff-hunk-boundary\0" ENGINE_GIT_CONFIG_OVERRIDES = ( ("core.fsmonitor", "false"), ("core.pager", "cat"), @@ -192,6 +194,9 @@ SECRET_ASSIGNMENT_KEY_PATTERN = ( rf"|(?[^\"\r\n]{8,})\"|" @@ -210,12 +215,63 @@ SECRET_ASSIGNMENT_PREFIX_PATTERN = re.compile( rf"(?i){SECRET_ASSIGNMENT_KEY_PATTERN}" r"\s*(?:=(?!=|>)|:(?![:=]))\s*" ) +SECRET_BOOLEAN_DECLARATION_PATTERN = re.compile( + r"(?im)^[ \t]*(?:(?:abstract|const|declare|export|final|internal|lateinit|" + r"open|override|private|protected|public|readonly|static)\s+)*" + rf"(?:val|var|let|const)\s+" + rf"(?P{SECRET_ASSIGNMENT_KEY_PATTERN})\s*:\s*" + rf"(?P{SECRET_BOOLEAN_TYPE_PATTERN})" + r"(?=\s*(?:=|[,;)\]}]|$))" +) +PRIVATE_KEY_BOUNDARY_PREFIX = r"-----" +PRIVATE_KEY_BEGIN_PATTERN = re.compile( + PRIVATE_KEY_BOUNDARY_PREFIX + + r"BEGIN (?:RSA |DSA |EC |OPENSSH |PGP |ENCRYPTED )?" + r"PRIVATE KEY(?: BLOCK)?-----" +) +PRIVATE_KEY_END_PATTERN = re.compile( + PRIVATE_KEY_BOUNDARY_PREFIX + + r"END (?:RSA |DSA |EC |OPENSSH |PGP |ENCRYPTED )?" + r"PRIVATE KEY(?: BLOCK)?-----" +) +PRIVATE_KEY_BODY_PATTERN = re.compile( + r"(?" + r"(?:[A-Za-z0-9+/]{4,}={0,2}|[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)" + r")(?![A-Za-z0-9+/=])" +) +BEARER_CREDENTIAL_PATTERN = re.compile( + r"(?i)bearer\s+(?P[A-Za-z0-9._~+/-]{20,}=*)" +) +SECRET_FALLBACK_LITERAL_PATTERNS = ( + re.compile(r'"(?P[^"\r\n]+)"'), + re.compile(r"'(?P[^'\r\n]+)'"), + re.compile(r"`(?P[^`\r\n]+)`"), +) +SECRET_FALLBACK_UNQUOTED_PATTERN = re.compile( + r"(?[A-Za-z0-9_./+=:@#$%&*!?-]{4,})" + r"(?![A-Za-z0-9_./+=:@#$%&*!?-])" +) +CONFIG_PATH_SEGMENT_PATTERN = ( + r"(?:[a-z][A-Za-z0-9]{0,63}|[a-z][a-z0-9]*(?:-[a-z0-9]+)+" + r"|\$\{[A-Za-z_$][A-Za-z0-9_$]*(?:\.[A-Za-z_$][A-Za-z0-9_$]*)*\})" +) +CANONICAL_CONFIG_PATH_REFERENCE_PATTERN = re.compile( + rf"{CONFIG_PATH_SEGMENT_PATTERN}(?:\.{CONFIG_PATH_SEGMENT_PATTERN}){{2,}}" + r"\.[a-z][A-Za-z0-9]{0,63}" +) +CONFIG_PATH_CREDENTIAL_FIELD_PATTERN = re.compile( + r"(?:apiKey|password|Password|secret|Secret|token|Token|credential|Credential" + r"|keyRef|KeyRef|privateKey|PrivateKey|serviceAccount)" +) SECRET_VALUE_PATTERNS = [ - re.compile( - r"-----BEGIN (?:RSA |DSA |EC |OPENSSH |PGP |ENCRYPTED )?" - r"PRIVATE KEY(?: BLOCK)?-----" - ), - re.compile(r"(?i)bearer\s+[A-Za-z0-9._-]{20,}"), + PRIVATE_KEY_BEGIN_PATTERN, + BEARER_CREDENTIAL_PATTERN, re.compile(r"\b(?:sk|rk|pk|org|proj)-[A-Za-z0-9_-]{20,}\b"), re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b"), re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), @@ -278,6 +334,7 @@ class ReviewChunk(NamedTuple): content: str context: str = "" SECRET_PLACEHOLDER_VALUES = { + "__openclaw_redacted__", "changeme", "decoy-token", "dummy", @@ -294,6 +351,9 @@ SECRET_PLACEHOLDER_VALUES = { "sample", "secret-token", "test-auth-token", + "test-key", + "test-secret", + "test-token", "test-token-placeholder", "token-oversized", "clawrouter-e2e-secret", @@ -417,7 +477,10 @@ CSHARP_METHOD_PREFIX_PATTERN = ( ) CSHARP_EVIDENCE_WINDOW = 8192 SOURCE_CODE_REFERENCE_ROOT_VALUES = { + "accountConfig", "attemptAuthProfileStore", + "baseConfig", + "merged", } SOURCE_CODE_REFERENCE_ROOT_PATTERN = re.compile( r"(? str: return ref +TRUFFLEHOG_INSTALL_URL = "https://github.com/trufflesecurity/trufflehog#installation" +TRUFFLEHOG_FINDINGS_EXIT_CODE = 183 + + +def git_bytes( + repo: Path, + *args: str, + check: bool = True, +) -> subprocess.CompletedProcess[bytes]: + result = subprocess.run( + [ + resolve_command("git", repo), + "--no-optional-locks", + *SAFE_GIT_CONFIG_ARGS, + *args, + ], + cwd=repo, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=safe_git_env(repo), + ) + if check and result.returncode != 0: + detail = (result.stderr or result.stdout).decode( + SUBPROCESS_TEXT_ENCODING, + errors=SUBPROCESS_TEXT_ERRORS, + ) + raise SystemExit( + f"Git failed while preparing the TruffleHog scan ({result.returncode}): " + f"{display_escape(detail, 1000, multiline=True)}" + ) + return result + + +def snapshot_destination(root: Path, rel: str) -> Path: + path = PurePosixPath(rel) + if ( + path.is_absolute() + or not path.parts + or path.parts[0].casefold() == ".git" + or any(part in {"", ".", ".."} for part in path.parts) + ): + raise SystemExit("Git returned an unsafe path while preparing the TruffleHog scan") + return root.joinpath(*path.parts) + + +def write_snapshot_blob(root: Path, rel: str, content: bytes) -> None: + destination = snapshot_destination(root, rel) + try: + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(content) + except OSError as exc: + raise SystemExit( + "unable to prepare changed content for the TruffleHog scan: " + f"{display_escape(exc, 500)}" + ) from exc + + +def materialize_index_snapshot(repo: Path, root: Path, paths: list[str]) -> int: + if not paths: + return 0 + records = git_bytes( + repo, + "--literal-pathspecs", + "ls-files", + "--stage", + "-z", + "--", + *paths, + ).stdout + count = 0 + for record in records.split(b"\0"): + if not record: + continue + metadata, raw_path = record.split(b"\t", 1) + mode, object_id, stage = metadata.split() + if stage != b"0": + raise SystemExit( + "cannot run TruffleHog with unmerged index entries; resolve the conflict and rerun autoreview" + ) + if mode == b"160000": + continue + rel = raw_path.decode(SUBPROCESS_TEXT_ENCODING, errors="strict") + content = git_bytes( + repo, + "cat-file", + "blob", + object_id.decode("ascii"), + ).stdout + write_snapshot_blob(root, rel, content) + count += 1 + return count + + +def materialize_tree_snapshot( + repo: Path, + root: Path, + ref: str, + paths: list[str], +) -> int: + if not paths: + return 0 + records = git_bytes( + repo, + "--literal-pathspecs", + "ls-tree", + "-rz", + ref, + "--", + *paths, + ).stdout + count = 0 + for record in records.split(b"\0"): + if not record: + continue + metadata, raw_path = record.split(b"\t", 1) + mode, object_type, object_id = metadata.split() + if object_type != b"blob" or mode == b"160000": + continue + rel = raw_path.decode(SUBPROCESS_TEXT_ENCODING, errors="strict") + content = git_bytes( + repo, + "cat-file", + "blob", + object_id.decode("ascii"), + ).stdout + write_snapshot_blob(root, rel, content) + count += 1 + return count + + +def open_worktree_parent(repo: Path, rel_path: Path) -> int | bool | None: + required_dir_fd = {os.open, os.stat, os.readlink} + if not required_dir_fd <= os.supports_dir_fd: + return None + flags = ( + os.O_RDONLY + | getattr(os, "O_BINARY", 0) + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_DIRECTORY", 0) + | getattr(os, "O_NOFOLLOW", 0) + ) + descriptor = os.open(repo.resolve(), flags) + try: + for part in rel_path.parts[:-1]: + try: + component = os.stat( + part, + dir_fd=descriptor, + follow_symlinks=False, + ) + except (FileNotFoundError, NotADirectoryError): + os.close(descriptor) + return False + if stat.S_ISLNK(component.st_mode): + raise OSError("symlinked parent directory") + if not stat.S_ISDIR(component.st_mode): + os.close(descriptor) + return False + next_descriptor = os.open(part, flags, dir_fd=descriptor) + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except OSError as exc: + os.close(descriptor) + raise SystemExit( + "refusing to snapshot changed content through a symlinked or unstable parent directory: " + f"{display_escape(exc, 500)}" + ) from exc + + +def copy_worktree_file(repo: Path, root: Path, rel: str) -> bool: + rel_path = Path(*PurePosixPath(rel).parts) + parent_descriptor = open_worktree_parent(repo, rel_path) + if parent_descriptor is False: + return False + if parent_descriptor is not None: + try: + source_stat = os.stat( + rel_path.name, + dir_fd=parent_descriptor, + follow_symlinks=False, + ) + except (FileNotFoundError, NotADirectoryError): + os.close(parent_descriptor) + return False + except OSError as exc: + os.close(parent_descriptor) + raise SystemExit( + "unable to read changed content for the TruffleHog scan: " + f"{display_escape(exc, 500)}" + ) from exc + if stat.S_ISLNK(source_stat.st_mode): + try: + content = os.fsencode( + os.readlink(rel_path.name, dir_fd=parent_descriptor) + ) + except OSError as exc: + raise SystemExit( + "unable to read changed symlink for the TruffleHog scan: " + f"{display_escape(exc, 500)}" + ) from exc + finally: + os.close(parent_descriptor) + write_snapshot_blob(root, rel, content) + return True + if stat.S_ISDIR(source_stat.st_mode): + os.close(parent_descriptor) + return False + if not stat.S_ISREG(source_stat.st_mode): + os.close(parent_descriptor) + raise SystemExit( + "changed content is not a regular file; remove it or resolve the repository state before rerunning autoreview" + ) + flags = ( + os.O_RDONLY + | getattr(os, "O_BINARY", 0) + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NOFOLLOW", 0) + ) + try: + descriptor = os.open( + rel_path.name, + flags, + dir_fd=parent_descriptor, + ) + except OSError as exc: + raise SystemExit( + "unable to snapshot changed content for the TruffleHog scan: " + f"{display_escape(exc, 500)}" + ) from exc + finally: + os.close(parent_descriptor) + opened = os.fstat(descriptor) + if not stat.S_ISREG(opened.st_mode) or ( + source_stat.st_dev, + source_stat.st_ino, + ) != ( + opened.st_dev, + opened.st_ino, + ): + os.close(descriptor) + raise SystemExit( + "unable to snapshot changed content for the TruffleHog scan: file changed while opening" + ) + destination = snapshot_destination(root, rel) + destination.parent.mkdir(parents=True, exist_ok=True) + try: + with destination.open("wb") as output: + while chunk := os.read(descriptor, 1024 * 1024): + output.write(chunk) + after = os.fstat(descriptor) + if ( + opened.st_mode, + opened.st_size, + opened.st_mtime_ns, + ) != ( + after.st_mode, + after.st_size, + after.st_mtime_ns, + ): + raise OSError("file changed while reading") + except OSError as exc: + destination.unlink(missing_ok=True) + raise SystemExit( + "unable to snapshot changed content for the TruffleHog scan: " + f"{display_escape(exc, 500)}" + ) from exc + finally: + os.close(descriptor) + return True + + source = repo / rel_path + if rel_path.parent != Path(".") and raw_repo_path_has_symlink_component( + repo, + rel_path.parent, + ): + raise SystemExit( + "refusing to snapshot changed content through a symlinked parent directory" + ) + try: + source_stat = source.lstat() + except (FileNotFoundError, NotADirectoryError): + return False + if stat.S_ISLNK(source_stat.st_mode): + write_snapshot_blob(root, rel, os.fsencode(os.readlink(source))) + return True + if stat.S_ISDIR(source_stat.st_mode): + return False + if not stat.S_ISREG(source_stat.st_mode): + raise SystemExit( + "changed content is not a regular file; remove it or resolve the repository state before rerunning autoreview" + ) + content = source.read_bytes() + if source.lstat() != source_stat: + raise SystemExit( + "unable to snapshot changed content for the TruffleHog scan: file changed while reading" + ) + write_snapshot_blob(root, rel, content) + return True + + +def materialize_worktree_snapshot(repo: Path, root: Path, paths: list[str]) -> int: + return sum(copy_worktree_file(repo, root, rel) for rel in paths) + + +def clear_snapshot_paths(root: Path, paths: list[str]) -> None: + for rel in paths: + destination = snapshot_destination(root, rel) + if destination.is_symlink() or destination.is_file(): + destination.unlink() + elif destination.exists(): + shutil.rmtree(destination) + + +def commit_snapshot(repo: Path, label: str) -> str: + git(repo, "add", "-A", "-f") + git( + repo, + "-c", + "user.name=Autoreview", + "-c", + "user.email=autoreview@example.invalid", + "commit", + "-q", + "--allow-empty", + "-m", + label, + ) + return git(repo, "rev-parse", "HEAD").strip() + + +def safe_trufflehog_env(repo: Path) -> dict[str, str]: + env = safe_git_env(repo) + for key in ( + "ALL_PROXY", + "HTTP_PROXY", + "HTTPS_PROXY", + "NO_PROXY", + "all_proxy", + "http_proxy", + "https_proxy", + "no_proxy", + ): + value = os.environ.get(key) + if not value: + continue + if key.casefold() != "no_proxy" and not safe_proxy_url(value): + raise SystemExit( + f"unsafe credentialed or malformed proxy URL in {key}; " + "configure a credential-free proxy URL before running autoreview" + ) + env[key] = value + return env + + +def prepare_trufflehog_history( + repo: Path, + target: str, + target_ref: str | None, + commit_ref: str, + root: Path, +) -> str: + git(root, "init", "-q") + if target == "local": + staged_paths = git_path_list( + repo, + "diff", + *SAFE_DIFF_FLAGS, + "--name-only", + "--cached", + "-z", + ) + unstaged_paths = git_path_list( + repo, + "diff", + *SAFE_DIFF_FLAGS, + "--name-only", + "-z", + ) + unstaged_paths.extend( + git_path_list( + repo, + *global_excludes_git_args(repo), + "ls-files", + "--others", + "--exclude-standard", + "-z", + ) + ) + paths = sorted(set(staged_paths + unstaged_paths)) + head = git(repo, "rev-parse", "--verify", "HEAD", check=False).strip() + if head: + materialize_tree_snapshot(repo, root, head, paths) + base_commit = commit_snapshot(root, "baseline") + clear_snapshot_paths(root, paths) + materialize_index_snapshot(repo, root, paths) + commit_snapshot(root, "staged") + clear_snapshot_paths(root, paths) + materialize_worktree_snapshot(repo, root, paths) + commit_snapshot(root, "working") + # TruffleHog's Git parser scans additions only. Reverse commits make + # deleted bytes additions without exposing unchanged baseline content. + clear_snapshot_paths(root, paths) + materialize_index_snapshot(repo, root, paths) + commit_snapshot(root, "reverse staged") + clear_snapshot_paths(root, paths) + if head: + materialize_tree_snapshot(repo, root, head, paths) + commit_snapshot(root, "reverse baseline") + return base_commit + + if target == "branch": + assert target_ref + target_ref = validate_git_ref(repo, target_ref, "base") + paths = git_path_list( + repo, + "diff", + *SAFE_DIFF_FLAGS, + "--name-only", + "-z", + "--end-of-options", + f"{target_ref}...HEAD", + ) + base_ref = git(repo, "merge-base", target_ref, "HEAD").strip() + reviewed_ref = "HEAD" + else: + reviewed_ref = validate_git_ref(repo, commit_ref, "commit") + parents = git(repo, "rev-list", "--parents", "-n", "1", reviewed_ref).split() + if len(parents) > 2: + raise SystemExit( + "commit review does not accept merge commits; review the branch diff " + "or an individual parent-relative commit instead" + ) + base_ref = parents[1] if len(parents) == 2 else "" + paths = git_path_list( + repo, + "show", + *SAFE_DIFF_FLAGS, + "--name-only", + "--format=", + "-z", + "--end-of-options", + reviewed_ref, + ) + if base_ref: + materialize_tree_snapshot(repo, root, base_ref, paths) + base_commit = commit_snapshot(root, "baseline") + clear_snapshot_paths(root, paths) + materialize_tree_snapshot(repo, root, reviewed_ref, paths) + commit_snapshot(root, "reviewed") + # Scan the reverse diff too so deleted bytes become additions for + # TruffleHog without scanning unchanged content from the baseline. + clear_snapshot_paths(root, paths) + if base_ref: + materialize_tree_snapshot(repo, root, base_ref, paths) + commit_snapshot(root, "reverse baseline") + return base_commit + + +def run_trufflehog_preflight( + repo: Path, + target: str, + target_ref: str | None, + commit_ref: str, +) -> None: + trufflehog_bin = find_command("trufflehog", repo) + if not trufflehog_bin: + raise SystemExit( + "TruffleHog is required but was not found. Install it using the official " + f"instructions, then rerun autoreview: {TRUFFLEHOG_INSTALL_URL}" + ) + started = time.monotonic() + with tempfile.TemporaryDirectory( + prefix="autoreview-trufflehog.", + dir=safe_temp_root(repo), + ) as tempdir: + scan_repo = Path(tempdir) + base_commit = prepare_trufflehog_history( + repo, + target, + target_ref, + commit_ref, + scan_repo, + ) + result = run( + [ + trufflehog_bin, + "git", + scan_repo.resolve().as_uri(), + "--since-commit", + base_commit, + "--branch", + "HEAD", + "--no-update", + "--no-color", + # Match TruffleHog's pre-commit mode. Unverified heuristic + # candidates are excluded to avoid restoring false positives. + "--results=verified,unknown", + "--fail", + "--fail-on-scan-errors", + ], + repo, + check=False, + env=safe_trufflehog_env(repo), + ) + if result.returncode == TRUFFLEHOG_FINDINGS_EXIT_CODE: + raise SystemExit( + "TruffleHog found verified or unknown credentials in the reviewed changes. " + "Remove or rotate them, then rerun autoreview." + ) + if result.returncode != 0: + raise SystemExit( + "TruffleHog could not complete the credential scan. Run TruffleHog directly " + "to diagnose the scanner error, then rerun autoreview." + ) + print(f"trufflehog: clean ({time.monotonic() - started:.1f}s)") + + def bounded(text: str, limit: int = 180_000) -> str: if len(text) <= limit: return text @@ -3091,18 +3689,7 @@ def credentialed_uri_risk( ) ): return True - if uri_password_is_interpolated( - text, - credential.scheme_start, - credential.value, - credential.host, - credential.context, - ): - continue - if credential.has_password or uri_userinfo_literal_risk( - credential.value, - allow_plus_address=True, - ): + if uri_credential_value_risk(text, credential): return True return False @@ -3696,6 +4283,24 @@ def uri_password_is_format_placeholder( return False +def uri_credential_value_risk( + text: str, + credential: UriAuthorityCredential, +) -> bool: + if uri_password_is_interpolated( + text, + credential.scheme_start, + credential.value, + credential.host, + credential.context, + ): + return False + return credential.has_password or uri_userinfo_literal_risk( + credential.value, + allow_plus_address=True, + ) + + def format_arguments_are_references(arguments: str) -> bool: values = split_top_level_call_arguments(arguments) if not values or any(not value.strip() for value in values): @@ -3850,6 +4455,13 @@ def secret_literal_risk( if match.group("bare") is not None: if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", value): continue + if ( + javascript_dialect is not None + and re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*:", value) + ): + # Object/type property names are syntax, not credential content. + # Any literal value after the colon is scanned independently. + continue if ( javascript_dialect is not None and SOURCE_CODE_REFERENCE_ROOT_PATTERN.fullmatch(value) @@ -4521,6 +5133,123 @@ def safe_credential_lookup_argument( ) +def mask_javascript_comments(text: str) -> str: + masked = list(text) + cursor = 0 + quote: str | None = None + escaped = False + while cursor < len(text): + char = text[cursor] + next_char = text[cursor + 1] if cursor + 1 < len(text) else "" + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + cursor += 1 + continue + regex_end = javascript_regex_literal_end(text, cursor) + if regex_end is not None: + cursor = regex_end + continue + if char == "/" and next_char == "/": + comment_end = text.find("\n", cursor + 2) + comment_end = len(text) if comment_end < 0 else comment_end + for index in range(cursor, comment_end): + masked[index] = " " + cursor = comment_end + continue + if char == "/" and next_char == "*": + comment_end = text.find("*/", cursor + 2) + comment_end = len(text) if comment_end < 0 else comment_end + 2 + for index in range(cursor, comment_end): + if masked[index] not in "\r\n": + masked[index] = " " + cursor = comment_end + continue + if char in {'"', "'", "`"}: + quote = char + cursor += 1 + return "".join(masked) + + +def safe_javascript_config_path_literal( + text: str, + literal: re.Match[str], + *, + call_target: str, + context_start: int, + context_end: int, + argument_index: int, + javascript_dialect: str | None = None, +) -> bool: + if javascript_dialect is None: + return False + value = literal.group("value") + if CANONICAL_CONFIG_PATH_REFERENCE_PATTERN.fullmatch(value) is None: + return False + final_segment = value.rsplit(".", 1)[-1] + if CONFIG_PATH_CREDENTIAL_FIELD_PATTERN.search(final_segment) is None: + return False + normalized_target = call_target.replace("?.", ".").rsplit(".", 1)[-1] + secret_file_reader = normalized_target in { + "readCredentialFile", + "readCredentialFileSync", + "readSecretFile", + "readSecretFileSync", + "tryReadCredentialFile", + "tryReadCredentialFileSync", + "tryReadSecretFile", + "tryReadSecretFileSync", + } + quote_start = literal.start("value") - 1 + quote_end = literal.end("value") + 1 + argument_prefix = mask_javascript_comments(text[context_start:quote_start]) + property_match = re.search( + r"(?:^|[^A-Za-z0-9_$])(?PconfigPath|path)\s*:\s*$", + argument_prefix, + ) + if property_match is not None: + return ( + property_match.group("key") == "configPath" and secret_file_reader + ) or ( + property_match.group("key") == "path" + and normalized_target == "normalizeResolvedSecretInputString" + ) + return ( + secret_file_reader + and argument_index == 1 + and not text[context_start:quote_start].strip() + and not text[quote_end:context_end].strip() + ) + + +def mask_safe_javascript_config_path_literals( + argument: str, + call_target: str, + *, + argument_index: int, + javascript_dialect: str | None = None, +) -> str: + spans = [ + literal.span("value") + for pattern in SECRET_FALLBACK_LITERAL_PATTERNS + for literal in pattern.finditer(argument) + if safe_javascript_config_path_literal( + argument, + literal, + call_target=call_target, + context_start=0, + context_end=len(argument), + argument_index=argument_index, + javascript_dialect=javascript_dialect, + ) + ] + return redact_review_spans(argument, spans) + + def prompt_service_segment_is_secret_like(segment: str) -> bool: suffix = re.search(r"\d{4,}$", segment) if suffix is None: @@ -4671,10 +5400,16 @@ def call_arguments_risk( if public_risk: return True continue - if not safe_credential_lookup_argument( - call_target, argument, index - ) and fallback_secret_risk( + scanned_argument = mask_safe_javascript_config_path_literals( argument, + call_target, + argument_index=index, + javascript_dialect=javascript_dialect, + ) + if not safe_credential_lookup_argument( + call_target, scanned_argument, index + ) and fallback_secret_risk( + scanned_argument, minimum_length=12, javascript_dialect=javascript_dialect, ): @@ -4682,6 +5417,102 @@ def call_arguments_risk( return False +def truncated_call_arguments_risk( + arguments: str, + call_target: str, + *, + javascript_dialect: str | None = None, +) -> bool: + if call_arguments_risk( + arguments, + call_target, + javascript_dialect=javascript_dialect, + ): + return True + for argument in split_top_level_call_arguments(arguments): + match = SECRET_FALLBACK_UNQUOTED_PATTERN.fullmatch(argument.strip()) + if match is None: + continue + value = match.group("value") + if value.casefold() in SECRET_PLACEHOLDER_VALUES or any( + pattern.fullmatch(value) + for pattern in UNQUOTED_SECRET_REFERENCE_PATTERNS + ): + continue + if len(value) >= 7 and ( + any(character.isdigit() for character in value) + or re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", value) is None + ): + return True + return False + + +def balanced_delimiter_end( + text: str, + start: int, + opener: str, + closer: str, +) -> int | None: + depth = 0 + quote: str | None = None + escaped = False + line_comment = False + block_comment = False + index = start + while index < len(text): + char = text[index] + next_char = text[index + 1] if index + 1 < len(text) else "" + if line_comment: + if char == "\n": + line_comment = False + index += 1 + continue + if block_comment: + if char == "*" and next_char == "/": + block_comment = False + index += 2 + else: + index += 1 + continue + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + index += 1 + continue + regex_end = javascript_regex_literal_end(text, index) + if regex_end is not None: + index = regex_end + elif char == "/" and next_char == "/": + line_comment = True + index += 2 + elif char == "/" and next_char == "*": + block_comment = True + index += 2 + elif char == "#" and not javascript_private_member_marker(text, index): + line_comment = True + index += 1 + elif char in {'"', "'", "`"}: + quote = char + index += 1 + elif char == opener: + depth += 1 + index += 1 + elif char == closer: + depth -= 1 + if depth == 0: + return index + 1 + index += 1 + elif depth == 0: + return None + else: + index += 1 + return None + + def safe_secret_call_suffix( text: str, end: int, @@ -4692,73 +5523,31 @@ def safe_secret_call_suffix( if end >= len(text) or text[end] != "(": return False - def balanced_end(start: int, opener: str, closer: str) -> int | None: - depth = 0 - quote: str | None = None - escaped = False - line_comment = False - block_comment = False - index = start - while index < len(text): - char = text[index] - next_char = text[index + 1] if index + 1 < len(text) else "" - if line_comment: - if char == "\n": - line_comment = False - index += 1 - continue - if block_comment: - if char == "*" and next_char == "/": - block_comment = False - index += 2 - else: - index += 1 - continue - if quote is not None: - if escaped: - escaped = False - elif char == "\\": - escaped = True - elif char == quote: - quote = None - index += 1 - continue - regex_end = javascript_regex_literal_end(text, index) - if regex_end is not None: - index = regex_end - elif char == "/" and next_char == "/": - line_comment = True - index += 2 - elif char == "/" and next_char == "*": - block_comment = True - index += 2 - elif char == "#" and not javascript_private_member_marker( - text, - index, - ): - line_comment = True - index += 1 - elif char in {'"', "'", "`"}: - quote = char - index += 1 - elif char == opener: - depth += 1 - index += 1 - elif char == closer: - depth -= 1 - if depth == 0: - return index + 1 - index += 1 - elif depth == 0: - return None - else: - index += 1 - return None - def safe_call_end(start: int, target: str) -> int | None: - cursor = balanced_end(start, "(", ")") + cursor = balanced_delimiter_end(text, start, "(", ")") if cursor is None: - return None + if javascript_dialect is None: + return None + boundary = re.search(r"(?m)^;\r?$", text[start + 1 :]) + visible_end = ( + start + 1 + boundary.start() + if boundary is not None + else len(text) + ) + visible_arguments = text[start + 1 : visible_end] + return ( + None + if any( + pattern.search(visible_arguments) + for pattern in SECRET_FALLBACK_LITERAL_PATTERNS + ) + or truncated_call_arguments_risk( + visible_arguments, + target, + javascript_dialect=javascript_dialect, + ) + else visible_end + ) arguments = text[start + 1 : cursor - 1] return ( None @@ -4812,7 +5601,7 @@ def safe_secret_call_suffix( chained_target = "" continue if call_start < len(text) and text[call_start] == "[": - cursor = balanced_end(call_start, "[", "]") + cursor = balanced_delimiter_end(text, call_start, "[", "]") if cursor is None: return False chained_target = "" @@ -5149,16 +5938,173 @@ def basic_authorization_risk(text: str) -> bool: return False +def safe_self_reference_assignment( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> bool: + if javascript_dialect is None: + return False + value = ( + match.group("reference_value") + or match.group("call_value") + or match.group("bare_value") + ) + if value is None: + return False + key = re.split(r"\s*[:=]\s*", match.group(0), maxsplit=1)[0].strip("\"'") + if value != key: + return False + separator = re.search(r"[:=]", match.group(0)) + if separator is None or separator.group(0) != "=": + return False + line_end_candidates = [ + position + for position in ( + text.find("\n", match.end()), + text.find("\r", match.end()), + ) + if position >= 0 + ] + line_end = min(line_end_candidates, default=len(text)) + if not safe_secret_assignment_suffix( + text[:line_end], + match.end(), + javascript_dialect=javascript_dialect, + ): + return False + return safe_javascript_reference_suffix( + text, + line_end, + typescript=javascript_dialect == "typescript", + ) + + +def unsafe_multiline_self_reference_assignment( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> bool: + if javascript_dialect is None: + return False + value = ( + match.group("reference_value") + or match.group("call_value") + or match.group("bare_value") + ) + if value is None: + return False + key = re.split(r"\s*[:=]\s*", match.group(0), maxsplit=1)[0].strip("\"'") + separator = re.search(r"[:=]", match.group(0)) + if value != key or separator is None or separator.group(0) != "=": + return False + line_end_candidates = [ + position + for position in ( + text.find("\n", match.end()), + text.find("\r", match.end()), + ) + if position >= 0 + ] + line_end = min(line_end_candidates, default=len(text)) + return line_end < len(text) and not safe_javascript_reference_suffix( + text, + line_end, + typescript=javascript_dialect == "typescript", + ) + + +def boolean_declaration_initializer_range( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> tuple[int, int] | None: + initializer = re.match(r"\s*=\s*", text[match.end() :]) + if initializer is None: + return None + start = match.end() + initializer.end() + expression = fallback_expression( + text[start:], + typescript=javascript_dialect == "typescript", + ) + return start, start + len(expression) + + +def boolean_declaration_initializer_risk( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> bool: + initializer_range = boolean_declaration_initializer_range( + text, + match, + javascript_dialect=javascript_dialect, + ) + if initializer_range is None: + return False + start, end = initializer_range + return secret_literal_risk( + text[start:end], + minimum_length=8, + javascript_dialect=javascript_dialect, + ) + + +def boolean_declaration_initializer_spans( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + initializer_range = boolean_declaration_initializer_range( + text, + match, + javascript_dialect=javascript_dialect, + ) + if initializer_range is None: + return [] + start, end = initializer_range + if not secret_literal_risk( + text[start:end], + minimum_length=8, + javascript_dialect=javascript_dialect, + ): + return [] + return top_level_fallback_value_spans(text, start, end) + + def secret_text_risk( text: str, *, javascript_dialect: str | None = None, ) -> bool: + if DIFF_HUNK_CONTENT_BOUNDARY in text: + return any( + secret_text_risk(segment, javascript_dialect=javascript_dialect) + for segment in text.split(DIFF_HUNK_CONTENT_BOUNDARY) + ) uri_authorities = uri_authority_ranges(text) if credentialed_uri_risk(text, uri_authorities) or basic_authorization_risk(text) or any( pattern.search(text) for pattern in SECRET_VALUE_PATTERNS ): return True + boolean_declarations = tuple(SECRET_BOOLEAN_DECLARATION_PATTERN.finditer(text)) + boolean_declaration_positions = { + match.start("boolean_key") for match in boolean_declarations + } + if any( + boolean_declaration_initializer_risk( + text, + match, + javascript_dialect=javascript_dialect, + ) + for match in boolean_declarations + ): + return True safe_uri_credentials = interpolated_empty_password_uri_ranges( text, uri_authorities, @@ -5180,6 +6126,21 @@ def secret_text_risk( else frozenset() ) for prefix in assignment_prefixes: + if prefix.start() in boolean_declaration_positions: + continue + assignment = SECRET_ASSIGNMENT_PATTERN.match(text, prefix.start()) + if assignment is not None and safe_self_reference_assignment( + text, + assignment, + javascript_dialect=javascript_dialect, + ): + continue + if assignment is not None and unsafe_multiline_self_reference_assignment( + text, + assignment, + javascript_dialect=javascript_dialect, + ): + return True fallback = top_level_fallback_suffix( text[prefix.end() :], allow_chained_assignment=( @@ -5198,6 +6159,8 @@ def secret_text_risk( assignment_scan_text, safe_uri_credentials, ): + if match.start() in boolean_declaration_positions: + continue quoted = any( match.group(name) is not None for name in ("double_value", "single_value", "backtick_value") @@ -5216,6 +6179,12 @@ def secret_text_risk( separator_match = re.search(r"[:=]", match.group(0)) assert separator_match is not None separator = separator_match.group(0) + if safe_self_reference_assignment( + text, + match, + javascript_dialect=javascript_dialect, + ): + continue if ( key.strip("\"'").lower() == "credentials" and value.lower() in FETCH_CREDENTIAL_MODE_VALUES @@ -5365,22 +6334,904 @@ def require_no_secret_values( ) +def assignment_fallback_literal_spans( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + line_end_candidates = [ + position + for position in ( + text.find("\n", match.end()), + text.find("\r", match.end()), + ) + if position >= 0 + ] + line_end = min(line_end_candidates, default=len(text)) + expression_end = line_end + if ( + javascript_dialect is not None + and line_end < len(text) + and not safe_javascript_reference_suffix( + text, + line_end, + typescript=javascript_dialect == "typescript", + ) + ): + continued = fallback_expression( + text[line_end:], + typescript=javascript_dialect == "typescript", + ) + expression_end = line_end + len(continued) + fallback_start = match.end() + if match.group("call_value") is not None: + whitespace = re.match(r"[ \t]*", text[fallback_start:expression_end]) + assert whitespace is not None + call_start = fallback_start + whitespace.end() + if call_start < line_end and text[call_start] == "(": + depth = 0 + quote: str | None = None + escaped = False + cursor = call_start + while cursor < expression_end: + char = text[cursor] + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + elif char in {'"', "'", "`"}: + quote = char + elif char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + fallback_start = cursor + 1 + break + cursor += 1 + operator_span = top_level_fallback_operator_span( + text, + fallback_start, + expression_end, + ) + if operator_span is None: + return [] + expression_start = operator_span[1] + depth = 0 + quote: str | None = None + escaped = False + cursor = expression_start + while cursor < line_end: + char = text[cursor] + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + elif char in {'"', "'", "`"}: + quote = char + elif char in "([{": + depth += 1 + elif char in ")]}": + if depth == 0: + expression_end = cursor + break + depth -= 1 + elif depth == 0 and char in ";,": + expression_end = cursor + break + cursor += 1 + return top_level_fallback_value_spans( + text, + expression_start, + expression_end, + include_nested=True, + ) + + +def top_level_fallback_operator_span( + text: str, + start: int, + end: int, +) -> tuple[int, int] | None: + stack: list[str] = [] + pairs = {"(": ")", "[": "]", "{": "}"} + quote: str | None = None + escaped = False + line_comment = False + block_comment = False + cursor = start + while cursor < end: + char = text[cursor] + next_char = text[cursor + 1] if cursor + 1 < end else "" + if line_comment: + if char in "\r\n": + line_comment = False + cursor += 1 + continue + if block_comment: + if char == "*" and next_char == "/": + block_comment = False + cursor += 2 + else: + cursor += 1 + continue + if quote is not None: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == quote: + quote = None + cursor += 1 + continue + if char == "/" and next_char == "/": + line_comment = True + cursor += 2 + continue + if char == "/" and next_char == "*": + block_comment = True + cursor += 2 + continue + if char == "#": + line_comment = True + cursor += 1 + continue + if char in {'"', "'", "`"}: + quote = char + cursor += 1 + continue + if char in pairs: + stack.append(pairs[char]) + cursor += 1 + continue + if stack and char == stack[-1]: + stack.pop() + cursor += 1 + continue + if not stack and text.startswith(("||", "??"), cursor): + return cursor, cursor + 2 + if not stack and text.startswith("or", cursor): + before = text[cursor - 1] if cursor > start else "" + after = text[cursor + 2] if cursor + 2 < end else "" + if not (before.isalnum() or before == "_") and not ( + after.isalnum() or after == "_" + ): + return cursor, cursor + 2 + cursor += 1 + return None + + +def top_level_fallback_value_spans( + text: str, + start: int, + end: int, + *, + include_nested: bool = False, +) -> list[tuple[int, int]]: + spans: list[tuple[int, int]] = [] + stack: list[str] = [] + pairs = {"(": ")", "[": "]", "{": "}"} + cursor = start + while cursor < end: + char = text[cursor] + next_char = text[cursor + 1] if cursor + 1 < end else "" + if char == "/" and next_char == "/": + newline = re.search(r"[\r\n]", text[cursor + 2 : end]) + if newline is None: + break + cursor += 2 + newline.end() + continue + if char == "/" and next_char == "*": + comment_end = text.find("*/", cursor + 2, end) + if comment_end < 0: + break + cursor = comment_end + 2 + continue + if char == "#": + newline = re.search(r"[\r\n]", text[cursor + 1 : end]) + if newline is None: + break + cursor += 1 + newline.end() + continue + if char in {'"', "'", "`"}: + quote = char + value_start = cursor + 1 + cursor = value_start + escaped = False + while cursor < end: + current = text[cursor] + if escaped: + escaped = False + elif current == "\\": + escaped = True + elif current == quote: + value = text[value_start:cursor] + repeatable_nested_value = ( + include_nested + and value.casefold() not in SECRET_PLACEHOLDER_VALUES + and len(value) >= 7 + and any(character.isdigit() for character in value) + and not any( + pattern.fullmatch(value) + for pattern in QUOTED_SECRET_REFERENCE_PATTERNS + ) + ) + if cursor > value_start and ( + not stack or repeatable_nested_value + ): + spans.append((value_start, cursor)) + cursor += 1 + break + cursor += 1 + continue + if char in pairs: + stack.append(pairs[char]) + cursor += 1 + continue + if stack and char == stack[-1]: + stack.pop() + cursor += 1 + continue + if not stack: + bare = SECRET_FALLBACK_UNQUOTED_PATTERN.match(text, cursor, end) + if bare is not None: + value = bare.group("value") + if len(value) >= 7 and ( + any(candidate.isdigit() for candidate in value) + or re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", value) is None + ): + spans.append(bare.span("value")) + cursor = bare.end() + continue + cursor += 1 + return sorted(set(spans)) + + +def review_call_literal_spans( + text: str, + match: re.Match[str], + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + whitespace = re.match(r"[ \t]*", text[match.end() :]) + assert whitespace is not None + call_start = match.end() + whitespace.end() + if call_start >= len(text) or text[call_start] != "(": + return [] + call_end = balanced_delimiter_end(text, call_start, "(", ")") + if call_end is None or not secret_text_risk( + text[match.start() : call_end], + javascript_dialect=javascript_dialect, + ): + return [] + candidates = sorted( + ( + literal + for pattern in SECRET_FALLBACK_LITERAL_PATTERNS + for literal in pattern.finditer(text, call_start + 1, call_end - 1) + ), + key=lambda literal: literal.start("value"), + ) + argument_ranges: list[tuple[int, int]] = [] + argument_start = call_start + 1 + for argument in split_top_level_call_arguments( + text[call_start + 1 : call_end - 1] + ): + argument_end = argument_start + len(argument) + argument_ranges.append((argument_start, argument_end)) + argument_start = argument_end + 1 + candidates_with_argument_index: list[tuple[re.Match[str], int]] = [] + argument_index = 0 + for literal in candidates: + while ( + argument_index < len(argument_ranges) + and argument_ranges[argument_index][1] <= literal.start("value") + ): + argument_index += 1 + matched_index = ( + argument_index + if argument_index < len(argument_ranges) + and argument_ranges[argument_index][0] + <= literal.start("value") + < argument_ranges[argument_index][1] + else -1 + ) + candidates_with_argument_index.append((literal, matched_index)) + return [ + literal.span("value") + for literal, argument_index in candidates_with_argument_index + if len(literal.group("value")) >= 7 + and any(char.isalpha() for char in literal.group("value")) + and literal.group("value").casefold() not in SECRET_PLACEHOLDER_VALUES + and not any( + pattern.fullmatch(literal.group("value")) + for pattern in QUOTED_SECRET_REFERENCE_PATTERNS + ) + and not safe_javascript_config_path_literal( + text, + literal, + call_target=match.group("call_value") or "", + context_start=( + argument_ranges[argument_index][0] + if 0 <= argument_index < len(argument_ranges) + else call_start + 1 + ), + context_end=( + argument_ranges[argument_index][1] + if 0 <= argument_index < len(argument_ranges) + else call_end - 1 + ), + argument_index=argument_index, + javascript_dialect=javascript_dialect, + ) + ] + + +def review_repeatable_secret_spans( + text: str, + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + boolean_declarations = tuple(SECRET_BOOLEAN_DECLARATION_PATTERN.finditer(text)) + boolean_declaration_positions = { + match.start("boolean_key") for match in boolean_declarations + } + spans = [ + span + for match in boolean_declarations + for span in boolean_declaration_initializer_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + ] + for match in SECRET_ASSIGNMENT_PATTERN.finditer(text): + if match.start() in boolean_declaration_positions: + continue + selected_name = next( + ( + name + for name in ( + "double_value", + "single_value", + "backtick_value", + "reference_value", + "call_value", + "bare_value", + ) + if match.group(name) is not None + ), + None, + ) + if selected_name is None: + continue + if safe_self_reference_assignment( + text, + match, + javascript_dialect=javascript_dialect, + ): + continue + fallback_spans = ( + assignment_fallback_literal_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + if selected_name in {"reference_value", "call_value", "bare_value"} + else [] + ) + if fallback_spans: + spans.extend(fallback_spans) + continue + if selected_name == "call_value": + spans.extend( + review_call_literal_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + ) + continue + if secret_text_risk( + match.group(0), + javascript_dialect=javascript_dialect, + ): + spans.append(match.span(selected_name)) + for pattern in SECRET_VALUE_PATTERNS: + spans.extend( + match.span("credential") + if pattern is BEARER_CREDENTIAL_PATTERN + else match.span() + for match in pattern.finditer(text) + ) + spans.extend( + match.span("credential") + for match in BASIC_AUTHORIZATION_PATTERN.finditer(text) + ) + for authority_range in uri_authority_ranges(text): + credential = uri_authority_credential(text, authority_range) + if ( + credential is not None + and credential.value + and uri_credential_value_risk(text, credential) + ): + spans.append((credential.value_start, credential.value_end)) + return spans + + +def review_secret_value_spans( + text: str, + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + boolean_declarations = tuple(SECRET_BOOLEAN_DECLARATION_PATTERN.finditer(text)) + boolean_declaration_positions = { + match.start("boolean_key") for match in boolean_declarations + } + spans = [ + span + for match in boolean_declarations + for span in boolean_declaration_initializer_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + ] + for match in SECRET_ASSIGNMENT_PATTERN.finditer(text): + if match.start() in boolean_declaration_positions: + continue + for name in ( + "double_value", + "single_value", + "backtick_value", + "reference_value", + "call_value", + "bare_value", + ): + if match.group(name) is not None: + if safe_self_reference_assignment( + text, + match, + javascript_dialect=javascript_dialect, + ): + break + fallback_spans = ( + assignment_fallback_literal_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + if name in {"reference_value", "call_value", "bare_value"} + else [] + ) + if fallback_spans: + spans.extend(fallback_spans) + break + if name == "call_value": + spans.extend( + review_call_literal_spans( + text, + match, + javascript_dialect=javascript_dialect, + ) + ) + break + if secret_text_risk( + match.group(0), + javascript_dialect=javascript_dialect, + ): + spans.append(match.span(name)) + break + for pattern in SECRET_VALUE_PATTERNS: + spans.extend( + match.span("credential") + if pattern is BEARER_CREDENTIAL_PATTERN + else match.span() + for match in pattern.finditer(text) + ) + spans.extend( + match.span("credential") + for match in BASIC_AUTHORIZATION_PATTERN.finditer(text) + ) + for authority_range in uri_authority_ranges(text): + credential = uri_authority_credential(text, authority_range) + if ( + credential is not None + and credential.value + and uri_credential_value_risk(text, credential) + ): + spans.append((credential.value_start, credential.value_end)) + return spans + + +def redact_review_spans(text: str, spans: list[tuple[int, int]]) -> str: + merged: list[tuple[int, int]] = [] + for start, end in sorted(spans): + if start >= end: + continue + if merged and start <= merged[-1][1]: + merged[-1] = (merged[-1][0], max(end, merged[-1][1])) + else: + merged.append((start, end)) + if not merged: + return text + parts: list[str] = [] + cursor = 0 + for start, end in merged: + parts.extend((text[cursor:start], "redacted")) + cursor = end + parts.append(text[cursor:]) + return "".join(parts) + + +def known_secret_fragment_pattern(fragments: list[str]) -> re.Pattern[str] | None: + if not fragments: + return None + alternatives = "|".join(re.escape(fragment) for fragment in fragments) + return re.compile(rf"(?=(?P{alternatives}))") + + +def known_secret_fragment_spans( + text: str, + pattern: re.Pattern[str] | None, +) -> list[tuple[int, int]]: + if pattern is None: + return [] + return [match.span("fragment") for match in pattern.finditer(text)] + + +def javascript_regex_literal_ranges(text: str) -> list[tuple[int, int]]: + ranges: list[tuple[int, int]] = [] + cursor = 0 + while cursor < len(text): + end = javascript_regex_literal_end(text, cursor) + if end is None: + cursor += 1 + continue + ranges.append((cursor, end)) + cursor = end + return ranges + + +def repeated_secret_fragment_spans( + text: str, + pattern: re.Pattern[str] | None, + *, + javascript_dialect: str | None = None, +) -> list[tuple[int, int]]: + matches = known_secret_fragment_spans(text, pattern) + contexts = string_contexts_at(text, {start for start, _ in matches}) + regex_ranges = ( + javascript_regex_literal_ranges(text) + if javascript_dialect is not None + else [] + ) + regex_index = 0 + spans: list[tuple[int, int]] = [] + for start, end in matches: + while ( + regex_index < len(regex_ranges) + and regex_ranges[regex_index][1] <= start + ): + regex_index += 1 + if ( + regex_index < len(regex_ranges) + and regex_ranges[regex_index][0] <= start + and end <= regex_ranges[regex_index][1] + ): + spans.append((start, end)) + continue + token_start = start + while token_start > 0 and re.match(r"[A-Za-z0-9_$.-]", text[token_start - 1]): + token_start -= 1 + token_end = end + while token_end < len(text) and re.match(r"[A-Za-z0-9_$.-]", text[token_end]): + token_end += 1 + key_position = re.match( + r"\s*(?:=(?!=|>)|:(?![:=]))", + text[token_end:], + ) is not None + unquoted_token_substring = ( + contexts[start] is None + and (token_start < start or end < token_end) + ) + unquoted_identifier = ( + contexts[start] is None + and token_start == start + and token_end == end + and re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$.-]*", text[start:end]) + is not None + ) + if contexts[start] is None and ( + key_position or unquoted_token_substring or unquoted_identifier + ): + continue + spans.append((start, end)) + return spans + + +def private_key_body_spans(text: str) -> list[tuple[int, int]]: + matches = list(PRIVATE_KEY_BODY_PATTERN.finditer(text)) + contexts = string_contexts_at(text, {match.start() for match in matches}) + wrapper_parts: list[str] = [] + cursor = 0 + for match in matches: + wrapper_parts.append(text[cursor : match.start()]) + cursor = match.end() + wrapper_parts.append(text[cursor:]) + body_only_line = re.fullmatch( + r"(?:(?:#|//|/\*|\*)\s*)?[\s\"'`,;+\[\]]*", + "".join(wrapper_parts).strip(), + ) is not None + return [ + match.span() + for match in matches + if body_only_line or contexts[match.start()] is not None + ] + + +def private_key_token_spans(text: str) -> list[tuple[int, int]]: + escaped_spans = [ + match.span("body") + for match in ESCAPED_PRIVATE_KEY_BODY_PATTERN.finditer(text) + ] + ordinary_spans = [ + match.span() + for match in PRIVATE_KEY_BODY_PATTERN.finditer(text) + if not ( + match.start() > 0 + and text[match.start() - 1] == "\\" + and match.group(0)[:1] in {"n", "r"} + ) + and not any( + start < match.end() and match.start() < end + for start, end in escaped_spans + ) + ] + return sorted(escaped_spans + ordinary_spans) + + +def normalized_private_key_fragment(text: str, start: int, end: int) -> str: + fragment = text[start:end] + if start > 0 and text[start - 1] == "\\" and fragment[:1] in {"n", "r"}: + return fragment[1:] + return fragment + + +def likely_private_key_token(value: str) -> bool: + encoded = value.rstrip("=") + if not encoded: + return False + entropy = -sum( + (frequency := encoded.count(char) / len(encoded)) * math.log2(frequency) + for char in set(encoded) + ) + return ( + (len(encoded) >= 48 or encoded.startswith("MII")) + and any(char.islower() for char in encoded) + and any(char.isupper() for char in encoded) + and any(char.isdigit() or char in "+/=" for char in value) + and entropy >= 4.25 + ) + + +def wrapped_private_key_body_matches(text: str) -> list[re.Match[str]]: + matches = list(PRIVATE_KEY_BODY_PATTERN.finditer(text)) + if not matches: + return [] + wrapper_parts: list[str] = [] + cursor = 0 + for match in matches: + wrapper_parts.append(text[cursor : match.start()]) + cursor = match.end() + wrapper_parts.append(text[cursor:]) + if re.fullmatch( + r"(?:(?:#|//|/\*|\*)\s*)?[\s\\\"'`,;+\[\]]*" + r"(?:\*/[\s\\\"'`,;+\[\]]*)?", + "".join(wrapper_parts).strip(), + ) is None: + return [] + return matches + + +def orphan_private_key_body_spans(text: str) -> list[tuple[int, int]]: + matches = wrapped_private_key_body_matches(text) + if not matches: + return [] + values = [match.group(0) for match in matches] + raw_encoded = "".join(values) + encoded = raw_encoded.rstrip("=") + if not encoded: + return [] + entropy = -sum( + (frequency := encoded.count(char) / len(encoded)) * math.log2(frequency) + for char in set(encoded) + ) + mixed_short_chunks = ( + len(encoded) >= 20 + and len(matches) >= 2 + and all( + any(char.isdigit() for char in value) + and any(char.isupper() or char in "+/" for char in value) + for value in values + ) + and any(char.islower() for char in encoded) + and entropy >= 4.25 + ) + long_base64_line = len(matches) == 1 and likely_private_key_token(raw_encoded) + if not long_base64_line and not mixed_short_chunks: + return [] + return [match.span() for match in matches] + + +def bare_source_identifier_span(text: str, start: int, end: int) -> bool: + if re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", text[start:end]) is None: + return False + before = text[:start].rstrip()[-1:] + after = text[end:].lstrip()[:1] + return before in {"=", "+", "(", "[", "{", ",", ":"} and after in { + "+", + ")", + "]", + "}", + ",", + ";", + ":", + } + + +def explicit_private_key_body_spans( + text: str, + *, + in_pem: bool, +) -> list[tuple[int, int]]: + if not in_pem: + return [] + candidates = private_key_token_spans(text) + if not candidates: + return [] + wrapper_parts: list[str] = [] + cursor = 0 + for start, end in candidates: + wrapper_parts.append(text[cursor:start]) + cursor = end + wrapper_parts.append(text[cursor:]) + wrapper_text = "".join(wrapper_parts).strip() + escaped_body_only = ("\\n" in wrapper_text or "\\r" in wrapper_text) and re.fullmatch( + r"(?:(?:\\[nr])|\s)*", + wrapper_text, + ) is not None + if escaped_body_only: + return candidates + contexts = string_contexts_at(text, {start for start, _ in candidates}) + wrapped = {match.span() for match in wrapped_private_key_body_matches(text)} + return [ + (start, end) + for start, end in candidates + if not bare_source_identifier_span(text, start, end) + and ( + (start, end) in wrapped + or contexts[start] is not None + or any(char.isdigit() or char in "+/=" for char in text[start:end]) + ) + ] + + +def markerless_private_key_body_spans(text: str) -> list[tuple[int, int]]: + spans = orphan_private_key_body_spans(text) + matches = list(PRIVATE_KEY_BODY_PATTERN.finditer(text)) + contexts = string_contexts_at(text, {match.start() for match in matches}) + spans.extend( + match.span() + for match in matches + if contexts[match.start()] is not None + and likely_private_key_token(match.group(0)) + ) + return sorted(set(spans)) + + +def pem_line_body_spans( + text: str, + in_pem: bool, +) -> tuple[list[tuple[int, int]], bool]: + markers = sorted( + [ + (match.start(), match.end(), True) + for match in PRIVATE_KEY_BEGIN_PATTERN.finditer(text) + ] + + [ + (match.start(), match.end(), False) + for match in PRIVATE_KEY_END_PATTERN.finditer(text) + ] + ) + if not markers and not in_pem: + return markerless_private_key_body_spans(text), False + spans: list[tuple[int, int]] = [] + cursor = 0 + for start, end, begins_pem in markers: + if in_pem: + spans.extend( + (cursor + body_start, cursor + body_end) + for body_start, body_end in explicit_private_key_body_spans( + text[cursor:start], + in_pem=in_pem, + ) + ) + else: + spans.extend( + (cursor + body_start, cursor + body_end) + for body_start, body_end in markerless_private_key_body_spans( + text[cursor:start] + ) + ) + if begins_pem: + if in_pem: + raise SystemExit("refusing review bundle with nested private-key BEGIN markers") + in_pem = True + else: + if not in_pem: + raise SystemExit( + "refusing review bundle with a private-key END marker but no visible BEGIN" + ) + in_pem = False + cursor = end + if in_pem: + spans.extend( + (cursor + body_start, cursor + body_end) + for body_start, body_end in explicit_private_key_body_spans( + text[cursor:], + in_pem=in_pem, + ) + ) + else: + spans.extend( + (cursor + body_start, cursor + body_end) + for body_start, body_end in markerless_private_key_body_spans(text[cursor:]) + ) + return spans, in_pem + + +def private_key_body_fragments(text: str) -> set[str]: + # Do not join short markerless chunks across lines. Without PEM boundaries, + # code and data are indistinguishable; explicit markers or long tokens redact. + fragments: set[str] = set() + in_private_key = False + for line in text.split("\n"): + spans, in_private_key = pem_line_body_spans(line, in_private_key) + fragments.update( + fragment + for start, end in spans + if (fragment := normalized_private_key_fragment(line, start, end)) + ) + if in_private_key: + raise SystemExit("refusing review bundle with an unterminated private-key block") + return fragments + + def unified_diff_contents(patch: str) -> tuple[str, str]: old_content: list[str] = [] new_content: list[str] = [] in_hunk = False prefix_columns = 1 - for line in patch.splitlines(): + for line in patch.split("\n"): + line = line.removesuffix("\r") hunk_header = re.match(r"^(@{2,})", line) if hunk_header: - old_content.append(";") - new_content.append(";") + old_content.append(DIFF_HUNK_CONTENT_BOUNDARY) + new_content.append(DIFF_HUNK_CONTENT_BOUNDARY) in_hunk = True prefix_columns = len(hunk_header.group(1)) - 1 continue if line.startswith("diff --"): - old_content.append(";") - new_content.append(";") + old_content.append(DIFF_HUNK_CONTENT_BOUNDARY) + new_content.append(DIFF_HUNK_CONTENT_BOUNDARY) in_hunk = False continue prefix = line[:prefix_columns] @@ -5399,30 +7250,9 @@ def unified_diff_contents(patch: str) -> tuple[str, str]: return "\n".join(old_content), "\n".join(new_content) -def unified_diff_metadata(patch: str) -> str: - metadata: list[str] = [] - in_hunk = False - prefix_columns = 1 - for line in patch.splitlines(): - hunk_header = re.match(r"^(@{2,})", line) - if hunk_header: - metadata.append(line) - in_hunk = True - prefix_columns = len(hunk_header.group(1)) - 1 - continue - if line.startswith("diff --"): - metadata.append(line) - in_hunk = False - continue - prefix = line[:prefix_columns] - hunk_content = ( - in_hunk - and len(prefix) == prefix_columns - and set(prefix) <= {"+", "-", " "} - ) - if not hunk_content: - metadata.append(line) - return "\n".join(metadata) +REVIEW_SECURITY_REDACTION = ( + "[security-sensitive review material omitted before model review]" +) def sensitive_repo_path_risk(rel: str) -> str | None: @@ -5444,6 +7274,7 @@ def sensitive_repo_path_risk(rel: str) -> str | None: if ( any(pattern.search(normalized) for pattern in SENSITIVE_NAME_PATTERNS) and not design_token_artifact_path(path, SENSITIVE_NAME_PATTERNS) + and not github_workflow_path(path) ): return "sensitive filename" return None @@ -5480,6 +7311,14 @@ def design_token_artifact_path( ) +def github_workflow_path(path: Path) -> bool: + return ( + path.parts[:2] == (".github", "workflows") + and len(path.parts) == 3 + and path.suffix in {".yml", ".yaml"} + ) + + def credential_store_path(normalized: str) -> bool: path = Path(normalized) credential_directory = any( @@ -5538,6 +7377,7 @@ def tracked_sensitive_repo_path_risk(rel: str) -> str | None: if ( any(pattern.search(normalized) for pattern in TRACKED_SENSITIVE_NAME_PATTERNS) and not design_token_artifact_path(path, TRACKED_SENSITIVE_NAME_PATTERNS) + and not github_workflow_path(path) ): return "sensitive filename" return None @@ -5619,58 +7459,144 @@ def diff_section_paths(section: str) -> tuple[str | None, str | None]: return old_path, new_path +def tracked_sensitive_paths(paths: list[str]) -> set[str]: + return { + rel + for rel in paths + if tracked_sensitive_repo_path_risk(rel) is not None + } + + +def omit_tracked_sensitive_diff_units( + patch: str, + paths: list[str], + blocked_paths: set[str], +) -> str: + if not blocked_paths: + return patch + units = review_bundle_units(patch) + diff_indexes = [ + index for index, unit in enumerate(units) if unit.startswith("diff --git ") + ] + if len(diff_indexes) != len(paths): + return REVIEW_SECURITY_REDACTION + "\n" + path_by_unit = dict(zip(diff_indexes, paths)) + retained = [ + unit + for index, unit in enumerate(units) + if path_by_unit.get(index) not in blocked_paths + ] + retained.insert(0, REVIEW_SECURITY_REDACTION + "\n") + return "".join(retained) + + +def redact_review_patch_metadata(patch: str) -> str: + redacted: list[str] = [] + metadata: list[str] = [] + in_hunk = False + prefix_columns = 1 + old_remaining: list[int] = [] + new_remaining = 0 + + def flush_metadata() -> None: + if not metadata: + return + text = "".join(metadata) + redacted.append( + REVIEW_SECURITY_REDACTION + "\n" + if secret_text_risk(text) + else text + ) + metadata.clear() + + def parse_range_count(token: str) -> int | None: + match = re.fullmatch(r"[+-]\d+(?:,(\d+))?", token) + if match is None: + return None + return int(match.group(1) or "1") + + for line in literal_lf_lines(patch): + body = line.rstrip("\r\n") + if body.startswith("diff --"): + flush_metadata() + metadata.append(line) + in_hunk = False + prefix_columns = 1 + old_remaining = [] + new_remaining = 0 + continue + hunk_header = re.match( + r"^(?P@{2,}) (?P.+?) (?P=marker)(?: .*)?$", + body, + ) + if hunk_header: + flush_metadata() + metadata.append(line) + marker = hunk_header.group("marker") + ranges = hunk_header.group("ranges").split() + old_counts = [ + count + for token in ranges + if token.startswith("-") + and (count := parse_range_count(token)) is not None + ] + new_counts = [ + count + for token in ranges + if token.startswith("+") + and (count := parse_range_count(token)) is not None + ] + prefix_columns = len(marker) - 1 + in_hunk = ( + len(old_counts) == prefix_columns + and len(new_counts) == 1 + ) + old_remaining = old_counts + new_remaining = new_counts[0] if new_counts else 0 + continue + prefix = body[:prefix_columns] + hunk_content = ( + in_hunk + and len(prefix) == prefix_columns + and set(prefix) <= {"+", "-", " "} + ) + if hunk_content: + flush_metadata() + redacted.append(line) + for index, marker in enumerate(prefix): + if marker != "+": + old_remaining[index] = max(0, old_remaining[index] - 1) + if any(marker != "-" for marker in prefix): + new_remaining = max(0, new_remaining - 1) + if new_remaining == 0 and not any(old_remaining): + in_hunk = False + else: + metadata.append(line) + flush_metadata() + return "".join(redacted) + + def validate_review_patch( label: str, paths: list[str], patch: str, limit: int | None = None, ) -> str: - blocked = [ - f"{display_escape(rel, 500)} ({risk})" - for rel in paths - if (risk := tracked_sensitive_repo_path_risk(rel)) is not None - ] - if blocked: - details = "\n".join(f"- {item}" for item in blocked[:20]) - more = f"\n... {len(blocked) - 20} more" if len(blocked) > 20 else "" - raise SystemExit( - f"refusing to include tracked sensitive paths in {label}:\n" - f"{details}{more}" - ) patch_bytes = len(patch.encode("utf-8")) if limit is not None and patch_bytes > limit: raise SystemExit( f"{label} is too large to review safely " f"({patch_bytes} bytes; limit {limit}); split the change into smaller review targets" ) - require_no_secret_values(label, unified_diff_metadata(patch)) - sections = [ - unit - for unit in review_bundle_units(patch) - if unit.startswith("diff --git ") - ] - if sections: - expected_paths = set(paths) - for section in sections: - old_path, new_path = diff_section_paths(section) - old_content, new_content = unified_diff_contents(section) - for rel, content in ( - (old_path, old_content), - (new_path, new_content), - ): - javascript_dialect = ( - javascript_review_dialect(rel) - if rel is not None and rel in expected_paths - else None - ) - require_no_secret_values( - f"{label} {rel or ''}", - content, - javascript_dialect=javascript_dialect, - ) - else: - for content in unified_diff_contents(patch): - require_no_secret_values(label, content) + blocked_paths = tracked_sensitive_paths(paths) + patch = omit_tracked_sensitive_diff_units(patch, paths, blocked_paths) + patch = redact_review_patch_metadata(patch) + patch_bytes = len(patch.encode("utf-8")) + if limit is not None and patch_bytes > limit: + raise SystemExit( + f"{label} is too large to review safely after metadata redaction " + f"({patch_bytes} bytes; limit {limit}); split the change into smaller review targets" + ) return patch @@ -5779,15 +7705,12 @@ def file_bundle_snapshot( text = data.decode("utf-8") except UnicodeDecodeError: return "", True, "non-UTF-8 file" - if secret_text_risk( - text, - javascript_dialect=javascript_review_dialect(rel), - ): - return "", True, "secret-like content" return text, False, None -def safe_untracked_file_snapshots(repo: Path) -> list[tuple[str, str, bool]]: +def collect_untracked_file_snapshots( + repo: Path, +) -> tuple[list[tuple[str, str, bool]], int]: files = git_path_list( repo, *global_excludes_git_args(repo), @@ -5796,7 +7719,7 @@ def safe_untracked_file_snapshots(repo: Path) -> list[tuple[str, str, bool]]: "--exclude-standard", "-z", ) - blocked: list[str] = [] + omitted = 0 included: list[tuple[str, str, bool]] = [] for rel in files: content, truncated, risk = file_bundle_snapshot( @@ -5806,31 +7729,48 @@ def safe_untracked_file_snapshots(repo: Path) -> list[tuple[str, str, bool]]: allow_binary_omission=True, ) if risk: - blocked.append(f"{display_escape(rel, 500)} ({risk})") + if ( + sensitive_repo_path_risk(rel) is not None + or risk + in { + "secret-like content", + "symlink", + "path outside repository", + } + ): + omitted += 1 + else: + raise SystemExit( + "cannot safely include untracked file " + f"{display_escape(rel, 500)}: {risk}" + ) else: included.append((rel, content, truncated)) - if blocked: - details = "\n".join(f"- {item}" for item in blocked[:20]) - more = f"\n... {len(blocked) - 20} more" if len(blocked) > 20 else "" - raise SystemExit( - "refusing to include untracked sensitive files in review bundle; " - "stage, ignore, remove, or redact them before running autoreview:\n" - f"{details}{more}" - ) - return included + return included, omitted + + +def safe_untracked_file_snapshots(repo: Path) -> list[tuple[str, str, bool]]: + snapshots, _omitted = collect_untracked_file_snapshots(repo) + return snapshots def safe_untracked_files(repo: Path) -> list[str]: return [rel for rel, _content, _truncated in safe_untracked_file_snapshots(repo)] -def local_status(repo: Path, untracked: list[str]) -> str: +def local_status(repo: Path, untracked: list[str], *, redact: bool = False) -> str: + if redact: + return REVIEW_SECURITY_REDACTION status = git(repo, "status", "--short", "--untracked-files=no").rstrip() lines = [status] if status else [] lines.extend(f"?? {rel}" for rel in untracked) return "\n".join(lines) +def redact_secret_like_review_metadata(text: str) -> str: + return REVIEW_SECURITY_REDACTION if secret_text_risk(text) else text + + def local_bundle(repo: Path) -> tuple[str, bool]: staged_patch = git(repo, "diff", *SAFE_DIFF_FLAGS, "--cached", "--patch") unstaged_patch = git(repo, "diff", *SAFE_DIFF_FLAGS, "--patch") @@ -5865,22 +7805,57 @@ def local_bundle(repo: Path) -> tuple[str, bool]: "--name-only", "-z", ) - untracked_snapshots = safe_untracked_file_snapshots(repo) + untracked_snapshots, omitted_untracked = collect_untracked_file_snapshots(repo) untracked = [rel for rel, _content, _truncated in untracked_snapshots] - if not staged_patch.strip() and not unstaged_patch.strip() and not untracked: + omitted_tracked = len( + tracked_sensitive_paths(staged_paths) | tracked_sensitive_paths(unstaged_paths) + ) + if ( + not staged_patch.strip() + and not unstaged_patch.strip() + and not untracked + and not omitted_untracked + ): raise SystemExit("no local changes to review") staged_patch = validate_review_patch("local staged diff", staged_paths, staged_patch) unstaged_patch = validate_review_patch("local unstaged diff", unstaged_paths, unstaged_patch) parts = [ "# Git Status", - local_status(repo, untracked), + redact_secret_like_review_metadata( + local_status( + repo, + untracked, + redact=bool(omitted_tracked or omitted_untracked), + ) + ), "# Staged Diff", - git(repo, "diff", *SAFE_DIFF_FLAGS, "--cached", "--stat"), + ( + REVIEW_SECURITY_REDACTION + if tracked_sensitive_paths(staged_paths) + else redact_secret_like_review_metadata( + git(repo, "diff", *SAFE_DIFF_FLAGS, "--cached", "--stat") + ) + ), staged_patch, "# Unstaged Diff", - git(repo, "diff", *SAFE_DIFF_FLAGS, "--stat"), + ( + REVIEW_SECURITY_REDACTION + if tracked_sensitive_paths(unstaged_paths) + else redact_secret_like_review_metadata( + git(repo, "diff", *SAFE_DIFF_FLAGS, "--stat") + ) + ), unstaged_patch, ] + if omitted_tracked or omitted_untracked: + parts[0:0] = [ + "# Review Input Redactions", + REVIEW_SECURITY_REDACTION, + ( + f"Omitted tracked changes: {omitted_tracked}; " + f"omitted untracked files: {omitted_untracked}." + ), + ] input_truncated = False if untracked: parts.append("# Untracked Files") @@ -6093,18 +8068,29 @@ def branch_bundle(repo: Path, base_ref: str) -> tuple[str, bool]: diff_range, ), ) + omitted_tracked = bool(tracked_sensitive_paths(branch_paths)) branch_patch = validate_review_patch("branch diff", branch_paths, branch_patch) return "\n\n".join( [ "# Branch Diff", - f"base: {base_ref}", - git( - repo, - "diff", - *SAFE_DIFF_FLAGS, - "--stat", - "--end-of-options", - diff_range, + ( + REVIEW_SECURITY_REDACTION + if secret_text_risk(base_ref) + else f"base: {base_ref}" + ), + ( + REVIEW_SECURITY_REDACTION + if omitted_tracked + else redact_secret_like_review_metadata( + git( + repo, + "diff", + *SAFE_DIFF_FLAGS, + "--stat", + "--end-of-options", + diff_range, + ) + ) ), branch_patch, ] @@ -6164,20 +8150,28 @@ def commit_bundle(repo: Path, commit_ref: str) -> tuple[str, bool]: commit_ref, ), ) + omitted_tracked = bool(tracked_sensitive_paths(commit_paths)) commit_patch = validate_review_patch("commit diff", commit_paths, commit_patch) + commit_summary = git( + repo, + "show", + *SAFE_DIFF_FLAGS, + "--stat", + "--format=fuller", + "--end-of-options", + commit_ref, + ) + if omitted_tracked or secret_text_risk(commit_summary): + commit_summary = REVIEW_SECURITY_REDACTION return "\n\n".join( [ "# Commit Diff", - f"commit: {commit_ref}", - git( - repo, - "show", - *SAFE_DIFF_FLAGS, - "--stat", - "--format=fuller", - "--end-of-options", - commit_ref, + ( + REVIEW_SECURITY_REDACTION + if secret_text_risk(commit_ref) + else f"commit: {commit_ref}" ), + commit_summary, commit_patch, ] ), False @@ -6643,11 +8637,15 @@ def render_review_prompt( datasets: str, chunk_position: tuple[int, int] | None = None, ) -> str: - target_line = f"{target} {target_ref}" if target_ref else target + safe_target_ref = ( + REVIEW_SECURITY_REDACTION + if target_ref and secret_text_risk(target_ref) + else target_ref + ) + target_line = f"{target} {safe_target_ref}" if safe_target_ref else target branch = current_branch(repo) - require_no_secret_values("current branch", branch) - if target_ref: - require_no_secret_values("review target ref", target_ref) + if secret_text_risk(branch): + branch = REVIEW_SECURITY_REDACTION scope_policy = review_scope_policy() chunk_policy = "" if chunk_position: @@ -6684,6 +8682,7 @@ def render_review_prompt( - Before returning, sweep the bundle once more for independent defects in other files or failure modes that you may have stopped scanning for after an earlier find. - Include security findings: injection, secret leaks, authz/authn bypass, path traversal, unsafe deserialization, unsafe filesystem or shell use, privacy leaks, and credential handling. - Do not reject legitimate functionality merely because it touches shell, filesystem, network, auth, or sensitive data. Report a security finding only when the patch creates a concrete exploitable risk, removes an important safety check, or lacks validation at a trust boundary. + - Security-sensitive bundle material may be redacted or omitted before review. Continue reviewing the material that is present. A redaction notice is not itself a defect and does not prove either safety or vulnerability in the omitted material. - For each finding, use the smallest file/line location that demonstrates the issue. - If there are no actionable findings, return an empty findings array and mark the patch correct. @@ -9879,6 +11878,7 @@ def main() -> int: return 0 review_source_snapshot = source_tree_snapshot(repo) + run_trufflehog_preflight(repo, target, target_ref, args.commit) if target == "local": bundle, bundle_truncated = local_bundle(repo) elif target == "branch": diff --git a/.agents/skills/autoreview/scripts/autoreview_test.py b/.agents/skills/autoreview/scripts/autoreview_test.py index 2648d5e9d0d8..f3cbc20292cc 100644 --- a/.agents/skills/autoreview/scripts/autoreview_test.py +++ b/.agents/skills/autoreview/scripts/autoreview_test.py @@ -102,6 +102,80 @@ class AutoreviewCursorTests(unittest.TestCase): self.assertIn("review engine result was not structured JSON", str(exc_info.exception)) +class AutoreviewSecretScannerTests(unittest.TestCase): + def test_boolean_declarations_are_not_credential_material(self) -> None: + secret_field = "is" + "Secret" + client_secret_field = "hasClient" + "Secret" + cases = ( + (f"val {secret_field}: Boolean? = null,", None), + (f"var {client_secret_field}: Boolean = false", None), + (f"abstract val {secret_field}: Boolean?", None), + (f"val {secret_field}: Boolean?", None), + (f"const {client_secret_field}: boolean = true;", "typescript"), + (f"declare const {client_secret_field}: boolean;", "typescript"), + (f"let {secret_field}: Bool? = nil", None), + (f"let {secret_field}: Bool?", None), + ) + + for content, javascript_dialect in cases: + with self.subTest(content=content): + self.assertFalse( + AUTOREVIEW.secret_text_risk( + content, + javascript_dialect=javascript_dialect, + ) + ) + + def test_boolean_and_null_literal_values_are_not_credentials(self) -> None: + cases = ( + ("is" + "Secret", "true"), + ("requires" + "Password", "false"), + ("access" + "Token", "null"), + ) + for field_name, literal in cases: + content = f"{field_name} = {literal}" + with self.subTest(content=content): + self.assertFalse(AUTOREVIEW.secret_text_risk(content)) + + def test_boolean_annotation_does_not_hide_real_credential_literal(self) -> None: + literal_value = "actual-production-" + "secret" + secret_field = "is" + "Secret" + client_secret_field = "hasClient" + "Secret" + cases = ( + (f'val {secret_field}: Boolean? = "{literal_value}",', None), + (f'var {client_secret_field}: Boolean = "{literal_value}"', None), + ( + f'const {client_secret_field}: boolean = "{literal_value}";', + "typescript", + ), + (f'let {secret_field}: Bool? = "{literal_value}"', None), + ) + + for content, javascript_dialect in cases: + with self.subTest(content=content): + self.assertTrue( + AUTOREVIEW.secret_text_risk( + content, + javascript_dialect=javascript_dialect, + ) + ) + + def test_boolean_prefix_values_remain_credentials(self) -> None: + field_name = "client" + "Secret" + for prefix in ("Boolean", "boolean", "Bool"): + literal_value = prefix + "-prod-credential" + content = f"{field_name}: {literal_value}" + with self.subTest(content=content): + self.assertTrue(AUTOREVIEW.secret_text_risk(content)) + + def test_boolean_type_tokens_in_config_remain_credentials(self) -> None: + field_name = "client" + "Secret" + for literal_value in ("Boolean?", "Boolean?=abc1234"): + content = f"{field_name}: {literal_value}" + with self.subTest(content=content): + self.assertTrue(AUTOREVIEW.secret_text_risk(content)) + + class AutoreviewCompatibilityTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: @@ -553,8 +627,13 @@ class AutoreviewCompatibilityTests(unittest.TestCase): source.write_text("after\n") cursor_bin = root / "cursor-agent" + trufflehog_bin = root / "trufflehog" record_path = root / "record.json" AUTOREVIEW.write_executable(cursor_bin, AUTOREVIEW.fake_cursor_script()) + AUTOREVIEW.write_executable( + trufflehog_bin, + "#!/usr/bin/env python3\nraise SystemExit(0)\n", + ) env = os.environ.copy() env.update( { @@ -563,7 +642,10 @@ class AutoreviewCompatibilityTests(unittest.TestCase): "GIT_CONFIG_GLOBAL": str(root / "hostile-gitconfig"), "NODE_OPTIONS": "--require=hostile.js", "PYTHONPATH": str(root / "hostile-python"), - "PATH": f"{repo}{os.pathsep}{env.get('PATH', '')}", + "PATH": ( + f"{root}{os.pathsep}{repo}{os.pathsep}" + f"{env.get('PATH', '')}" + ), "HOME": str(root), "USERPROFILE": str(root), } diff --git a/.agents/skills/autoreview/tests/fixtures/typescript-benign-config-path-references.ts b/.agents/skills/autoreview/tests/fixtures/typescript-benign-config-path-references.ts new file mode 100644 index 000000000000..1b45f94e09b7 --- /dev/null +++ b/.agents/skills/autoreview/tests/fixtures/typescript-benign-config-path-references.ts @@ -0,0 +1,30 @@ +declare const accountId: string; +declare const filePath: string; +declare const secretRef: string; +declare const tryReadSecretFileSync: (...args: unknown[]) => string; +declare const normalizeResolvedSecretInputString: (options: unknown) => string; + +export const passwordFile = tryReadSecretFileSync(filePath, "IRC password file", { + credentialDiagnostic: { + configPath: `channels.irc.accounts.${accountId}.passwordFile`, + }, +}); +export const nickservFile = tryReadSecretFileSync(filePath, "IRC NickServ password file", { + credentialDiagnostic: { + configPath: `channels.irc.accounts.${accountId}.nickserv.passwordFile`, + }, +}); +export const botSecret = normalizeResolvedSecretInputString({ + value: secretRef, + path: `channels.nextcloud-talk.accounts.${accountId}.botSecret`, +}); +export const botSecretFile = tryReadSecretFileSync(filePath, "Nextcloud bot secret file", { + credentialDiagnostic: { + configPath: `channels.nextcloud-talk.accounts.${accountId}.botSecretFile`, + }, +}); +export const tokenFile = tryReadSecretFileSync( + filePath, + `channels.telegram.accounts.${accountId}.tokenFile`, + { rejectSymlink: true }, +); diff --git a/.agents/skills/autoreview/tests/fixtures/typescript-benign-references.ts b/.agents/skills/autoreview/tests/fixtures/typescript-benign-references.ts new file mode 100644 index 000000000000..924bdd7ab5fb --- /dev/null +++ b/.agents/skills/autoreview/tests/fixtures/typescript-benign-references.ts @@ -0,0 +1,55 @@ +type SecretRef = { source: "env"; id: string }; +type CredentialUnavailableDiagnostic = { path: string; reason: string }; + +declare const tokenRef: SecretRef; +declare const keyRef: SecretRef; +declare const inlinePassword: string; +declare const inlineSecret: string; +declare const accountFileToken: string; +declare const baseFileToken: string; +declare const passwordResolution: { password: string }; +declare const secretResolution: { secret: string }; +declare const tokenResolution: { token: string }; +declare const accountTokenFile: { token: string }; +declare const channelTokenFile: { token: string }; +declare const merged: { apiPassword: string; passwordFile: string }; +declare const tryReadSecretFileSync: (...args: unknown[]) => string; +declare const normalizeResolvedSecretInputString: (options: unknown) => string; +declare const resolveToken: (options: unknown) => { value: string }; + +const filePassword = tryReadSecretFileSync(merged.passwordFile, "IRC password file", { + credentialDiagnostic: { + configPath: `channels.irc.accounts.${accountId}.passwordFile`, + report: (diagnostic: CredentialUnavailableDiagnostic) => diagnostic, + }, +}); +const configPassword = normalizeResolvedSecretInputString({ + value: merged.apiPassword, + path: "channels.nextcloud-talk.apiPassword", +}); +const token = resolveToken({ accountId }); +const priorPasswordFileError = /IRC password file.*must not be a symlink/; + +export type CredentialPlumbing = { + tokenRef?: SecretRef; + keyRef?: SecretRef; + credentialDiagnostics?: CredentialUnavailableDiagnostic[]; +}; + +export const resolvedCredentialPlumbing = { + token: tokenRef, + apiKey: keyRef, + password: filePassword, + configPassword, + nextPassword: inlinePassword, + secret: inlineSecret, + accountToken: accountFileToken, + baseToken: baseFileToken, + resolvedPassword: passwordResolution.password, + resolvedSecret: secretResolution.secret, + resolvedToken: tokenResolution.token, + accountTokenFile: accountTokenFile.token, + channelTokenFile: channelTokenFile.token, + apiPassword: merged.apiPassword, + channelAccessToken: token.value, +}; diff --git a/.agents/skills/autoreview/tests/fixtures/typescript-sensitive-literals.ts b/.agents/skills/autoreview/tests/fixtures/typescript-sensitive-literals.ts new file mode 100644 index 000000000000..1fb677b71144 --- /dev/null +++ b/.agents/skills/autoreview/tests/fixtures/typescript-sensitive-literals.ts @@ -0,0 +1,10 @@ +const password = "FAKE-CorrectHorseBattery-Staple-2026!"; +const credential = "FAKE_A7f9K2m4Q8v6N3x5R1p0T9z8"; +const apiKey = "sk-proj-FAKE00000000000000000000000000000000000000000000"; +const githubToken = "ghp_FAKE000000000000000000000000000000"; +const awsAccessKey = "AKIAFAKE000000000000"; +const slackToken = "xoxb-FAKE000000000-FAKE000000000-FAKE000000000000000000000000"; +const authorization = "Bearer eyJhbGciOiJIUzI1NiJ9.RkFLRS1OT1QtQS1SRUFM.TOKENFAKESIGNATURE"; +const resolvedToken = resolveToken({ value: "FAKE_B8g0L3n5R9w7P4y6S2q1U0a9" }); +const filePassword = tryReadSecretFileSync(path, "FAKE-A7f9K2m4Q8v6N3x5R1p0T9z8"); +const password = readPassword("alice", "FAKE correct horse secret battery 2026"); diff --git a/.agents/skills/autoreview/tests/test_autoreview_hardening.py b/.agents/skills/autoreview/tests/test_autoreview_hardening.py index 95ef20e22073..d0863a5906c6 100644 --- a/.agents/skills/autoreview/tests/test_autoreview_hardening.py +++ b/.agents/skills/autoreview/tests/test_autoreview_hardening.py @@ -21,6 +21,9 @@ from pathlib import Path SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "autoreview" +FIXTURES = Path(__file__).with_name("fixtures") +PRIVATE_KEY_BEGIN_TEXT = "BEGIN " + "PRIVATE KEY" +RSA_PRIVATE_KEY_BEGIN_TEXT = "BEGIN RSA " + "PRIVATE KEY" def load_helper() -> dict[str, object]: @@ -62,10 +65,451 @@ def realistic_secret_value() -> str: return "A7f9K2m4Q8v6" + "N3x5R1p0T9z8" +def add_fake_trufflehog( + helper: dict[str, object], + root: Path, + env: dict[str, str], +) -> None: + helper["write_executable"]( + root / "trufflehog", + "#!/usr/bin/env python3\nraise SystemExit(0)\n", + ) + env["PATH"] = f"{root}{os.pathsep}{env.get('PATH', '')}" + + class AutoreviewHardeningTests(unittest.TestCase): def setUp(self) -> None: self.helper = load_helper() + def test_trufflehog_missing_binary_has_platform_neutral_guidance(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + with mock.patch.dict( + self.helper["run_trufflehog_preflight"].__globals__, + {"find_command": lambda _name, _repo: None}, + ): + with self.assertRaises(SystemExit) as error: + self.helper["run_trufflehog_preflight"]( + repo, + "local", + None, + "HEAD", + ) + + message = str(error.exception) + self.assertIn("TruffleHog is required but was not found", message) + self.assertIn(self.helper["TRUFFLEHOG_INSTALL_URL"], message) + self.assertNotIn("brew", message.casefold()) + + def test_trufflehog_scans_staged_and_working_versions_separately(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + source = repo / "runtime.txt" + source.write_text("base\n", encoding="utf-8") + git(repo, "add", "runtime.txt") + git(repo, "commit", "-q", "-m", "base") + source.write_text("staged version\n", encoding="utf-8") + git(repo, "add", "runtime.txt") + source.write_text("working version\n", encoding="utf-8") + + original_find_command = self.helper["find_command"] + original_run = self.helper["run"] + scanned: dict[str, str] = {} + + def find_command(name: str, checkout: Path) -> str | None: + if name == "trufflehog": + return "/trusted/trufflehog" + return original_find_command(name, checkout) + + def run_scanner( + command: list[str], + cwd: Path, + **_kwargs: object, + ) -> subprocess.CompletedProcess[str]: + if command[0] != "/trusted/trufflehog": + return original_run(command, cwd, **_kwargs) + self.assertEqual( + command[0:2], + [ + "/trusted/trufflehog", + "git", + ], + ) + self.assertEqual(command[3], "--since-commit") + self.assertEqual(command[5:7], ["--branch", "HEAD"]) + self.assertEqual( + command[7:], + [ + "--no-update", + "--no-color", + "--results=verified,unknown", + "--fail", + "--fail-on-scan-errors", + ], + ) + scan_path = command[2].removeprefix("file://") + if os.name == "nt": + scan_path = scan_path.lstrip("/") + scan_repo = Path(scan_path) + commits = git( + scan_repo, + "log", + "--reverse", + "--format=%H", + ).splitlines() + scanned["staged"] = git( + scan_repo, + "show", + f"{commits[1]}:runtime.txt", + ) + scanned["working"] = git( + scan_repo, + "show", + f"{commits[2]}:runtime.txt", + ) + return subprocess.CompletedProcess(command, 0, "", "") + + with mock.patch.dict( + self.helper["run_trufflehog_preflight"].__globals__, + { + "find_command": find_command, + "run": run_scanner, + }, + ): + self.helper["run_trufflehog_preflight"]( + repo, + "local", + None, + "HEAD", + ) + + self.assertEqual( + scanned, + { + "staged": "staged version\n", + "working": "working version\n", + }, + ) + + def test_trufflehog_scans_only_changed_content_at_reviewed_ref(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + unchanged = repo / "unchanged.txt" + changed = repo / "changed.txt" + unchanged.write_text("unchanged\n", encoding="utf-8") + changed.write_text("base\n", encoding="utf-8") + git(repo, "add", "unchanged.txt", "changed.txt") + git(repo, "commit", "-q", "-m", "base") + base = git(repo, "rev-parse", "HEAD").strip() + changed.write_text("reviewed version\n", encoding="utf-8") + git(repo, "add", "changed.txt") + git(repo, "commit", "-q", "-m", "change") + reviewed_commit = git(repo, "rev-parse", "HEAD").strip() + changed.write_text("later working version\n", encoding="utf-8") + + for target, target_ref, commit_ref in ( + ("branch", base, "HEAD"), + ("commit", None, reviewed_commit), + ): + with self.subTest(target=target), tempfile.TemporaryDirectory() as scan_dir: + scan_repo = Path(scan_dir) + base_commit = self.helper["prepare_trufflehog_history"]( + repo, + target, + target_ref, + commit_ref, + scan_repo, + ) + + commits = git( + scan_repo, + "log", + "--reverse", + "--format=%H", + ).splitlines() + self.assertEqual(commits[0], base_commit) + self.assertEqual(len(commits), 3) + self.assertEqual( + git( + scan_repo, + "show", + f"{commits[1]}:changed.txt", + ), + "reviewed version\n", + ) + with self.assertRaises(subprocess.CalledProcessError): + subprocess.run( + [ + "git", + "show", + f"{commits[1]}:unchanged.txt", + ], + cwd=scan_repo, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + def test_trufflehog_history_scans_deleted_content_in_reverse_commit(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + source = repo / "removed.txt" + source.write_text("removed baseline content\n", encoding="utf-8") + git(repo, "add", "removed.txt") + git(repo, "commit", "-q", "-m", "base") + base = git(repo, "rev-parse", "HEAD").strip() + source.unlink() + git(repo, "add", "removed.txt") + git(repo, "commit", "-q", "-m", "remove credential") + + with tempfile.TemporaryDirectory() as scan_dir: + scan_repo = Path(scan_dir) + scan_base = self.helper["prepare_trufflehog_history"]( + repo, + "branch", + base, + "HEAD", + scan_repo, + ) + commits = git( + scan_repo, + "log", + "--reverse", + "--format=%H", + ).splitlines() + + self.assertEqual(commits[0], scan_base) + self.assertEqual(len(commits), 3) + self.assertEqual( + git(scan_repo, "show", f"{commits[2]}:removed.txt"), + "removed baseline content\n", + ) + with self.assertRaises(subprocess.CalledProcessError): + subprocess.run( + ["git", "show", f"{commits[1]}:removed.txt"], + cwd=scan_repo, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + def test_trufflehog_snapshot_supports_directory_to_file_transition(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + nested = repo / "entry" / "nested.txt" + nested.parent.mkdir() + nested.write_text("nested\n", encoding="utf-8") + git(repo, "add", "entry/nested.txt") + git(repo, "commit", "-q", "-m", "directory") + base = git(repo, "rev-parse", "HEAD").strip() + nested.unlink() + nested.parent.rmdir() + (repo / "entry").write_text("file\n", encoding="utf-8") + git(repo, "add", "-A") + git(repo, "commit", "-q", "-m", "file") + + with tempfile.TemporaryDirectory() as scan_dir: + scan_repo = Path(scan_dir) + self.helper["prepare_trufflehog_history"]( + repo, + "branch", + base, + "HEAD", + scan_repo, + ) + commits = git( + scan_repo, + "log", + "--reverse", + "--format=%H", + ).splitlines() + self.assertEqual( + git(scan_repo, "show", f"{commits[1]}:entry"), + "file\n", + ) + + @unittest.skipIf(os.name == "nt", "Windows filenames cannot use Git pathspec magic prefixes") + def test_trufflehog_snapshot_treats_git_paths_as_literals(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + rel = ":(exclude).txt" + source = repo / rel + source.write_text("base\n", encoding="utf-8") + git(repo, "--literal-pathspecs", "add", "--", rel) + git(repo, "commit", "-q", "-m", "base") + source.write_text("staged\n", encoding="utf-8") + git(repo, "--literal-pathspecs", "add", "--", rel) + + with tempfile.TemporaryDirectory() as index_dir: + index_root = Path(index_dir) + self.helper["materialize_index_snapshot"]( + repo, + index_root, + [rel], + ) + self.assertEqual( + (index_root / rel).read_text(encoding="utf-8"), + "staged\n", + ) + + git(repo, "commit", "-q", "-m", "staged") + with tempfile.TemporaryDirectory() as tree_dir: + tree_root = Path(tree_dir) + self.helper["materialize_tree_snapshot"]( + repo, + tree_root, + "HEAD", + [rel], + ) + self.assertEqual( + (tree_root / rel).read_text(encoding="utf-8"), + "staged\n", + ) + + def test_trufflehog_snapshot_force_stages_ignored_materialized_files(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + (repo / ".gitignore").write_text("ignored.txt\n", encoding="utf-8") + (repo / "ignored.txt").write_text("review me\n", encoding="utf-8") + + commit = self.helper["commit_snapshot"](repo, "snapshot") + + self.assertEqual( + git(repo, "show", f"{commit}:ignored.txt"), + "review me\n", + ) + + def test_trufflehog_snapshot_rejects_symlinked_parent_directories(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + root = Path(tempdir) + repo = init_repo(root) + outside = root / "outside" + outside.mkdir() + (outside / "review.txt").write_text("outside\n", encoding="utf-8") + parent = repo / "nested" + try: + parent.symlink_to(outside, target_is_directory=True) + except OSError as exc: + if os.name == "nt" and getattr(exc, "winerror", None) == 1314: + self.skipTest("Windows symlink privilege is not available") + raise + + with tempfile.TemporaryDirectory() as snapshot_dir: + with self.assertRaisesRegex(SystemExit, "symlinked parent"): + self.helper["copy_worktree_file"]( + repo, + Path(snapshot_dir), + "nested/review.txt", + ) + + def test_local_trufflehog_snapshot_supports_path_type_transitions(self) -> None: + for transition in ("directory-to-file", "file-to-directory"): + with self.subTest(transition=transition), tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + entry = repo / "entry" + nested = entry / "nested.txt" + if transition == "directory-to-file": + entry.mkdir() + nested.write_text("nested\n", encoding="utf-8") + git(repo, "add", "entry/nested.txt") + else: + entry.write_text("file\n", encoding="utf-8") + git(repo, "add", "entry") + git(repo, "commit", "-q", "-m", "base") + + if transition == "directory-to-file": + nested.unlink() + entry.rmdir() + entry.write_text("file\n", encoding="utf-8") + expected_path = "entry" + expected_content = "file\n" + else: + entry.unlink() + entry.mkdir() + nested.write_text("nested\n", encoding="utf-8") + expected_path = "entry/nested.txt" + expected_content = "nested\n" + + with tempfile.TemporaryDirectory() as scan_dir: + scan_repo = Path(scan_dir) + self.helper["prepare_trufflehog_history"]( + repo, + "local", + None, + "HEAD", + scan_repo, + ) + commits = git( + scan_repo, + "log", + "--reverse", + "--format=%H", + ).splitlines() + self.assertEqual( + git(scan_repo, "show", f"{commits[2]}:{expected_path}"), + expected_content, + ) + + def test_trufflehog_findings_and_errors_do_not_leak_scanner_output(self) -> None: + for returncode, expected in ( + ( + self.helper["TRUFFLEHOG_FINDINGS_EXIT_CODE"], + "found verified or unknown credentials", + ), + (1, "could not complete the credential scan"), + ): + with self.subTest(returncode=returncode), tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + (repo / "runtime.txt").write_text("review me\n", encoding="utf-8") + original_find_command = self.helper["find_command"] + original_run = self.helper["run"] + scanner_output = "detected-value-that-must-not-leak" + + def find_command(name: str, checkout: Path) -> str | None: + if name == "trufflehog": + return "/trusted/trufflehog" + return original_find_command(name, checkout) + + def run_scanner( + command: list[str], + cwd: Path, + **_kwargs: object, + ) -> subprocess.CompletedProcess[str]: + if command[0] != "/trusted/trufflehog": + return original_run(command, cwd, **_kwargs) + return subprocess.CompletedProcess( + command, + returncode, + scanner_output, + scanner_output, + ) + + output = io.StringIO() + with ( + mock.patch.dict( + self.helper["run_trufflehog_preflight"].__globals__, + { + "find_command": find_command, + "run": run_scanner, + }, + ), + contextlib.redirect_stdout(output), + contextlib.redirect_stderr(output), + self.assertRaises(SystemExit) as error, + ): + self.helper["run_trufflehog_preflight"]( + repo, + "local", + None, + "HEAD", + ) + + combined = output.getvalue() + str(error.exception) + self.assertIn(expected, combined) + self.assertNotIn(scanner_output, combined) + def test_powershell_harness_exposes_runnable_engines_only(self) -> None: harness = SCRIPT.with_name("test-review-harness.ps1").read_text(encoding="utf-8") @@ -73,16 +517,23 @@ class AutoreviewHardeningTests(unittest.TestCase): for disabled_engine in ("droid", "copilot", "opencode", "cursor"): self.assertNotIn(f"'{disabled_engine}'", harness) - def test_local_bundle_blocks_sensitive_untracked_file(self) -> None: + def test_local_bundle_omits_sensitive_untracked_file_without_blocking(self) -> None: for rel in (".env", "tokens/session.dat", "secrets/local.py"): with self.subTest(rel=rel), tempfile.TemporaryDirectory() as tempdir: repo = init_repo(Path(tempdir)) path = repo / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_text("placeholder=true\n", encoding="utf-8") + (repo / "review.py").write_text("print('review me')\n", encoding="utf-8") - with self.assertRaisesRegex(SystemExit, "untracked sensitive files"): - self.helper["local_bundle"](repo) + bundle, truncated = self.helper["local_bundle"](repo) + + self.assertIn("# Review Input Redactions", bundle) + self.assertIn(self.helper["REVIEW_SECURITY_REDACTION"], bundle) + self.assertNotIn(rel, bundle) + self.assertNotIn("placeholder=true", bundle) + self.assertIn("print('review me')", bundle) + self.assertFalse(truncated) def test_local_bundle_marks_untracked_binary_input_incomplete(self) -> None: with tempfile.TemporaryDirectory() as tempdir: @@ -706,71 +1157,147 @@ class AutoreviewHardeningTests(unittest.TestCase): "", ) - def test_review_patch_escapes_controls_in_blocked_paths(self) -> None: + def test_review_patch_does_not_disclose_controls_in_omitted_paths(self) -> None: path = ".env.\x1b]52;c;VEVTVA==\x07\udc9b" - with self.assertRaises(SystemExit) as raised: - self.helper["validate_review_patch"]( - "local staged diff", - [path], - "", - ) - - message = str(raised.exception) - self.assertNotIn("\x1b", message) - self.assertNotIn("\x07", message) - self.assertNotIn("\udc9b", message) - self.assertIn( - r".env.\x1b]52;c;VEVTVA==\x07\udc9b", - message, + redacted = self.helper["validate_review_patch"]( + "local staged diff", + [path], + "", ) - def test_review_patch_scans_reconstructed_content_not_diff_markers( + self.assertEqual( + redacted, + self.helper["REVIEW_SECURITY_REDACTION"] + "\n", + ) + self.assertNotIn("\x1b", redacted) + self.assertNotIn("\x07", redacted) + self.assertNotIn("\udc9b", redacted) + + def test_review_patch_omits_everything_when_sensitive_paths_cannot_be_mapped( self, ) -> None: patch = ( - "@@ -0,0 +1,4 @@\n" - '+ "https://token=" + "hardcoded123@host/repo",\n' - '+ "DATABASE_URL=https:"\n' - '+ + f"//token={literal_username}:${{PASSWORD}}@host",\n' - '+ \'curl "https:\'\n' + "commit metadata that must not survive a mapping failure\n" + "diff --cc .env\n" + "@@@ -1,1 -1,1 +1,1 @@@\n" + "++placeholder=true\n" ) - self.assertTrue(self.helper["secret_text_risk"](patch)) - self.assertFalse( - any( - self.helper["secret_text_risk"](line) - for line in patch.splitlines() - ) - ) - self.assertEqual( - self.helper["validate_review_patch"]( - "local unstaged diff", - ["safe.py"], - patch, - ), + redacted = self.helper["validate_review_patch"]( + "branch diff", + [".env"], patch, ) - def test_review_patch_scans_diff_metadata_line_by_line(self) -> None: - credential = "AKIA" + "ABCDEFGHIJKLMNOP" - patch = ( - f"diff --git a/{credential}.txt b/{credential}.txt\n" - "new file mode 100644\n" - "--- /dev/null\n" - f"+++ b/{credential}.txt\n" - "@@ -0,0 +1 @@\n" - "+public content\n" + self.assertEqual( + redacted, + self.helper["REVIEW_SECURITY_REDACTION"] + "\n", + ) + self.assertNotIn("placeholder", redacted) + self.assertNotIn("commit metadata", redacted) + + def test_review_metadata_redaction_is_independent_of_path_classification( + self, + ) -> None: + credential = "ghp_" + "A" * 24 + metadata = f" M {credential}.txt\n" + + self.assertEqual( + self.helper["redact_secret_like_review_metadata"](metadata), + self.helper["REVIEW_SECURITY_REDACTION"], + ) + self.assertNotIn( + credential, + self.helper["redact_secret_like_review_metadata"](metadata), ) - with self.assertRaisesRegex(SystemExit, "secret-like content"): + def test_review_patch_redacts_metadata_but_preserves_code_content(self) -> None: + patch = ( + "Authorization: Basic dXNlcjpwYXNzd29yZA==\n" + "diff --git a/src/runtime.ts b/src/runtime.ts\n" + "--- a/src/runtime.ts\n" + "+++ b/src/runtime.ts\n" + "@@ -0,0 +1 @@\n" + '+const token = "ordinary-hardcoded-value-12345";\n' + ) + + validated = self.helper["validate_review_patch"]( + "commit diff", + ["src/runtime.ts"], + patch, + ) + + self.assertNotIn("dXNlcjpwYXNzd29yZA==", validated) + self.assertIn("ordinary-hardcoded-value-12345", validated) + + def test_review_patch_redacts_standard_diff_paths_but_preserves_hunks(self) -> None: + credential = "ghp_" + "A" * 24 + patch = ( + f"diff --git a/{credential}.ts b/{credential}.ts\n" + f"--- a/{credential}.ts\n" + f"+++ b/{credential}.ts\n" + "@@ -0,0 +1 @@\n" + '+const token = "ordinary-hardcoded-value-12345";\n' + ) + + validated = self.helper["validate_review_patch"]( + "commit diff", + ["src/runtime.ts"], + patch, + ) + + self.assertNotIn(credential, validated) + self.assertIn("ordinary-hardcoded-value-12345", validated) + + def test_review_patch_preserves_combined_and_headerless_hunk_content(self) -> None: + credential_shaped_code = '+token = "ordinary-hardcoded-value-12345"\n' + for patch in ( + "@@ -0,0 +1 @@\n" + credential_shaped_code, + "diff --cc src/runtime.ts\n" + "@@@ -0,0 -0,0 +1 @@@\n" + "++token = \"ordinary-hardcoded-value-12345\"\n", + ): + with self.subTest(patch=patch): + validated = self.helper["validate_review_patch"]( + "commit diff", + ["src/runtime.ts"], + patch, + ) + self.assertIn("ordinary-hardcoded-value-12345", validated) + + def test_review_patch_stops_hunk_classification_at_declared_counts(self) -> None: + credential = "ghp_" + "A" * 24 + patch = ( + "@@ -0,0 +1 @@\n" + "+first file content\n" + f"--- a/{credential}.ts\n" + f"+++ b/{credential}.ts\n" + "@@ -0,0 +1 @@\n" + '+token = "ordinary-hardcoded-value-12345"\n' + ) + + validated = self.helper["validate_review_patch"]( + "commit diff", + ["first.ts", "second.ts"], + patch, + ) + + self.assertNotIn(credential, validated) + self.assertIn("ordinary-hardcoded-value-12345", validated) + + def test_review_patch_enforces_limit_after_metadata_redaction(self) -> None: + patch = "ghp_" + "A" * 24 + + with self.assertRaisesRegex(SystemExit, "after metadata redaction"): self.helper["validate_review_patch"]( - "local unstaged diff", - ["safe.txt"], + "commit diff", + [], patch, + 40, ) - def test_tracked_sensitive_paths_are_blocked_in_all_modes(self) -> None: + def test_tracked_sensitive_paths_are_omitted_in_all_modes(self) -> None: with tempfile.TemporaryDirectory() as tempdir: repo = init_repo(Path(tempdir)) (repo / "base.txt").write_text("base\n", encoding="utf-8") @@ -779,15 +1306,61 @@ class AutoreviewHardeningTests(unittest.TestCase): base = git(repo, "rev-parse", "HEAD").strip() (repo / ".env").write_text("placeholder=true\n", encoding="utf-8") - git(repo, "add", ".env") - with self.assertRaisesRegex(SystemExit, "tracked sensitive paths"): - self.helper["local_bundle"](repo) + (repo / "base.txt").write_text("base\nreview me\n", encoding="utf-8") + git(repo, "add", ".env", "base.txt") + local, local_truncated = self.helper["local_bundle"](repo) + self.assertIn(self.helper["REVIEW_SECURITY_REDACTION"], local) + self.assertNotIn(".env", local) + self.assertNotIn("placeholder=true", local) + self.assertIn("+review me", local) + self.assertFalse(local_truncated) git(repo, "commit", "-q", "-m", "sensitive path") - with self.assertRaisesRegex(SystemExit, "tracked sensitive paths"): - self.helper["branch_bundle"](repo, base) - with self.assertRaisesRegex(SystemExit, "tracked sensitive paths"): - self.helper["commit_bundle"](repo, "HEAD") + for bundle, truncated in ( + self.helper["branch_bundle"](repo, base), + self.helper["commit_bundle"](repo, "HEAD"), + ): + self.assertIn(self.helper["REVIEW_SECURITY_REDACTION"], bundle) + self.assertNotIn(".env", bundle) + self.assertNotIn("placeholder=true", bundle) + self.assertIn("+review me", bundle) + self.assertFalse(truncated) + + def test_secret_named_workflows_are_reviewable_in_all_modes(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + (repo / "base.txt").write_text("base\n", encoding="utf-8") + git(repo, "add", "base.txt") + git(repo, "commit", "-q", "-m", "base") + base = git(repo, "rev-parse", "HEAD").strip() + + workflow = repo / ".github" / "workflows" / "secret-scan.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text("name: Secret scan\n", encoding="utf-8") + untracked_bundle, _ = self.helper["local_bundle"](repo) + self.assertIn("secret-scan.yml", untracked_bundle) + + git(repo, "add", str(workflow.relative_to(repo))) + tracked_bundle, _ = self.helper["local_bundle"](repo) + self.assertIn("secret-scan.yml", tracked_bundle) + + git(repo, "commit", "-q", "-m", "add secret scanner") + branch_bundle, _ = self.helper["branch_bundle"](repo, base) + commit_bundle, _ = self.helper["commit_bundle"](repo, "HEAD") + self.assertIn("secret-scan.yml", branch_bundle) + self.assertIn("secret-scan.yml", commit_bundle) + + def test_case_variant_secret_named_workflows_remain_sensitive(self) -> None: + for rel in ( + ".GitHub/workflows/secret-scan.yml", + ".github/Workflows/secret-scan.yml", + ".github/workflows/secret-scan.YML", + ): + with self.subTest(rel=rel): + self.assertIsNotNone(self.helper["sensitive_repo_path_risk"](rel)) + self.assertIsNotNone( + self.helper["tracked_sensitive_repo_path_risk"](rel) + ) def test_tracked_source_names_and_env_templates_remain_reviewable(self) -> None: for rel in ( @@ -816,6 +1389,7 @@ class AutoreviewHardeningTests(unittest.TestCase): "token_count/generated.py", ".docker/Dockerfile", ".docker/scripts/build.sh", + ".github/workflows/secret-scan.yml", ): with self.subTest(rel=rel): self.assertIsNone(self.helper["tracked_sensitive_repo_path_risk"](rel)) @@ -832,6 +1406,19 @@ class AutoreviewHardeningTests(unittest.TestCase): with self.subTest(rel=rel): self.assertIsNone(self.helper["sensitive_repo_path_risk"](rel)) + def test_untracked_credential_shaped_source_content_is_reviewed(self) -> None: + with tempfile.TemporaryDirectory() as tempdir: + repo = init_repo(Path(tempdir)) + source = 'const token = "ordinary-hardcoded-value-12345";\n' + path = repo / "src" / "runtime.ts" + path.parent.mkdir() + path.write_text(source, encoding="utf-8") + + bundle, truncated = self.helper["local_bundle"](repo) + + self.assertIn("ordinary-hardcoded-value-12345", bundle) + self.assertFalse(truncated) + def test_untracked_design_token_artifacts_remain_reviewable(self) -> None: for rel in ( "design-tokens.json", @@ -1859,6 +2446,56 @@ class AutoreviewHardeningTests(unittest.TestCase): ) ) + def test_review_patch_allows_provider_references_and_test_placeholders( + self, + ) -> None: + token_name = "to" + "ken" + key_name = "api_" + "key" + secret_name = "api_" + "secret" + safe_patch = ( + "diff --git a/provider.ts b/provider.ts\n" + "--- a/provider.ts\n" + "+++ b/provider.ts\n" + "@@ -1 +1,6 @@\n" + f"-const {token_name} = data.session?.access_token;\n" + f"+const {token_name} = data.session?.access_token;\n" + "+const api" + f"Key = providerConfig.{key_name};\n" + "+const api" + "Sec" + f"ret = providerConfig.{secret_name};\n" + f'+const fixture = {{ {key_name}: "test-key" }};\n' + f'+const fixtureSecret = {{ {secret_name}: "test-secret" }};\n' + f'+const session = {{ access_{token_name}: "test-token" }};\n' + ) + + self.assertEqual( + self.helper["validate_review_patch"]( + "branch diff", + ["provider.ts"], + safe_patch, + ), + safe_patch, + ) + + def test_provider_reference_allowlist_still_rejects_real_credentials( + self, + ) -> None: + key_name = "api_" + "key" + literal_value = "actual-production-" + "secret" + structured_value = "ghp_" + "ActualToken1234567890" + unsafe_values = ( + f'const config = {{ {key_name}: "{literal_value}" }};', + f'const config = {{ {key_name}: "{structured_value}" }};', + f'const config = {{ {key_name}: "test-key-extra" }};', + ) + + for content in unsafe_values: + with self.subTest(content=content): + self.assertTrue( + self.helper["secret_text_risk"]( + content, + javascript_dialect="typescript", + ) + ) + def test_secret_detector_allows_typescript_object_secret_references(self) -> None: content = ( "async function configure(context: RuntimeContext) {\n" @@ -2310,6 +2947,175 @@ class AutoreviewHardeningTests(unittest.TestCase): ) ) + def test_secret_detector_allows_typescript_credential_plumbing_fixture(self) -> None: + source = (FIXTURES / "typescript-benign-references.ts").read_text( + encoding="utf-8" + ) + + patch = ( + "diff --git a/src/credential-plumbing.ts b/src/credential-plumbing.ts\n" + "new file mode 100644\n" + "--- /dev/null\n" + "+++ b/src/credential-plumbing.ts\n" + f"@@ -0,0 +1,{len(source.splitlines())} @@\n" + + "".join(f"+{line}\n" for line in source.splitlines()) + ) + validated = self.helper["validate_review_patch"]( + "typescript credential plumbing fixture", + ["src/credential-plumbing.ts"], + patch, + ) + for reference in ( + "filePassword", + "passwordResolution.password", + "tokenResolution.token", + "CredentialUnavailableDiagnostic", + "tokenRef", + "keyRef", + ): + self.assertIn(reference, validated) + + def test_secret_detector_allows_typescript_member_reference_assignment(self) -> None: + source = "legacyXSearchResolvedRecord.apiKey = resolution.value;" + patch = ( + "diff --git a/src/runtime-web-tools.ts b/src/runtime-web-tools.ts\n" + "--- a/src/runtime-web-tools.ts\n" + "+++ b/src/runtime-web-tools.ts\n" + "@@ -20,2 +20,3 @@ function resolveLegacySearch() {\n" + f" {source}\n" + "+const contractDigest = digestRuntimeWebOwnerContract(contract);\n" + ) + + self.assertFalse( + self.helper["secret_text_risk"]( + source, + javascript_dialect="typescript", + ) + ) + self.assertEqual( + self.helper["validate_review_patch"]( + "typescript member reference assignment", + ["src/runtime-web-tools.ts"], + patch, + ), + patch, + ) + + fake_literal = next( + line + for line in (FIXTURES / "typescript-sensitive-literals.ts") + .read_text(encoding="utf-8") + .splitlines() + if line.strip() + ) + self.assertTrue( + self.helper["secret_text_risk"]( + fake_literal, + javascript_dialect="typescript", + ) + ) + def test_review_bundle_preserves_typescript_config_paths(self) -> None: + source = (FIXTURES / "typescript-benign-config-path-references.ts").read_text( + encoding="utf-8" + ) + patch = ( + "diff --git a/src/config-path-references.ts b/src/config-path-references.ts\n" + "new file mode 100644\n" + "--- /dev/null\n" + "+++ b/src/config-path-references.ts\n" + f"@@ -0,0 +1,{len(source.splitlines())} @@\n" + + "".join(f"+{line}\n" for line in source.splitlines()) + ) + + validated = self.helper["validate_review_patch"]( + "typescript config path references", + ["src/config-path-references.ts"], + patch, + ) + + for config_path in ( + "channels.irc.accounts.${accountId}.passwordFile", + "channels.irc.accounts.${accountId}.nickserv.passwordFile", + "channels.nextcloud-talk.accounts.${accountId}.botSecret", + "channels.nextcloud-talk.accounts.${accountId}.botSecretFile", + "channels.telegram.accounts.${accountId}.tokenFile", + ): + self.assertIn(config_path, validated) + + token_term = "To" + "ken" + truncated_call_patch = ( + "diff --git a/src/token.ts b/src/token.ts\n" + "--- a/src/token.ts\n" + "+++ b/src/token.ts\n" + "@@ -40,3 +40,4 @@ function resolveAccountToken() {\n" + f"+ const account{token_term} = resolveRuntime{token_term}Value({{\n" + "+ value: accountConfig.token,\n" + "@@ -70,3 +71,4 @@ function resolveConfigToken() {\n" + f"+ const config{token_term} = resolveRuntime{token_term}Value({{\n" + "+ value: merged.token,\n" + ) + self.assertEqual( + self.helper["validate_review_patch"]( + "typescript truncated credential calls fixture", + ["src/token.ts"], + truncated_call_patch, + ), + truncated_call_patch, + ) + + def test_secret_detector_rejects_sensitive_literal_fixture_corpus(self) -> None: + source = (FIXTURES / "typescript-sensitive-literals.ts").read_text( + encoding="utf-8" + ) + corpus = [line for line in source.splitlines() if line.strip()] + + self.assertGreaterEqual(len(corpus), 7) + for literal_assignment in corpus: + with self.subTest(literal_assignment=literal_assignment): + self.assertTrue( + self.helper["secret_text_risk"]( + literal_assignment, + javascript_dialect="typescript", + ) + ) + truncated_literal = ( + "const incompleteToken = resolveToken({ value: \"" + + realistic_secret_value() + + "\";" + ) + self.assertTrue( + self.helper["secret_text_risk"]( + truncated_literal, + javascript_dialect="typescript", + ) + ) + truncated_short_literal = ( + "const incompleteToken = resolveToken({ value: \"" + + "short" + + "pwd" + + "\";" + ) + self.assertTrue( + self.helper["secret_text_risk"]( + truncated_short_literal, + javascript_dialect="typescript", + ) + ) + + def test_known_secret_fragment_scan_handles_many_javascript_regexes(self) -> None: + fragment = "password file" + regex_count = 2_000 + source = ";".join(f"/{fragment} {index}/" for index in range(regex_count)) + pattern = self.helper["known_secret_fragment_pattern"]([fragment]) + + spans = self.helper["repeated_secret_fragment_spans"]( + source, + pattern, + javascript_dialect="typescript", + ) + + self.assertEqual(len(spans), regex_count) + def test_lifecycle_reference_scan_is_bounded_for_non_matching_identifier(self) -> None: source = "const value = resolved" + "A" * 100_000 + "X;" @@ -2319,104 +3125,6 @@ class AutoreviewHardeningTests(unittest.TestCase): self.assertEqual(spans, frozenset()) self.assertLess(time.monotonic() - started, 5.0) - def test_review_patch_scopes_source_references_to_typescript_files(self) -> None: - property_name = "pass" + "word" - reference = "context.driverPass" + "word" - source_patch = ( - "diff --git a/src/runtime.ts b/src/runtime.ts\n" - "--- a/src/runtime.ts\n" - "+++ b/src/runtime.ts\n" - "@@ -0,0 +1 @@\n" - "+function configure(context: RuntimeContext) { return { " - + property_name - + ": " - + reference - + " }; }\n" - ) - narrow_source_patch = ( - "diff --git a/src/runtime.ts b/src/runtime.ts\n" - "--- a/src/runtime.ts\n" - "+++ b/src/runtime.ts\n" - "@@ -40,2 +40,3 @@ function configure(context: RuntimeContext) {\n" - " return {\n" - "+ " - + property_name - + ": " - + reference - + ",\n" - " };\n" - ) - config_patch = ( - "diff --git a/config.yml b/config.yml\n" - "--- a/config.yml\n" - "+++ b/config.yml\n" - "@@ -0,0 +1 @@\n" - "+" - + property_name - + ": " - + reference - + "\n" - ) - - self.assertEqual( - self.helper["validate_review_patch"]( - "local staged diff", - ["src/runtime.ts"], - source_patch, - ), - source_patch, - ) - self.assertEqual( - self.helper["validate_review_patch"]( - "local staged diff", - ["src/runtime.ts"], - narrow_source_patch, - ), - narrow_source_patch, - ) - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["validate_review_patch"]( - "local staged diff", - ["src/runtime.ts", "config.yml"], - source_patch + config_patch, - ) - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["validate_review_patch"]( - "local staged diff", - ["config.yml", "src/runtime.ts"], - source_patch + config_patch, - ) - - def test_review_patch_scans_rename_sides_with_their_own_file_types(self) -> None: - property_name = "pass" + "word" - reference = "context.driverPass" + "word" - patch = ( - "diff --git a/src/runtime.ts b/config.yml\n" - "similarity index 80%\n" - "rename from src/runtime.ts\n" - "rename to config.yml\n" - "--- a/src/runtime.ts\n" - "+++ b/config.yml\n" - "@@ -1 +1 @@\n" - "-function configure(context: RuntimeContext) { return { " - + property_name - + ": " - + reference - + " }; }\n" - "+" - + property_name - + ": " - + reference - + "\n" - ) - - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["validate_review_patch"]( - "branch diff", - ["src/runtime.ts", "config.yml"], - patch, - ) - def test_review_patch_decodes_git_quoted_source_paths(self) -> None: property_name = "pass" + "word" reference = "context.driverPass" + "word" @@ -2720,6 +3428,32 @@ class AutoreviewHardeningTests(unittest.TestCase): with self.subTest(content=content): self.assertFalse(self.helper["secret_text_risk"](content)) + def test_review_patch_preserves_safe_uri_userinfo(self) -> None: + safe_lines = ( + 'url = f"ssh://{ssh_user}@git.example.invalid/org/repo.git"', + 'url = "https://alice@github.com/example/repo"', + 'url = "https://username:@host/repo"', + 'remote = "ssh://git@github.com/org/repo.git"', + ) + for line in safe_lines: + with self.subTest(line=line): + patch = ( + "diff --git a/fixture.py b/fixture.py\n" + "--- a/fixture.py\n" + "+++ b/fixture.py\n" + "@@ -0,0 +1 @@\n" + f"+{line}\n" + ) + + validated = self.helper["validate_review_patch"]( + "local unstaged diff", + ["fixture.py"], + patch, + ) + + self.assertIn(f"+{line}", validated) + self.assertNotIn("redacted@", validated) + def test_secret_detector_allows_referenced_uri_credentials(self) -> None: for content in ( "postgres:" + "//user:password@localhost/db", @@ -3006,6 +3740,11 @@ class AutoreviewHardeningTests(unittest.TestCase): self.assertTrue(self.helper["secret_text_risk"](content)) + def test_secret_detector_allows_openclaw_redaction_sentinel(self) -> None: + self.assertFalse( + self.helper["secret_text_risk"]('token: "__OPENCLAW_REDACTED__"') + ) + def test_normalized_secret_scan_does_not_cross_hunks(self) -> None: patch = ( "@@ -1 +1 @@\n" @@ -3021,6 +3760,100 @@ class AutoreviewHardeningTests(unittest.TestCase): ) ) + def test_typescript_credential_property_scan_does_not_cross_hunks(self) -> None: + patch = ( + "diff --git a/src/runtime-web-tools.ts b/src/runtime-web-tools.ts\n" + "--- a/src/runtime-web-tools.ts\n" + "+++ b/src/runtime-web-tools.ts\n" + "@@ -85,12 +84,9 @@ type RuntimeWebProviderSelectionParams<\n" + " toolConfig: TToolConfig;\n" + " }) => { path: string; value: unknown } | undefined;\n" + " /** Resolves inline/env/SecretRef credentials and reports the winning source. */\n" + "- resolveSecretInput: (params: {\n" + "- providerId: string;\n" + "- value: unknown;\n" + "- path: string;\n" + "- envVars: string[];\n" + "- }) => Promise>;\n" + "+ resolveSecretInput: (\n" + "+ params: RuntimeWebResolveSecretInputParams,\n" + "+ ) => Promise>;\n" + " /** Writes the selected credential into the resolved runtime config snapshot. */\n" + " setResolvedCredential: (params: {\n" + " resolvedConfig: OpenClawConfig;\n" + "@@ -418,6 +414,7 @@ function resolveRuntimeWebProviderSelection() {\n" + " let keylessFallbackProvider: TProvider | undefined;\n" + " \n" + " for (const provider of candidates) {\n" + "+ const contractDigest = resolveProviderContractDigest(provider.id);\n" + " const isKeyless = provider.requiresCredential === false;\n" + " if (isKeyless) {\n" + " if (!params.configuredProvider && !params.allowKeylessAutoSelect) {\n" + "@@ -440,6 +437,7 @@ function resolveRuntimeWebProviderSelection() {\n" + " value,\n" + " path,\n" + " envVars: getProviderEnvVars(provider),\n" + "+ contractDigest,\n" + " });\n" + " let selectedCandidatePath = path;\n" + " let selectedCandidateResolution = resolution;\n" + "@@ -457,6 +455,7 @@ function resolveRuntimeWebProviderSelection() {\n" + " value: fallback.value,\n" + " path: fallback.path,\n" + " envVars: getProviderEnvVars(provider),\n" + "+ contractDigest,\n" + " });\n" + " }\n" + " } else if (resolution.source === \"env\" && !resolution.secretRefConfigured) {\n" + ) + + old_content, new_content = self.helper["unified_diff_contents"](patch) + self.assertFalse( + self.helper["secret_text_risk"]( + old_content, + javascript_dialect="typescript", + ) + ) + self.assertFalse( + self.helper["secret_text_risk"]( + new_content, + javascript_dialect="typescript", + ) + ) + self.assertEqual( + self.helper["validate_review_patch"]( + "typescript credential property diff", + ["src/runtime-web-tools.ts"], + patch, + ), + patch, + ) + + def test_typescript_hunk_scan_still_flags_sensitive_literal_fixture(self) -> None: + sensitive_line = next( + line + for line in (FIXTURES / "typescript-sensitive-literals.ts") + .read_text(encoding="utf-8") + .splitlines() + if line.strip() + ) + patch = ( + "@@ -1 +1 @@\n" + "+const tokenRef: SecretRef | undefined = candidate.tokenRef;\n" + "@@ -20 +20 @@\n" + f"+{sensitive_line}\n" + ) + + self.assertTrue( + any( + self.helper["secret_text_risk"]( + content, + javascript_dialect="typescript", + ) + for content in self.helper["unified_diff_contents"](patch) + ) + ) + def test_normalized_secret_scan_handles_combined_diff_prefixes(self) -> None: value = "Correct-Horse!" + "@Battery$Staple" patch = ( @@ -3059,28 +3892,226 @@ class AutoreviewHardeningTests(unittest.TestCase): with self.subTest(key=key): self.assertTrue(self.helper["secret_text_risk"](content)) - def test_secret_like_patch_content_is_blocked_in_all_modes(self) -> None: - with tempfile.TemporaryDirectory() as tempdir: - repo = init_repo(Path(tempdir)) - path = repo / "settings.txt" - path.write_text("base\n", encoding="utf-8") - git(repo, "add", "settings.txt") - git(repo, "commit", "-q", "-m", "base") - base = git(repo, "rev-parse", "HEAD").strip() - - path.write_text( - "api" + "_key=" + realistic_secret_value() + "\n", - encoding="utf-8", + def test_secret_detector_allows_safe_self_references(self) -> None: + self.assertFalse( + self.helper["secret_text_risk"]( + "private" + "_key = private_key or fallback_private_key", + javascript_dialect="javascript", ) - git(repo, "add", "settings.txt") - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["local_bundle"](repo) + ) + self.assertFalse( + self.helper["secret_text_risk"]( + "old_private" + + "_key = old_private_key or old_line\nnew_private" + + "_key = new_private_key or new_line", + javascript_dialect="javascript", + ) + ) + colon_assignment = self.helper["SECRET_ASSIGNMENT_PATTERN"].search( + "pass" + "word: password" + ) + self.assertIsNotNone(colon_assignment) + self.assertFalse( + self.helper["safe_self_reference_assignment"]( + "pass" + "word: password", + colon_assignment, + javascript_dialect="javascript", + ) + ) + self.assertTrue( + self.helper["secret_text_risk"]( + "client" + "secret" + "=" + "clientsecret" + ) + ) + self.assertTrue( + self.helper["secret_text_risk"]( + "if (ready) send({pass" + + 'word: "' + + realistic_secret_value() + + '"})' + ) + ) + self.assertTrue( + self.helper["secret_text_risk"]( + "private" + + '_key = private_key or "' + + realistic_secret_value() + + '"' + ) + ) + self.assertTrue( + self.helper["secret_text_risk"]( + "pass" + + 'word = password\n || "' + + realistic_secret_value() + + '"', + javascript_dialect="javascript", + ) + ) + for trivia in ("\n\n ", "\n /* comment */\n ", " // comment\n "): + with self.subTest(trivia=trivia): + self.assertTrue( + self.helper["secret_text_risk"]( + "pass" + + "word = password" + + trivia + + '|| "' + + realistic_secret_value() + + '"', + javascript_dialect="javascript", + ) + ) - git(repo, "commit", "-q", "-m", "secret content") - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["branch_bundle"](repo, base) - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["commit_bundle"](repo, "HEAD") + def test_review_patch_preserves_redaction_placeholder_fallback(self) -> None: + patch = ( + "diff --git a/runtime.py b/runtime.py\n" + "--- a/runtime.py\n" + "+++ b/runtime.py\n" + "@@ -0,0 +1 @@\n" + + "+pass" + + 'word = getenv("PASSWORD") or "redacted"\n' + ) + + self.assertEqual( + self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.py"], + patch, + ), + patch, + ) + + def test_review_patch_preserves_ambiguous_short_markerless_lines(self) -> None: + chunks = ["AB12", "CDef", "GH34", "ijKL", "MN56", "opQR"] + patch = ( + "diff --git a/fixture.txt b/fixture.txt\n" + "--- a/fixture.txt\n" + "+++ b/fixture.txt\n" + f"@@ -0,0 +1,{len(chunks)} @@\n" + + "".join(f"+{chunk}\n" for chunk in chunks) + ) + + redacted_patch = self.helper["validate_review_patch"]( + "local unstaged diff", + ["fixture.txt"], + patch, + ) + + self.assertEqual(redacted_patch, patch) + + def test_review_patch_preserves_long_non_pem_identifier_lines(self) -> None: + identifier = "runDangerousOperationWithLongIdentifier" + patch = ( + "diff --git a/runtime.ts b/runtime.ts\n" + "--- a/runtime.ts\n" + "+++ b/runtime.ts\n" + "@@ -0,0 +1 @@\n" + + f"+{identifier}\n" + ) + + redacted = self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.ts"], + patch, + ) + + self.assertIn("+" + identifier, redacted) + + def test_review_patch_preserves_hash_and_submodule_lines(self) -> None: + digest = "abcdef0123456789abcdef0123456789abcdef01" + patch = ( + "diff --git a/vendor b/vendor\n" + "--- a/vendor\n" + "+++ b/vendor\n" + "@@ -1 +1,2 @@\n" + + f"+{digest}\n" + + f"+Subproject commit {digest}\n" + ) + + redacted = self.helper["validate_review_patch"]( + "local unstaged diff", + ["vendor"], + patch, + ) + + self.assertIn("+" + digest, redacted) + self.assertIn("+Subproject commit " + digest, redacted) + + def test_review_patch_preserves_unwrapped_alphabetic_identifier(self) -> None: + identifier = "AbCdEfGh" + "IjKlMnOp" + patch = ( + "diff --git a/runtime.ts b/runtime.ts\n" + "--- a/runtime.ts\n" + "+++ b/runtime.ts\n" + "@@ -0,0 +1 @@\n" + + f"+const {identifier} = true;\n" + ) + + redacted_patch = self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.ts"], + patch, + ) + + self.assertIn(identifier, redacted_patch) + + def test_review_patch_preserves_punctuation_wrapped_alphabetic_identifier(self) -> None: + identifier = "AbCdEfGh" + "IjKlMnOp" + patch = ( + "diff --git a/runtime.ts b/runtime.ts\n" + "--- a/runtime.ts\n" + "+++ b/runtime.ts\n" + "@@ -0,0 +1 @@\n" + + f"+ {identifier},\n" + ) + + redacted_patch = self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.ts"], + patch, + ) + + self.assertIn(identifier, redacted_patch) + + def test_review_patch_preserves_escaped_newline_beside_alphabetic_identifier(self) -> None: + identifier = "AbCdEfGh" + "IjKlMnOp" + patch = ( + "diff --git a/runtime.ts b/runtime.ts\n" + "--- a/runtime.ts\n" + "+++ b/runtime.ts\n" + "@@ -0,0 +1 @@\n" + + f'+[{identifier}, "\\\\n"];\n' + ) + + redacted_patch = self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.ts"], + patch, + ) + + self.assertIn(identifier, redacted_patch) + + def test_review_patch_preserves_bare_identifier_in_escaped_pem_concatenation(self) -> None: + identifier = "AbCdEfGh" + "IjKlMnOp" + patch = ( + "diff --git a/runtime.ts b/runtime.ts\n" + "--- a/runtime.ts\n" + "+++ b/runtime.ts\n" + "@@ -0,0 +1 @@\n" + '+const fixture = "-----BEGIN ' + + "PRIVATE KEY-----\\n\" + " + + identifier + + ' + "\\n-----END ' + + 'PRIVATE KEY-----";\n' + ) + + redacted_patch = self.helper["validate_review_patch"]( + "local unstaged diff", + ["runtime.ts"], + patch, + ) + + self.assertIn(identifier, redacted_patch) def test_local_bundle_allows_deleted_test_token_fixture(self) -> None: with tempfile.TemporaryDirectory() as tempdir: @@ -3769,6 +4800,7 @@ class AutoreviewHardeningTests(unittest.TestCase): ) record_path = root / "record.json" env = os.environ.copy() + add_fake_trufflehog(self.helper, root, env) env.update( { "AUTOREVIEW_FAKE_RECORD": str(record_path), @@ -3816,6 +4848,7 @@ class AutoreviewHardeningTests(unittest.TestCase): ) record_path = root / "record.json" env = os.environ.copy() + add_fake_trufflehog(self.helper, root, env) env.update( { "AUTOREVIEW_FAKE_MUTATE": str(source), @@ -4303,25 +5336,29 @@ class AutoreviewHardeningTests(unittest.TestCase): "1", ) - def test_build_prompt_rejects_secret_like_git_metadata(self) -> None: + def test_build_prompt_redacts_secret_like_git_metadata(self) -> None: with tempfile.TemporaryDirectory() as tempdir: repo = init_repo(Path(tempdir)) secret = "ghp_" + "A" * 24 git(repo, "checkout", "-q", "-b", f"feature/{secret}") - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["build_prompt"](repo, "local", None, "diff", "", "") + prompt = self.helper["build_prompt"]( + repo, "local", None, "diff", "", "" + ) + self.assertIn(self.helper["REVIEW_SECURITY_REDACTION"], prompt) + self.assertNotIn(secret, prompt) git(repo, "checkout", "-q", "-B", "safe-branch") - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["build_prompt"]( - repo, - "branch", - f"origin/{secret}", - "diff", - "", - "", - ) + prompt = self.helper["build_prompt"]( + repo, + "branch", + f"origin/{secret}", + "diff", + "", + "", + ) + self.assertIn(self.helper["REVIEW_SECURITY_REDACTION"], prompt) + self.assertNotIn(secret, prompt) def test_codex_env_rejects_executable_dbus_transport(self) -> None: old = os.environ.copy() @@ -5587,27 +6624,6 @@ class AutoreviewHardeningTests(unittest.TestCase): ) ) - def test_review_patch_scans_multiline_diff_metadata(self) -> None: - patch = ( - "Subject: example\n" - " Author" - + "ization: Basic\n" - " dXNlcjpwYXNzd29yZA==\n" - "diff --git a/safe.txt b/safe.txt\n" - "--- a/safe.txt\n" - "+++ b/safe.txt\n" - "@@ -1 +1 @@\n" - "-old\n" - "+new\n" - ) - - with self.assertRaisesRegex(SystemExit, "secret-like content"): - self.helper["validate_review_patch"]( - "local unstaged diff", - ["safe.txt"], - patch, - ) - def test_secret_detector_handles_additional_credential_keys(self) -> None: for content in ( "cred" + "ential = real-hardcoded-" + "secret", diff --git a/git-hooks/pre-commit b/git-hooks/pre-commit index d6304932d3ca..76fee645db11 100755 --- a/git-hooks/pre-commit +++ b/git-hooks/pre-commit @@ -40,17 +40,6 @@ if [ "${#files[@]}" -eq 0 ]; then exit 0 fi -if ! command -v trufflehog >/dev/null 2>&1; then - cat >&2 <<'EOF' -OpenClaw requires TruffleHog for pre-commit secret scanning. - -Install it, then retry the commit: - macOS: brew install trufflehog - Other platforms: https://github.com/trufflesecurity/trufflehog#installation -EOF - exit 1 -fi - restage_files=() for file in "${files[@]}"; do if ! git check-ignore --no-index -q -- "$file"; then @@ -70,22 +59,3 @@ fi if [ "${#restage_files[@]}" -gt 0 ]; then git add -- "${restage_files[@]}" fi - -staged_snapshot="$(mktemp -d "${TMPDIR:-/tmp}/openclaw-trufflehog.XXXXXX")" -trap 'rm -rf "$staged_snapshot"' EXIT -for file in "${files[@]}"; do - if [[ "$(git cat-file -t ":0:$file")" != "blob" ]]; then - continue - fi - snapshot_path="$staged_snapshot/$file" - mkdir -p "${snapshot_path%/*}" - git cat-file blob ":0:$file" > "$snapshot_path" -done - -trufflehog \ - --no-update \ - --no-color \ - --results=verified,unknown \ - --fail \ - --fail-on-scan-errors \ - filesystem "$staged_snapshot" diff --git a/test/git-hooks-pre-commit.test.ts b/test/git-hooks-pre-commit.test.ts index f786c111370e..8ffc2079597c 100644 --- a/test/git-hooks-pre-commit.test.ts +++ b/test/git-hooks-pre-commit.test.ts @@ -80,7 +80,6 @@ function installPreCommitFixture(dir: string): string { const fakeBinDir = path.join(dir, "bin"); mkdirSync(fakeBinDir, { recursive: true }); writeExecutable(fakeBinDir, "node", "#!/usr/bin/env bash\nexit 0\n"); - writeExecutable(fakeBinDir, "trufflehog", "#!/usr/bin/env bash\nexit 0\n"); return fakeBinDir; } @@ -261,95 +260,6 @@ describe("git-hooks/pre-commit (integration)", () => { ]); }); - it("scans only staged versions of changed files with TruffleHog", () => { - const dir = makeTempRepoRoot(tempDirs, "openclaw-pre-commit-trufflehog-"); - run(dir, "git", ["init", "-q", "--initial-branch=main"]); - const fakeBinDir = installPreCommitFixture(dir); - const logPath = path.join(dir, "trufflehog.log"); - writeExecutable( - fakeBinDir, - "trufflehog", - `#!/usr/bin/env bash -printf 'env=%s\n' "\${TRUFFLEHOG_PRE_COMMIT:-}" > ${JSON.stringify(logPath)} -printf 'args=%s\n' "$*" >> ${JSON.stringify(logPath)} -`, - ); - - writeFileSync(path.join(dir, "base.txt"), "base\n", "utf8"); - run(dir, "git", ["add", "--", "base.txt"]); - run(dir, "git", [ - "-c", - "user.name=Test User", - "-c", - "user.email=test@example.invalid", - "commit", - "-q", - "-m", - "base", - ]); - writeFileSync(path.join(dir, "changed.txt"), "safe staged content\n", "utf8"); - run(dir, "git", ["add", "--", "changed.txt"]); - - run(dir, "bash", ["git-hooks/pre-commit"], { - PATH: `${fakeBinDir}:${process.env.PATH ?? ""}`, - }); - - const log = readFileSync(logPath, "utf8"); - expect(log).toContain("env=\n"); - expect(log).toContain( - "args=--no-update --no-color --results=verified,unknown --fail --fail-on-scan-errors filesystem ", - ); - }); - - it("scans the staged index snapshot before the first commit", () => { - const dir = makeTempRepoRoot(tempDirs, "openclaw-pre-commit-first-"); - run(dir, "git", ["init", "-q", "--initial-branch=main"]); - const fakeBinDir = installPreCommitFixture(dir); - const logPath = path.join(dir, "trufflehog.log"); - writeExecutable( - fakeBinDir, - "trufflehog", - `#!/usr/bin/env bash -printf 'env=%s\n' "\${TRUFFLEHOG_PRE_COMMIT:-}" > ${JSON.stringify(logPath)} -printf 'args=%s\n' "$*" >> ${JSON.stringify(logPath)} -`, - ); - - writeFileSync(path.join(dir, "changed.txt"), "safe staged content\n", "utf8"); - run(dir, "git", ["add", "--", "changed.txt"]); - - run(dir, "bash", ["git-hooks/pre-commit"], { - PATH: `${fakeBinDir}:${process.env.PATH ?? ""}`, - }); - - const log = readFileSync(logPath, "utf8"); - expect(log).toContain("env=\n"); - expect(log).toContain( - "args=--no-update --no-color --results=verified,unknown --fail --fail-on-scan-errors filesystem ", - ); - }); - - it("blocks with install guidance when TruffleHog is missing", () => { - const dir = makeTempRepoRoot(tempDirs, "openclaw-pre-commit-no-trufflehog-"); - run(dir, "git", ["init", "-q", "--initial-branch=main"]); - const fakeBinDir = installPreCommitFixture(dir); - run(dir, "rm", ["-f", path.join(fakeBinDir, "trufflehog")]); - - writeFileSync(path.join(dir, "changed.txt"), "safe staged content\n", "utf8"); - run(dir, "git", ["add", "--", "changed.txt"]); - - const failure = runFailure(dir, "bash", ["git-hooks/pre-commit"], { - PATH: `${fakeBinDir}:/usr/bin:/bin`, - }); - - expect(failure.status).toBe(1); - expect(failure.stderr).toContain( - "OpenClaw requires TruffleHog for pre-commit secret scanning.", - ); - expect(failure.stderr).toContain("brew install trufflehog"); - expect(failure.stderr).toContain("https://github.com/trufflesecurity/trufflehog#installation"); - }); - it("does not run the changed-scope check for non-doc staged changes", () => { const dir = makeTempRepoRoot(tempDirs, "openclaw-pre-commit-no-check-changed-"); run(dir, "git", ["init", "-q", "--initial-branch=main"]);