#!/usr/bin/env python3 """Claude Code PreToolUse hook: wrap noisy Bash commands. Reads hook JSON from stdin and prints a JSON response understood by Claude Code. Install via `.claude/settings.json` hooks. Keep this script project-local during experiments so it can be versioned and reviewed. """ from __future__ import annotations import copy from dataclasses import dataclass import importlib.util import json import os from pathlib import Path import shlex import re import shutil import stat import subprocess import sys _SCRIPT_DIR = Path(__file__).resolve().parent def _load_optional_helper(module_name: str): """선택적 helper 를 SCRIPT_DIR 그다음 ../lib 순서로 찾는다. 없으면 None. 이 훅은 helper 없이도 예전과 완전히 동일하게 동작해야 하므로 로딩 실패는 절대 밖으로 새어 나가지 않는다. """ for helper_dir in (_SCRIPT_DIR, _SCRIPT_DIR.parent / "lib"): helper_path = helper_dir / f"{module_name}.py" try: if not helper_path.is_file(): continue spec = importlib.util.spec_from_file_location(f"_context_guard_{module_name}", helper_path) if spec is None or spec.loader is None: continue module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module except Exception: # noqa: BLE001 - helper 부재/파손이 훅 출력을 바꾸면 안 된다. continue return None _hook_journal = _load_optional_helper("hook_journal") _hook_switch = _load_optional_helper("hook_switch") def _journal_start(): """저널 helper 가 있으면 단조 시계를 켠다. 없으면 None.""" if _hook_journal is None: return None try: return _hook_journal.start_clock() except Exception: # noqa: BLE001 return None def _journal_record(**fields) -> None: """저널 helper 가 있을 때만 한 줄 기록한다. 실패는 전부 삼킨다.""" if _hook_journal is None: return try: _hook_journal.record("bash", **fields) except Exception: # noqa: BLE001 pass def _switch_says_off() -> bool: """세션 스위치로 bash 훅이 꺼져 있는지 본다. helper 가 없으면 항상 False.""" if _hook_switch is None: return False try: return bool(_hook_switch.is_disabled("bash")) except Exception: # noqa: BLE001 return False def bash_disable_hint() -> str: """훅 메시지 끝에 붙일 한 줄 해제 안내. helper 가 없으면 기존 환경변수 문구.""" if _hook_switch is not None: try: return _hook_switch.disable_hint("bash") except Exception: # noqa: BLE001 pass return "Set CONTEXT_GUARD_DISABLE=1 to disable ContextGuard's Bash hook." # 저널용 stdin 크기. load_hook_payload 가 채운다. _hook_input_bytes = 0 ENV_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=.*") WRAPPER_BASENAMES = frozenset({ "trim_command_output.py", "context-guard-trim-output", "sanitize_output.py", "context-guard-sanitize-output", }) MINISHELL_ROUTE_POLICY_VERSION = "minishell-route-v1" MINISHELL_EXPLICIT_NOOP_ARGV = frozenset({ ("kubectl", "get", "pods"), ("kubectl", "version"), ("docker", "ps"), ("docker", "images"), ("docker", "compose", "ps"), }) MINISHELL_MAX_COMMAND_BYTES = 65_536 MINISHELL_MAX_LEXICAL_ITEMS = 4_096 MINISHELL_MAX_SEGMENTS = 8 MINISHELL_MAX_WORDS_PER_SEGMENT = 256 MINISHELL_MAX_HEREDOC_DELIMITER_BYTES = 64 MINISHELL_DENIED_ACTIVE_CHARS = frozenset(";&>()`*?[]{}") MINISHELL_DENIED_COMMAND_WORDS = frozenset({ "!", "case", "coproc", "do", "done", "elif", "else", "esac", "fi", "for", "function", "if", "in", "select", "then", "time", "until", "while", }) MINISHELL_DENIED_COMMAND_BASENAMES = frozenset({ "curl", "eval", "exec", "fetch", "ftp", "nc", "ncat", "netcat", "scp", "sftp", "socat", "ssh", "tee", "telnet", "wget", }) MINISHELL_DENIED_SHELL_BASENAMES = frozenset({ "bash", "dash", "fish", "ksh", "sh", "zsh", }) MINISHELL_HEREDOC_STDIN_CONSUMERS = frozenset({ "cut", "sed", "sort", "uniq", "wc", }) MINISHELL_HEREDOC_DELIMITER_RE = re.compile(r"^[A-Za-z0-9_]+$") # bash 가 접두사 할당으로 적용하는 `NAME+=VALUE` 형태 — MiniShell 은 이를 할당으로 # 표시하지 않으므로(§_is_unmodeled_assignment_prefix) 라우팅 접두사 구간에서 거부한다. MINISHELL_APPEND_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\+=") # 환경변수 접두사(`KEY=VALUE cmd`) 이름 화이트리스트 — FIX-5, 원칙 4의 유일한 예외. # denylist 는 구조적으로 종료하지 않는다(실측: 최소 denylist가 PAGER/EDITOR/VISUAL/ # PERL5LIB/RUBYOPT/PYTHONPATH/PYTHONSTARTUP/NODE_OPTIONS 8종을 놓침). 이 15개는 # "값을 실행 가능한 코드 경로로 해석하지 않는다"는 기준을 통과한 것만 포함한다. # 정확 이름 일치만 허용 — 접두사/글롭 매칭 금지(`TERM*`는 `TERMINFO`를 재승인시킨다). # TERM 은 TERMINFO/TERMINFO_DIRS 가, LANG/LC_* 는 LOCPATH/NLSPATH 가 배제되었기 # 때문에만 안전하다 — 이 조건부 안전성을 확장 심사 시 반드시 재확인할 것. MINISHELL_ALLOWED_ENV_PREFIX_NAMES = frozenset({ "LANG", "LC_ALL", "LC_CTYPE", "LC_NUMERIC", "LC_TIME", "LC_COLLATE", "LC_MESSAGES", "TZ", "NO_COLOR", "CLICOLOR", "CI", "COLUMNS", "LINES", "TERM", "NODE_ENV", }) CGW1_MAX_LINES = "220" # 기본 Bash 래핑에 붙는 escrow 플래그. 순서까지 계약이다 — 재진입 방지용 봉투 # 인식(classify_incoming_wrapper)이 이 형태를 그대로 매칭한다. CGW1_ESCROW_FLAGS = ("--digest", "markdown", "--artifact-receipt") CGW1_SHELL_ARGV = ("bash", "-c") CGW1_SENTINEL = "--context-guard-wrapper-v1" CGW1_COMMAND_SEARCH_DIFF = "command_search_diff" BASH_REFERENCE_FLAG = "--bash-reference-v1" BASH_REFERENCE_PUBLIC_COMMAND = "./node_modules/.bin/context-guard" BASH_REFERENCE_HANDLE_RE = re.compile(r"^cgr1p_[A-Za-z0-9_-]{43}$", re.ASCII) # 사용자의 탈출구. 이 훅은 실행을 막지 않지만 여전히 명령을 재작성하며, # 재작성은 신중한 워크플로에는 눈에 보이는 거부보다 오히려 더 침습적이다 # (trim 이 `git log --pretty=…` 를 망가뜨리거나, 하위 스크립트가 감싸인 출력을 # 읽거나, 종료 코드가 어긋나는 경우). 이 변수가 설정되면 분류 이전에 개입을 # 포기한다. DISABLE_ENV = "CONTEXT_GUARD_DISABLE" FAIL_OPEN_ENV = "CONTEXT_GUARD_SANITIZER_FAIL_OPEN" LEGACY_FAIL_OPEN_ENV = "CLAUDE_TOKEN_SANITIZER_FAIL_OPEN" FAIL_OPEN_VALUES = {"1", "true", "yes", "on"} MAX_HOOK_ENVELOPE_BYTES = 1_048_576 UNPARSEABLE_SANITIZER_RISK_RE = re.compile( r"(?i)(?:^|[\s;&|()])" r"(?:rg|grep|egrep|fgrep|journalctl|kubectl|oc|docker|podman|docker-compose|git|find)" r"(?:$|[\s;&|()])" ) def _approved_runtime_executable(name: str) -> str: """Resolve only from the fixed OS command path, never inherited PATH.""" found = shutil.which(name, path=os.defpath) if not found: raise RuntimeError(f"required runtime {name!r} is unavailable") canonical = os.path.realpath(found) if not os.path.isabs(canonical) or not os.path.isfile(canonical) or not os.access(canonical, os.X_OK): raise RuntimeError(f"required runtime {name!r} is not an executable regular file") return canonical def _approved_python_runtime() -> str: canonical = os.path.realpath(sys.executable) if not canonical or not os.path.isabs(canonical) or not os.path.isfile(canonical) or not os.access(canonical, os.X_OK): raise RuntimeError("approved Python runtime is unavailable") return canonical def _runtime_shell_argv() -> tuple[str, ...]: return ( _approved_runtime_executable("env"), "-u", "BASH_ENV", "-u", "ENV", "-u", "PYTHONHOME", "-u", "PYTHONPATH", "-u", "PYTHONSTARTUP", "-u", "SHELLOPTS", "-u", "BASHOPTS", "-u", "PS4", _approved_runtime_executable("bash"), "--noprofile", "--norc", "-p", "-c", ) class UnsafeAdjacentWrapperError(RuntimeError): """Adjacent wrapper is not a regular file opened without symlink following.""" def _isolated_wrapper_prefix(wrapper: str) -> list[str]: # O_NOFOLLOW+fstat rejects a symlink planted before this call; it cannot # close the gap between this check and the later, separate process that # actually execs the returned path (the shell string this hook emits is # run by the harness, not by a child of this process, so the validated # fd cannot be carried across that boundary). A post-check swap of the # wrapper still requires write/rename authority over the install # directory itself - the same authority needed to replace the CLI - so # this is accepted as a residual risk outside this hook's threat model. if not hasattr(os, "O_NOFOLLOW"): raise UnsafeAdjacentWrapperError("O_NOFOLLOW is required for adjacent wrappers") flags = os.O_RDONLY | os.O_NOFOLLOW if hasattr(os, "O_CLOEXEC"): flags |= os.O_CLOEXEC try: fd = os.open(wrapper, flags) except OSError as exc: raise UnsafeAdjacentWrapperError( f"wrapper could not be opened without following symlinks: {exc}" ) from exc try: if not stat.S_ISREG(os.fstat(fd).st_mode): raise UnsafeAdjacentWrapperError("wrapper is not a regular file") finally: os.close(fd) return [_approved_python_runtime(), "-I", os.path.realpath(wrapper)] # kubectl/docker/podman/oc 글로벌 옵션 중 다음 토큰을 value로 소비하는 형태. # `-n prod`, `--context=prod`, `-f file.yml` 같은 케이스를 hub로 흡수해 # `kubectl -n prod logs api`, `docker --context prod logs api`, # `docker compose -f compose.yml logs web` 가 sanitize wrapper를 거치도록 한다. _VALUE_TAKING_FLAGS = frozenset({ "-n", "--namespace", "--context", "--kubeconfig", "--cluster", "--user", "--token", "--as", "--as-group", "-s", "--server", "-c", "-H", "--host", "--config", "--log-level", "-f", "--file", "-p", "--project-name", }) # find 가 단순 path listing 이 아니라 임의 명령 출력을 발생시킬 수 있는 액션. # 이 액션들은 .env / 자격증명 파일 내용까지 노출 가능하므로 trim 대신 sanitize 로 라우팅한다. _FIND_OUTPUT_RISK_ACTIONS = frozenset({ "-delete", "-exec", "-execdir", "-ok", "-okdir", "-fprint", "-fprint0", "-fprintf", "-fls", }) @dataclass(frozen=True) class MiniShellWord: value: str source_value: str active: tuple[bool, ...] barriers: frozenset[int] assignment_index: int | None active_tilde_sites: tuple[int, ...] @dataclass(frozen=True) class MiniShellParse: words: tuple[MiniShellWord, ...] segments: tuple[tuple[MiniShellWord, ...], ...] argv: tuple[str, ...] consumed: int denial_reason: str | None = None lexical_items: int = 0 heredoc_delimiter: str | None = None @dataclass(frozen=True) class CommandDecision: action: str parsed: MiniShellParse reason: str | None = None reason_code: str | None = None route_code: str | None = None policy_version: str = MINISHELL_ROUTE_POLICY_VERSION # 훅이 래핑을 포기한 이유. `action == "noop"` 이면서 이 값이 있으면 "조용한 # 통과"가 아니라 "래핑 불가로 판단해 원본 그대로 통과"를 뜻한다. 실행을 # 막지 않으므로 사용자에게는 보이지 않고, 진단 표면에서만 읽는다. decline_reason: str | None = None class HookInputError(ValueError): def __init__(self, reason_code: str): super().__init__(reason_code) self.reason_code = reason_code def reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]: decoded: dict[str, object] = {} for key, value in pairs: if key in decoded: raise HookInputError("duplicate_json_key") decoded[key] = value return decoded def reject_nonfinite_json_number(value: str) -> object: raise HookInputError(f"non_finite_json_number_{value.lower()}") def load_hook_payload() -> dict[str, object]: global _hook_input_bytes raw_payload = sys.stdin.buffer.read(MAX_HOOK_ENVELOPE_BYTES + 1) # 저널이 쓸 입력 크기. 파싱이 실패해도 읽은 바이트 수는 사실이므로 먼저 남긴다. _hook_input_bytes = len(raw_payload) if len(raw_payload) > MAX_HOOK_ENVELOPE_BYTES: raise HookInputError("envelope_too_large") try: payload_text = raw_payload.decode("utf-8") payload = json.loads( payload_text, object_pairs_hook=reject_duplicate_keys, parse_constant=reject_nonfinite_json_number, ) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HookInputError("malformed_json") from exc except RecursionError as exc: raise HookInputError("json_nesting_too_deep") from exc if not isinstance(payload, dict): raise HookInputError("top_level_not_object") return payload def select_tool_input(payload: dict[str, object]) -> dict[str, object]: has_snake_case = "tool_input" in payload has_camel_case = "toolInput" in payload if has_snake_case and has_camel_case: if payload["tool_input"] != payload["toolInput"]: raise HookInputError("conflicting_tool_input_aliases") tool_input = payload["tool_input"] elif has_snake_case: tool_input = payload["tool_input"] elif has_camel_case: tool_input = payload["toolInput"] else: raise HookInputError("missing_tool_input") if not isinstance(tool_input, dict): raise HookInputError("tool_input_not_object") command = tool_input.get("command") if not isinstance(command, str) or not command: raise HookInputError("missing_or_invalid_command") return tool_input def find_wrapper(kind: str) -> str | None: script_dir = os.path.dirname(os.path.abspath(__file__)) if kind == "sanitize": candidates = [ os.path.join(script_dir, "context-guard-sanitize-output"), os.path.join(script_dir, "sanitize_output.py"), ] else: candidates = [ os.path.join(script_dir, "context-guard-trim-output"), os.path.join(script_dir, "trim_command_output.py"), ] for path in candidates: if os.path.exists(path): return path return None def fail_open_source_env() -> str | None: """FAIL_OPEN 변수가 설정되었는지 보고한다. 이 변수는 더 이상 훅의 동작을 바꾸지 않는다. 원래 목적이 "차단당하느니 감싸지 않은 채로 실행하라" 였는데, 이제는 아무것도 차단하지 않으므로 그 목적이 기본 동작이 되었다. 재작성 자체를 끄려면 `CONTEXT_GUARD_DISABLE` 을 쓴다. 진단 목적으로만 남긴다. """ canonical_value = os.environ.get(FAIL_OPEN_ENV) if canonical_value is not None: return FAIL_OPEN_ENV if canonical_value.strip().lower() in FAIL_OPEN_VALUES else None if os.environ.get(LEGACY_FAIL_OPEN_ENV, "").strip().lower() in FAIL_OPEN_VALUES: return LEGACY_FAIL_OPEN_ENV return None def fail_open_enabled() -> bool: return fail_open_source_env() is not None def print_noop() -> None: print("{}") def print_deny_response(reason: str) -> None: print(json.dumps({ "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason, } }, ensure_ascii=False)) def print_ask_response(reason: str) -> None: """실행 여부를 사람에게 넘긴다. 훅은 거부하지 않는다.""" print(json.dumps({ "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "ask", "permissionDecisionReason": reason, } }, ensure_ascii=False)) def decline_invalid_hook_input(reason_code: str) -> None: """훅이 자기 stdin 을 읽지 못했을 때 개입을 포기한다. 여기서 실행을 막으면 호스트 페이로드 스키마가 조금만 바뀌어도 모든 사용자의 Bash 가 조용히 전면 중단된다. 훅은 명령이 실행되지 못하는 이유가 되어서는 안 된다. """ print( f"context-guard-rewrite-bash: could not read hook input ({reason_code}); " "leaving the command unchanged", file=sys.stderr, ) print_noop() def decline_missing_wrapper(wrapper_name: str, effect: str) -> None: """래퍼가 설치돼 있지 않으면 감싸지 않고 통과시킨다.""" print( f"context-guard-rewrite-bash: {wrapper_name} is not installed next to " f"context-guard-rewrite-bash; running this command {effect}. Reinstall " "ContextGuard to restore output guarding.", file=sys.stderr, ) print_noop() def deny_self_protection(reason: str) -> None: """훅 자신의 실행 봉투 재진입/위조만 거부한다. 사용자 명령에 대한 정책이 아니므로 fail-open 상태를 참조하지 않는다. """ print(f"context-guard-rewrite-bash: {reason}", file=sys.stderr) print_deny_response(reason) def _exact_assignment_index( value: str, active: tuple[bool, ...], barriers: frozenset[int], ) -> int | None: for index, char in enumerate(value): if char != "=" or not active[index]: continue name = value[:index] if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name): return None if not all(active[:index]): return None if any(boundary <= index for boundary in barriers): return None return index return None def _tilde_prefix_end(word: MiniShellWord, start: int, assignment_site: bool) -> int | None: source = word.source_value if start >= len(source) or source[start] != "~" or not word.active[start]: return None if start in word.barriers: return None index = start + 1 while index < len(source): char = source[index] if word.active[index] and (char == "/" or (assignment_site and char == ":")): break if not word.active[index] or index in word.barriers: return None index += 1 if any(start < boundary <= index for boundary in word.barriers): return None return index def _assignment_tilde_sites(word: MiniShellWord) -> tuple[tuple[int, int], ...]: assignment_index = word.assignment_index if assignment_index is None: return () sites: list[tuple[int, int]] = [] delimiters = [assignment_index] delimiters.extend( index for index in range(assignment_index + 1, len(word.source_value)) if word.source_value[index] == ":" and word.active[index] ) for delimiter in delimiters: start = delimiter + 1 if start in word.barriers: continue end = _tilde_prefix_end(word, start, assignment_site=True) if end is not None: sites.append((start, end)) return tuple(sites) def _annotate_word_tildes(word: MiniShellWord) -> MiniShellWord: sites = list(_assignment_tilde_sites(word)) if word.source_value.startswith("~"): end = _tilde_prefix_end(word, 0, assignment_site=False) if end is not None: sites.append((0, end)) return MiniShellWord( value=word.source_value, source_value=word.source_value, active=word.active, barriers=word.barriers, assignment_index=word.assignment_index, active_tilde_sites=tuple(start for start, _end in sorted(set(sites))), ) # 셸 `-c` 본문을 한 겹만 들여다본다. 중첩을 무한히 따라가는 것은 이 휴리스틱의 # 목적(실수하는 모델을 잡는 것)에 필요 없고, 회피자를 막지도 못한다. _FIND_SCAN_MAX_DEPTH = 1 # 봉투 매칭에서 "값은 무엇이든 좋다"를 뜻하는 센티넬. _ANY_TOKEN = object() _FIND_SIDE_EFFECT_ACTIONS = frozenset({ "-delete", "-exec", "-execdir", "-fls", "-fprint", "-fprint0", "-fprintf", "-ok", "-okdir", }) def _raw_command_is_side_effecting_find(command: str) -> bool: """파싱 성공 여부와 무관하게 파괴적인 `find` 형태를 알아본다. `find . -exec rm {} \\;` 는 `{}` 때문에 MiniShell 문법을 통과하지 못하고 파싱 단계에서 탈락한다. 예전에는 그 파싱 실패가 우연히 제동 역할을 했다. 이제 파싱 실패는 통과를 뜻하므로, 되돌릴 수 없는 삭제만큼은 파싱과 무관하게 알아보고 사람에게 물어야 한다. 오탐 비용은 확인 키 한 번뿐이므로 보수적으로 넓게 잡는다. 거부가 아니라 질문이기 때문에 감당할 수 있는 판정이다. """ try: tokens = shlex.split(command) except ValueError: tokens = command.split() return _tokens_carry_side_effecting_find(tokens) # shlex 는 따옴표 밖의 `;` `&` `|` 를 단어에 붙여 둔다. `find . -delete;true` 는 # `-delete;true` 한 토큰이 되어 정확 일치를 빗나갔고, 이어서 MiniShell 이 `;` 를 # 거부해 decline → 원본 그대로 실행됐다. 문자 다섯 개로 ask 게이트가 사라진 것이다. # 이 게이트의 취지는 파싱 성공 여부와 무관하게 되돌릴 수 없는 삭제를 사람에게 # 물어보는 것이므로, 구분자를 떼어낸 조각으로도 본다. # 공백류는 넣지 않는다. shlex 는 따옴표 밖 공백에서 이미 나누므로, 토큰 안에 남은 # 개행은 구분자가 아니라 이스케이프나 따옴표에서 온 것이다. `find . $\<개행>-exec …` # 를 개행으로 쪼개면 `-exec` 가 나오지만 bash 는 그 줄 연결을 지워 `$-exec` 로 읽어 # 실제로는 -exec 를 실행하지 않는다. 쪼개면 없는 위험을 물어보게 된다. _FIND_TOKEN_SEPARATORS = ";&|()" def _token_action_candidates(token: str) -> list[str]: """토큰에서 구분자를 떼어낸 조각들. `-delete;true` → `-delete`, `true`.""" pieces = [token] for separator in _FIND_TOKEN_SEPARATORS: expanded: list[str] = [] for piece in pieces: expanded.extend(piece.split(separator)) pieces = expanded return [piece for piece in pieces if piece] def _follower_is_side_effecting_action(follower: str) -> bool: """토큰 자체나 구분자로 잘린 조각이 부수효과 액션인지 본다.""" if follower in _FIND_SIDE_EFFECT_ACTIONS: return True return any( piece in _FIND_SIDE_EFFECT_ACTIONS for piece in _token_action_candidates(follower) ) def _tokens_carry_side_effecting_find(tokens: list[str], depth: int = 0) -> bool: for index, token in enumerate(tokens): # `bash -c "find . -delete"` 는 스크립트 본문이 토큰 하나로 남아 겉에서는 # 보이지 않는다. 모델이 일상적으로 쓰는 형태라 한 겹은 들어가서 본다. if ( depth < _FIND_SCAN_MAX_DEPTH and os.path.basename(token) in MINISHELL_DENIED_SHELL_BASENAMES ): for offset, follower in enumerate(tokens[index + 1:], start=index + 1): if not follower.startswith("-"): break if re.fullmatch(r"-[^-]*c[^-]*", follower) is None: continue body = tokens[offset + 1] if offset + 1 < len(tokens) else None if body is None: break try: inner = shlex.split(body) except ValueError: inner = body.split() if _tokens_carry_side_effecting_find(inner, depth + 1): return True break if os.path.basename(token) != "find": continue if any( _follower_is_side_effecting_action(follower) for follower in tokens[index + 1:] ): return True return False def _side_effecting_find_ask(parsed: MiniShellParse) -> CommandDecision: return CommandDecision( action="ask", parsed=parsed, reason=( "[context-guard] This find command modifies or deletes files — confirm if you meant it. " + bash_disable_hint() ), reason_code="side_effecting_find_ask", route_code="ask", ) def _decline(parsed: MiniShellParse, reason_code: str) -> CommandDecision: """래핑을 포기하고 원본 명령을 그대로 통과시킨다. ContextGuard 는 실행을 중재하지 않는다. 인식하지 못했거나 의미를 보존한 채 감쌀 수 없는 명령은 거부 대상이 아니라 손대지 않는 대상이며, 결과는 ContextGuard 가 설치되지 않았을 때와 동일하다. `reason_code` 는 진단용으로만 남고 사용자에게 노출되지 않는다. """ # `reason_code` 는 그대로 둔다. 원인은 결과와 무관하게 원인이며, 기존 # 진단·오라클이 고정해 온 것도 결과가 아니라 원인이다. 바뀐 것은 그 # 원인이 실행 차단으로 이어지지 않는다는 점뿐이다. return CommandDecision( action="noop", parsed=parsed, reason_code=reason_code, route_code="noop", decline_reason=reason_code, ) def _denied_minishell(command: str, consumed: int, reason: str) -> MiniShellParse: return MiniShellParse( words=(), segments=(), argv=(), consumed=consumed, denial_reason=reason, ) def _dollar_starts_expansion( command: str, index: int, *, allow_quoted_literal: bool = False, ) -> bool: cursor = index + 1 while command.startswith("\\\n", cursor): cursor += 2 if cursor >= len(command): return False following = command[cursor] if allow_quoted_literal and following in {'"', "'"}: return True return following in "({$0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz?!#*@-" def parse_minishell(command: str) -> MiniShellParse: """Parse the fully consumed bounded MiniShell-v1 grammar. The parser intentionally keeps quote/escape provenance instead of reconstructing it from decoded argv. Only backslash-newline is removed without leaving a provenance barrier; every retained quote or escape can therefore suppress a local Bash assignment-style tilde site. """ try: command_bytes = len(command.encode("utf-8")) except UnicodeEncodeError: return _denied_minishell(command, 0, "invalid_utf8") if command_bytes > MINISHELL_MAX_COMMAND_BYTES: return _denied_minishell( command, min(len(command), MINISHELL_MAX_COMMAND_BYTES), "command_bytes_exceeded", ) if "\0" in command: return _denied_minishell(command, command.index("\0"), "nul_denied") raw_segments: list[list[MiniShellWord]] = [[]] chars: list[str] = [] active: list[bool] = [] barriers: set[int] = set() in_word = False quote: str | None = None fragment_kind: str | None = None lexical_items = 0 heredoc_delimiter: str | None = None index = 0 def bump_item() -> bool: nonlocal lexical_items lexical_items += 1 return lexical_items <= MINISHELL_MAX_LEXICAL_ITEMS def finish_word() -> str | None: nonlocal chars, active, barriers, in_word, fragment_kind if not in_word: return None if len(raw_segments[-1]) >= MINISHELL_MAX_WORDS_PER_SEGMENT: return "segment_words_exceeded" source_value = "".join(chars) active_tuple = tuple(active) barrier_set = frozenset(barriers) raw_segments[-1].append(MiniShellWord( value=source_value, source_value=source_value, active=active_tuple, barriers=barrier_set, assignment_index=_exact_assignment_index( source_value, active_tuple, barrier_set, ), active_tilde_sites=(), )) chars = [] active = [] barriers = set() in_word = False fragment_kind = None return None def deny(reason: str, at: int | None = None) -> MiniShellParse: return _denied_minishell(command, index if at is None else at, reason) while index < len(command): char = command[index] if quote is None: if char == " ": error = finish_word() if error is not None: return deny(error) index += 1 continue if char in "\t\r\n": return deny("forbidden_whitespace") if char == "\\": if index + 1 >= len(command): return deny("trailing_escape") escaped = command[index + 1] if escaped == "\n": index += 2 fragment_kind = None continue if escaped in "\t\r": return deny("forbidden_escaped_character") if not bump_item(): return deny("lexical_items_exceeded") in_word = True chars.append(escaped) active.append(False) fragment_kind = None index += 2 continue if char in {"'", '"'}: if not bump_item(): return deny("lexical_items_exceeded") in_word = True barriers.add(len(chars)) quote = char fragment_kind = f"quote:{char}" index += 1 continue if char == "#" and not in_word: if not raw_segments[-1]: return deny("comment_without_command") if not bump_item(): return deny("lexical_items_exceeded") newline = command.find("\n", index) if newline < 0: index = len(command) break if any(tail != " " for tail in command[newline + 1:]): return deny("leftover_after_comment", newline + 1) index = len(command) break if char == "|": error = finish_word() if error is not None: return deny(error) if not bump_item(): return deny("lexical_items_exceeded") if ( not raw_segments[-1] or len(raw_segments) >= MINISHELL_MAX_SEGMENTS or command.startswith("|&", index) ): return deny("invalid_pipeline") raw_segments.append([]) index += 1 continue if char == "<": error = finish_word() if error is not None: return deny(error) if ( heredoc_delimiter is not None or len(raw_segments) != 1 or not raw_segments[-1] or not command.startswith("<<", index) or command.startswith(("<<<", "<<-"), index) ): return deny("unsupported_redirect") if not bump_item(): return deny("lexical_items_exceeded") delimiter_quote_index = index + 2 if ( delimiter_quote_index >= len(command) or command[delimiter_quote_index] not in {"'", '"'} ): return deny("unquoted_heredoc_delimiter") delimiter_quote = command[delimiter_quote_index] delimiter_end = command.find( delimiter_quote, delimiter_quote_index + 1, ) if delimiter_end < 0: return deny("unterminated_heredoc_delimiter") delimiter = command[delimiter_quote_index + 1:delimiter_end] if ( not delimiter or len(delimiter.encode("ascii", "ignore")) != len(delimiter) or len(delimiter) > MINISHELL_MAX_HEREDOC_DELIMITER_BYTES or MINISHELL_HEREDOC_DELIMITER_RE.fullmatch(delimiter) is None ): return deny("invalid_heredoc_delimiter") if not bump_item(): return deny("lexical_items_exceeded") header_end = delimiter_end + 1 while header_end < len(command) and command[header_end] == " ": header_end += 1 if header_end >= len(command) or command[header_end] != "\n": return deny("heredoc_header_not_terminated", header_end) body_start = header_end + 1 line_start = body_start terminator_end: int | None = None while line_start <= len(command): line_end = command.find("\n", line_start) if line_end < 0: if command[line_start:] == delimiter: terminator_end = len(command) break if command[line_start:line_end] == delimiter: terminator_end = line_end + 1 break line_start = line_end + 1 if terminator_end is None: return deny("unterminated_heredoc", body_start) if terminator_end != len(command): return deny("leftover_after_heredoc", terminator_end) if not bump_item(): return deny("lexical_items_exceeded") heredoc_delimiter = delimiter index = len(command) break if char in MINISHELL_DENIED_ACTIVE_CHARS: return deny(f"active_{ord(char):02x}") if char == "$" and _dollar_starts_expansion( command, index, allow_quoted_literal=True, ): return deny("active_24") if fragment_kind != "unquoted": if not bump_item(): return deny("lexical_items_exceeded") fragment_kind = "unquoted" in_word = True chars.append(char) active.append(True) index += 1 continue if char in "\t\r\n": if quote == '"' and char == "\n" and index > 0 and command[index - 1] == "\\": index += 1 continue return deny("forbidden_quoted_whitespace") if char == quote: barriers.add(len(chars)) quote = None fragment_kind = None index += 1 continue if quote == "'": chars.append(char) active.append(False) index += 1 continue if char == "`" or (char == "$" and _dollar_starts_expansion(command, index)): return deny("active_double_quote_expansion") if char == "\\": if index + 1 >= len(command): return deny("trailing_double_quote_escape") escaped = command[index + 1] if escaped == "\n": index += 2 continue if escaped in "\t\r": return deny("forbidden_escaped_character") if escaped in {'$', '`', '"', "\\"}: chars.append(escaped) active.append(False) else: chars.extend(("\\", escaped)) active.extend((False, False)) index += 2 continue chars.append(char) active.append(False) index += 1 if quote is not None: return _denied_minishell(command, len(command), "unterminated_quote") error = finish_word() if error is not None: return _denied_minishell(command, len(command), error) if not raw_segments[-1]: return _denied_minishell(command, len(command), "empty_command") segments = tuple( tuple(_annotate_word_tildes(word) for word in segment) for segment in raw_segments ) words = tuple(word for segment in segments for word in segment) return MiniShellParse( words=words, segments=segments, argv=tuple(word.value for word in words), consumed=len(command), lexical_items=lexical_items, heredoc_delimiter=heredoc_delimiter, ) def split_single_safe_command(command: str) -> list[str] | None: parsed = parse_minishell(command) if parsed.denial_reason is not None: return None return list(parsed.argv) def command_basename(command: str) -> str: """Return a trusted routing identity only for a bare ASCII command token. Route predicates describe standard command identities, not arbitrary files that happen to share a basename. Normalizing ``./rg`` or ``/tmp/evil/grep`` to a trusted name would let a caller-selected executable inherit that route. Non-ASCII tokens are also rejected here so Unicode separator lookalikes cannot create a visually ambiguous identity. """ if ( not command or not command.isascii() or not command.isprintable() or "/" in command or "\\" in command ): return "" return command def strip_env_prefix(argv: list[str]) -> list[str]: """Return the executable argv after leading `KEY=VALUE` or `env` wrappers.""" i = 0 while i < len(argv) and ENV_ASSIGNMENT_RE.match(argv[i]): i += 1 if i < len(argv) and argv[i] == "env": i += 1 while i < len(argv): token = argv[i] if token in {"-i", "--ignore-environment"}: i += 1 continue if token in {"-u", "--unset"} and i + 1 < len(argv): i += 2 continue if token.startswith("-u") and token != "-u": i += 1 continue if token.startswith("--unset="): i += 1 continue if token.startswith("-"): i += 1 continue if ENV_ASSIGNMENT_RE.match(token): i += 1 continue break return argv[i:] def npm_script_args(rest: list[str]) -> list[str]: value_options = {"--prefix", "--workspace", "-w", "--filter", "--cwd", "-C"} i = 0 while i < len(rest): arg = rest[i] if arg in value_options: i += 2 continue if arg.startswith("-"): i += 1 continue break return rest[i:] def is_noisy_command(argv: list[str]) -> bool: argv = strip_env_prefix(argv) if not argv: return False first = command_basename(argv[0]) rest = argv[1:] if first in {"npm", "pnpm", "yarn", "bun"}: script_args = npm_script_args(rest) if not script_args: return False command = script_args[0] if command == "test": return True if command in {"run", "run-script"} and len(script_args) > 1: script = script_args[1] return script == "build" or script == "lint" or script.startswith("test") return command in {"build", "lint"} if first in {"pytest", "tox", "jest", "vitest"}: return True if first == "npx" and any(arg in {"jest", "vitest"} for arg in rest): return True if re.fullmatch(r"python(?:\d+(?:\.\d+)?)?", first) and len(argv) > 2 and argv[1] == "-m" and argv[2] in {"pytest", "unittest"}: return True if first == "go" and "test" in rest: return True if first == "cargo" and "test" in rest: return True if first in {"mvn", "mvnw"} and "test" in rest: return True if first in {"gradle", "gradlew"} and "test" in rest: return True if first == "make" and any(arg in {"test", "build", "lint"} for arg in rest): return True return False def _skip_leading_flags(rest: list[str]) -> list[str]: """rest 의 앞쪽 `-`/`--` 플래그(와 value-taking 플래그의 다음 토큰)를 건너뛴다. value-taking flag 목록(`_VALUE_TAKING_FLAGS`)에 들지 않은 `-`-시작 토큰은 boolean 이라고 가정한다. 알 수 없는 value flag 는 매칭 누락으로 이어지지만, 그래도 upper layer 가 미가공 명령으로 떨어뜨리는 안전한 degrade 이므로 보수적으로 처리. """ i = 0 while i < len(rest): token = rest[i] if not token.startswith("-"): break if "=" in token: i += 1 continue if token in _VALUE_TAKING_FLAGS and i + 1 < len(rest): i += 2 else: i += 1 return rest[i:] def is_dir_traversal_command(argv: list[str]) -> bool: """순수 path-listing 형태의 `find` / `tree` 만 trim wrapper 라우팅 대상. `find` 가 `-exec` / `-delete` / `-fprint*` 등 임의 명령 출력을 만들어내는 액션을 포함하면 `.env` 같은 자격증명 내용을 흘릴 수 있으므로 본 함수는 False 를 반환하고, `is_log_streaming_command` 가 sanitize 라우팅으로 대신 잡는다. `tree` 는 본질적으로 출력 형식이 fixed 이라 별도 분기가 없다. """ argv = strip_env_prefix(argv) if not argv: return False first = command_basename(argv[0]) rest = argv[1:] if first == "tree": return True if first == "find": return not any(arg in _FIND_OUTPUT_RISK_ACTIONS for arg in rest) if first == "fd": return True if first == "rg" and any(arg == "--files" for arg in rest): return True return False def is_log_streaming_command(argv: list[str]) -> bool: """Production 로그 스트림 / 자격증명을 흘릴 수 있는 명령은 sanitize wrapper 로 라우팅. 대상: - `kubectl logs` / `oc logs` / `podman logs` - `docker logs` / `docker compose logs` / `docker stack logs` / `podman compose|stack logs` - `docker-compose logs` (v1) - `journalctl` (systemd 로그, secret bearing 가능) - `find` 가 `-exec` / `-delete` / `-fprint` 같은 임의 출력 액션을 포함하는 형태 글로벌 옵션 (`-n prod`, `--context=stage`, `-f compose.yml`) 도 `_skip_leading_flags` 로 흡수한다. 한계: `kubectl exec ... -- cat /var/log/...` 같은 우회는 별도 룰이 필요하며 여기서는 처리하지 않는다. """ argv = strip_env_prefix(argv) if not argv: return False first = command_basename(argv[0]) rest = argv[1:] if first == "journalctl": return True if first == "find" and any(arg in _FIND_OUTPUT_RISK_ACTIONS for arg in rest): return True if first in {"kubectl", "oc"}: rest = _skip_leading_flags(rest) return bool(rest) and rest[0] == "logs" if first == "docker-compose": rest = _skip_leading_flags(rest) return bool(rest) and rest[0] == "logs" if first in {"docker", "podman"}: rest = _skip_leading_flags(rest) if not rest: return False sub = rest[0] if sub == "logs": return True if sub in {"compose", "stack"}: rest = _skip_leading_flags(rest[1:]) return bool(rest) and rest[0] == "logs" return False def _env_prefix_name(word: MiniShellWord) -> str | None: """할당 word 의 소스 텍스트에서 `=` 앞 변수 이름만 뽑아낸다. `word.assignment_index` 는 `_exact_assignment_index` 가 `source_value` 기준으로 확정한 활성(비인용) `=` 의 위치다. 그 교차 필드 불변식이 깨진 word 는 이름을 신뢰할 수 없으므로 `None` 을 돌려 호출자가 fail-closed 로 처리하게 한다. `source_value[:None]` 이 토큰 전체를 조용히 돌려주는 파이썬 슬라이스 특성 때문에 불변식 위반이 무증상으로 통과하지 않도록 명시적으로 막는다. """ index = word.assignment_index if index is None or not 0 <= index < len(word.source_value): return None if word.source_value[index] != "=": return None return word.source_value[:index] def _is_unmodeled_assignment_prefix(word: MiniShellWord) -> bool: """bash 는 환경 접두사로 적용하지만 MiniShell 이 할당으로 표시하지 않는 형태인가. `NAME+=VALUE` 는 bash 가 접두사 할당으로 실제 적용하지만(실측 확인), `_exact_assignment_index` 는 `=` 앞이 `NAME+` 라서 이름 문법을 만족하지 못해 `assignment_index` 를 남기지 않는다. 그 결과 이 word 는 할당이 아니라 명령어로 취급되어 FIX-5 이름 검사를 통째로 건너뛴다. 모델링하지 못하는 할당 형태는 안전을 증명할 수 없으므로 fail-closed 로 거부한다. 인용된 형태(`"FOO"+=x`)는 bash 가 할당으로 보지 않으므로 대상이 아니다 — `_exact_assignment_index` 와 동일한 활성/배리어 규칙을 적용한다. """ if word.assignment_index is not None: return False match = MINISHELL_APPEND_ASSIGNMENT_RE.match(word.source_value) if match is None: return False equals_index = match.end() - 1 if not all(word.active[: equals_index + 1]): return False return not any(boundary <= equals_index for boundary in word.barriers) def _env_operand_name(word: MiniShellWord) -> str | None: """`env` 피연산자에서 환경변수 이름을 뽑는다 — 셸 인용을 무시한다. coreutils `env` 는 셸 할당 문법을 검사하지 않는다. 인용 제거가 끝난 argv 원소가 `=` 를 포함하기만 하면 그대로 putenv() 한다. 따라서 셸이 할당으로 보지 않는 `env 'GIT_EXTERNAL_DIFF'=/tmp/evil.sh git diff` 나 `env NAME\\=v cmd` 도 실제로는 환경에 적용된다(실측 확인). `assignment_index` 는 인용된 문자를 비활성으로 보고 할당 표시를 남기지 않으므로, `env` 피연산자 구간에서는 인용이 제거된 `word.value` 를 기준으로 이름을 다시 판정해야 한다. `=` 가 없으면 그 word 가 곧 실행할 명령어이므로 `None` 을 돌려 소비를 멈춘다. """ equals_index = word.value.find("=") if equals_index <= 0: return None return word.value[:equals_index] def _has_unsafe_env_prefix_name( words: tuple[MiniShellWord, ...], start: int, end: int, ) -> bool: """[start, end) 구간의 환경변수 할당 이름이 시드 화이트리스트 밖이면 True. 정확 이름 일치만 검사한다(접두사/글롭 금지) — `TERM*` 글롭이 `TERMINFO` 를 재승인시키는 실패 형태를 피하기 위함(AC-5.6). 이름을 추출할 수 없는 word 는 안전을 증명할 수 없으므로 unsafe 로 간주한다(fail-closed). """ for index in range(start, end): name = _env_prefix_name(words[index]) if name is None or name not in MINISHELL_ALLOWED_ENV_PREFIX_NAMES: return True return False def _routing_start( words: tuple[MiniShellWord, ...], argv: tuple[str, ...], ) -> int: """라우팅이 시작되는 word 인덱스를 계산한다. 반환값 의미: `>= 0` 은 라우팅 시작 인덱스, `-1` 은 기존 `restricted_env_denied` (`env` 뒤에 알 수 없는 플래그가 오거나, `env` 뒤에 명령어 word 자체가 없는 경우), `-2` 는 신규 `unsafe_env_name_denied`(FIX-5 — 접두사 변수 이름이 화이트리스트 밖 이거나, 모델링하지 못하는 접두사 할당 형태). 두 원인은 §5.4/§5.6 측정이 `reason_code` 로 필터링하므로 호출자가 구분해서 처리해야 한다(classify_command 참고). 음수 센티넬을 인덱스로 다시 쓰면 파이썬 음수 인덱싱 때문에 조용히 잘못된 word 를 가리키므로, 모든 호출부는 인덱싱 전에 `< 0` 을 먼저 검사해야 한다. """ index = 0 saw_env = False # 각 반복은 `env` 또는 `--` 를 최소 한 개 소비하므로 word 수만큼이면 충분하다. # PreToolUse 훅 안에서 도는 코드라 구조적 종료 보장을 명시한다(무한 루프 = 행). for _ in range(len(words) + 1): assignment_start = index while index < len(words) and words[index].assignment_index is not None: index += 1 # 이름 검사는 어떤 조기 반환보다도 먼저 수행한다. 명령어 없는 할당 전용 # 세그먼트(`PATH=/tmp/evil`)도 `assignment_only_denied` 라는 다른 백스톱에 # 의존하지 않고 자신의 원인 코드로 거부되어야 §5.4/§5.6 측정이 눈을 뜬다. if _has_unsafe_env_prefix_name(words, assignment_start, index): return -2 # 모델링하지 못하는 접두사 할당(`NAME+=VALUE`)이 라우팅 헤드 자리에 오면 # 이름 검사를 건너뛴 채 명령어로 취급되므로 여기서 fail-closed 로 막는다. if index < len(words) and _is_unmodeled_assignment_prefix(words[index]): return -2 if saw_env: # coreutils `env` 문법은 `env [옵션]... [--] [NAME=VALUE]... [명령]` 이며 # `--` 는 할당 목록의 앞뒤 어느 쪽에도 올 수 있다. `--` 를 소비한 뒤에도 # 할당이 이어질 수 있으므로 루프 선두로 돌아가 이름 검사를 다시 수행한다. if index < len(words) and argv[index] == "--": index += 1 continue # `env` 피연산자는 셸 할당 문법이 아니라 "`=` 를 포함한 argv 원소" 규칙을 # 따른다. 인용으로 셸 할당 표시를 피한 형태도 env 가 그대로 적용하므로 # 인용 제거된 value 기준으로 한 번 더 검사한다(§_env_operand_name). if index < len(words): operand_name = _env_operand_name(words[index]) if operand_name is not None: if operand_name not in MINISHELL_ALLOWED_ENV_PREFIX_NAMES: return -2 index += 1 continue # 이름 문제가 아닌 미지의 `env` 플래그는 기존 원인을 유지한다. if index >= len(words) or argv[index].startswith("-"): return -1 if index >= len(words): return index if command_basename(argv[index]) != "env": return index # `env env NAME=VALUE cmd` 같은 중첩 호출도 각 단계마다 할당 구간을 검사한다. index += 1 saw_env = True # 도달 불가(매 반복이 word 를 최소 하나 소비한다). 방어적으로 fail-closed. return -1 def _routing_start_index(parsed: MiniShellParse) -> int: return _routing_start(parsed.words, parsed.argv) def _routing_argv(parsed: MiniShellParse) -> tuple[str, ...]: """라우팅 대상 argv. 거부 센티넬(`-1`/`-2`)은 빈 튜플로 fail-closed 처리한다. 센티넬을 그대로 슬라이스하면 파이썬 음수 인덱싱 때문에 `argv[-2:]` 가 마지막 두 토큰을 조용히 돌려주어, 불변식 위반이 예외가 아니라 "잘못된 word 에 대한 라우팅 결정"으로 둔갑한다. """ route_start = _routing_start_index(parsed) if route_start < 0: return () return parsed.argv[route_start:] def _wrapper_invocation(argv: tuple[str, ...]) -> tuple[str, int] | None: if not argv: return None # Incoming wrappers are recognized before the general command-identity # gate. Their generated envelopes intentionally contain absolute helper # paths, so this narrow recursion guard must inspect the real basename. head_basename = os.path.basename(argv[0]) if head_basename in WRAPPER_BASENAMES: return head_basename, 0 if ( re.fullmatch(r"python(?:\d+(?:\.\d+)?)?", head_basename) and len(argv) > 1 and os.path.basename(argv[1]) in WRAPPER_BASENAMES ): return os.path.basename(argv[1]), 1 if ( re.fullmatch(r"python(?:\d+(?:\.\d+)?)?", head_basename) and len(argv) > 2 and argv[1] == "-I" and os.path.basename(argv[2]) in WRAPPER_BASENAMES ): return os.path.basename(argv[2]), 2 return None def _wrapper_kind(basename: str) -> str: return "sanitize" if "sanitize" in basename else "trim" def _expected_cgw1_prefix(kind: str) -> tuple[str, ...]: script_dir = os.path.dirname(os.path.abspath(__file__)) if os.path.basename(__file__) == "rewrite_bash_for_token_budget.py": helper = "sanitize_output.py" if kind == "sanitize" else "trim_command_output.py" return ("python3", os.path.join(script_dir, helper)) helper = ( "context-guard-sanitize-output" if kind == "sanitize" else "context-guard-trim-output" ) return (os.path.join(script_dir, helper),) def _is_expected_direct_wrapper_path(argv: tuple[str, ...]) -> bool: """Whether argv starts with this package's exact generated helper path. Direct helper CLI use predates F-11 and remains ordinary. The exception is deliberately limited to the helper beside this entrypoint; an attacker-chosen path that merely shares its basename is not trusted. """ invocation = _wrapper_invocation(argv) if invocation is None or invocation[1] != 0: return False basename, _wrapper_index = invocation expected = _expected_cgw1_prefix(_wrapper_kind(basename)) return len(expected) == 1 and argv[0] == expected[0] def classify_incoming_wrapper( parsed: MiniShellParse, ) -> tuple[str, str | None, str | None] | None: """Classify raw wrapper input without probing the filesystem. Direct wrapper CLI use is not an execution envelope. A known wrapper combined with the reserved CGW1 sentinel or an exact v0 shell envelope is always incoming execution syntax and therefore denied at PreToolUse. """ if len(parsed.segments) != 1: return None route_start = _routing_start_index(parsed) if route_start < 0: return None route_argv = parsed.argv[route_start:] invocation = _wrapper_invocation(route_argv) if invocation is None: return None basename, wrapper_index = invocation kind = _wrapper_kind(basename) envelope_argv = route_argv[wrapper_index + 1:] sentinel_tokens = [ token for token in envelope_argv if CGW1_SENTINEL in token ] if sentinel_tokens: code = ( "nested_wrapper_denied" if len(sentinel_tokens) > 1 or any(token != CGW1_SENTINEL for token in sentinel_tokens) else "incoming_wrapper_denied" ) return (code, kind, None) # `--max-lines` 값은 고정하지 않는다. 값 하나만 바꾼 위조 봉투가 예전에는 # 라우팅 표 밖이라는 이유로 우연히 거부됐는데, 이제 그 우연이 사라졌다. # 봉투를 봉투로 알아보지 못하면 호스트 허용목록이 정규 래퍼 argv 형태를 # 신뢰하고 있을 때 임의 명령의 프롬프트가 함께 억제될 수 있다. # # 직접 CLI 사용(`context-guard-trim-output --max-lines 10 -- pytest`)은 # 여전히 통상 라우트를 탄다 — 아래 매칭이 격리된 runtime shell argv 를 # 그대로 요구하기 때문이다. 사람이 손으로 적는 형태가 아니다. prefix_matchers: tuple[tuple[object, ...], ...] = ( ("--max-lines", _ANY_TOKEN), # 기본 trim 라우트가 escrow 플래그를 달고 나가므로 그 형태도 봉투로 알아봐야 # 한다. 알아보지 못하면 재진입 방지 거부가 조용히 사라진다. ("--max-lines", _ANY_TOKEN, *CGW1_ESCROW_FLAGS), (CGW1_COMMAND_SEARCH_DIFF,), ("--mode", CGW1_COMMAND_SEARCH_DIFF), ) shell_argvs = (CGW1_SHELL_ARGV, _runtime_shell_argv()) for prefix in prefix_matchers: for shell_argv in shell_argvs: fixed = (*prefix, "--", *shell_argv) if len(envelope_argv) != len(fixed) + 1: continue if all( expected is _ANY_TOKEN or expected == actual for expected, actual in zip(fixed, envelope_argv[:-1]) ): return ("incoming_wrapper_denied", kind, envelope_argv[-1]) return None def is_already_wrapped(argv: list[str]) -> bool: """이 argv 가 이미 CGW1 실행 봉투를 지고 있는지 본다. 예전에는 `"exact"` 상태를 확인했는데 `classify_incoming_wrapper` 는 그 값을 한 번도 돌려주지 않는다 - `None`, `"incoming_wrapper_denied"`, `"nested_wrapper_denied"` 뿐이다. 그래서 이 헬퍼는 입력과 무관하게 항상 False 였고, 유일한 호출자인 테스트가 `assertFalse` 라 그 사실이 드러나지 않았다. 분류기가 봉투로 인정한 것을 그대로 인정한다. """ command = shell_join(argv) parsed = parse_minishell(command) if parsed.denial_reason is not None: return False return classify_incoming_wrapper(parsed) is not None def is_sanitizable_output_command(argv: list[str]) -> bool: argv = strip_env_prefix(argv) if not argv: return False first = command_basename(argv[0]) rest = argv[1:] if first in {"rg", "grep", "egrep", "fgrep"}: # `rg --files` is path listing rather than content search; the large # read/diet guards are better fits there. return not any(arg == "--files" for arg in rest) if first == "git" and rest: rest = git_subcommand_args(rest) if not rest: return False subcommand = rest[0] if subcommand == "grep": return True if subcommand in {"diff", "show"}: return True if subcommand == "log" and any(arg == "-p" or arg.startswith("--patch") for arg in rest[1:]): return True return False def git_subcommand_args(rest: list[str]) -> list[str]: value_options = {"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path", "--config-env"} i = 0 while i < len(rest): token = rest[i] if token == "--": return rest[i + 1:] if token in value_options and i + 1 < len(rest): i += 2 continue if any(token.startswith(prefix + "=") for prefix in value_options if prefix.startswith("--")): i += 1 continue if token in {"--no-pager", "--paginate", "--bare", "--literal-pathspecs", "--no-optional-locks"}: i += 1 continue if token.startswith("-"): i += 1 continue break return rest[i:] def _valid_n(value: str) -> bool: return value.isascii() and value.isdigit() and 1 <= int(value) <= 1_000_000 def _valid_range(value: str) -> bool: if ( not value or len(value.encode("utf-8")) > 64 or not value.isascii() ): return False return all( ( _valid_n(item) if "-" not in item else ( item.count("-") == 1 and _valid_n(item.split("-", 1)[0]) and ( not item.split("-", 1)[1] or _valid_n(item.split("-", 1)[1]) ) ) ) for item in value.split(",") ) def _valid_key(value: str) -> bool: parts = value.split(",") return ( 1 <= len(parts) <= 2 and all( part.isascii() and part.isdigit() and 1 <= len(part) <= 6 and int(part) >= 1 for part in parts ) ) def _printf_is_safe(argv: tuple[str, ...]) -> bool: if len(argv) < 2: return False index = 1 if argv[index] == "--": index += 1 elif argv[index].startswith("-"): return False return index < len(argv) _LS_SHORT_FLAGS = set("laAhtrS1dFpRincu") _LS_LONG_FLAGS = { "--all", "--almost-all", "--human-readable", "--reverse", "--recursive", "--directory", "--classify", "--group-directories-first", "--color=never", "--color=auto", "--no-group", } def _ls_is_safe(argv: tuple[str, ...]) -> bool: """`ls`가 producer 라우트로 허용되기에 안전한지 판단하는 순수 허용목록. 값(value)을 소비하는 `ls` 플래그가 존재하지 않고, 부수효과를 갖는 플래그도 없다는 성질 덕분에 짧은 플래그 클러스터(`-ltrh`)까지 안전하게 분해할 수 있다. 이 성질은 `sed`/`git`에는 성립하지 않으므로 이 패턴을 일반화하지 말 것 (설계 문서 4.1 참고). """ options_done = False for argument in argv[1:]: if not options_done and argument == "--": options_done = True continue if options_done: continue if argument.startswith("--"): if argument not in _LS_LONG_FLAGS: return False continue if argument.startswith("-") and argument != "-": if not set(argument[1:]).issubset(_LS_SHORT_FLAGS): return False continue return True def _cat_is_safe(argv: tuple[str, ...], *, allow_files: bool) -> bool: operands = 0 options_done = False for argument in argv[1:]: if not options_done and argument == "--": options_done = True continue if ( not options_done and argument.startswith("-") and argument != "-" ): if ( argument.startswith("--") or not argument[1:] or not set(argument[1:]).issubset(set("bnsETAvet")) ): return False continue operands += 1 return allow_files or operands == 0 def _cut_is_safe(argv: tuple[str, ...]) -> bool: selector: str | None = None delimiter = False index = 1 while index < len(argv): argument = argv[index] if argument == "--": return selector is not None and index == len(argv) - 1 if argument in {"-s", "--complement"}: index += 1 continue if argument in {"-f", "-c", "-b", "-d"}: if index + 1 >= len(argv): return False value = argv[index + 1] if argument == "-d": if delimiter or len(value.encode("utf-8")) != 1: return False delimiter = True else: if selector is not None or not _valid_range(value): return False selector = argument index += 2 continue if ( len(argument) > 2 and argument[:2] in {"-f", "-c", "-b", "-d"} ): option, value = argument[:2], argument[2:] if option == "-d": if delimiter or len(value.encode("utf-8")) != 1: return False delimiter = True else: if selector is not None or not _valid_range(value): return False selector = option index += 1 continue return False return selector is not None and (not delimiter or selector == "-f") _SED_SEGMENT_PATTERN = r"(?:[1-9]\d*|[1-9]\d*,(?:[1-9]\d*|\$))p" _SED_MAX_SEGMENTS = 8 _SED_SCRIPT_RE = re.compile( rf"{_SED_SEGMENT_PATTERN}(?:;{_SED_SEGMENT_PATTERN})" rf"{{0,{_SED_MAX_SEGMENTS - 1}}}" ) def _sed_route_shape(argv: tuple[str, ...]) -> tuple[bool, int]: """`sed`의 인자를 전체 스캔해 (안전 여부, 파일 피연산자 수)를 반환한다. design route-readmission-design-20260729.md §2.3 의 정확한 재현이다. 세 가지 문법 함정을 여기서 막는다: 1) 스크립트 위치는 조건부다 — `-e`/`--expression=` 이 하나라도 있으면 모든 피연산자가 파일이고, 없으면 첫 피연산자가 스크립트다. 2) GNU sed 는 옵션을 순열(permute)한다 — `sed -n '1,5p' -i f` 처럼 피연산자 뒤에도 옵션이 온다. 그래서 접두부만 훑는 스캔이 아니라 `--` 전까지 argv 전체를 훑어야 `-i` 를 놓치지 않는다. 3) 짧은 옵션 클러스터는 `-i` 를 밀반입할 수 있다(`-ni` == `-n -i`). `set(arg[1:]).issubset(allowed)` 패턴은 여기서 틀리다 — 클러스터는 전부 거부하고 정확한 토큰만 허용한다. 스크립트 본문 자체의 안전 경계(`w`/`W`/`s///w`/`e`/`s///e`/`r`/`R` 배제)는 `_SED_SCRIPT_RE` 의 `re.fullmatch` 가 담당한다. S009는 기존의 안전한 p-only SEG를 `;`로 최대 `_SED_MAX_SEGMENTS` 개 조합할 뿐 SEG 자체를 넓히지 않는다 — 이 함수는 그 정규식이 실제로 검사하는 대상이 진짜 스크립트임을 보장하는 역할만 한다. """ quiet_seen = False expressions: list[str] = [] operands: list[str] = [] options_done = False index = 1 while index < len(argv): token = argv[index] if options_done: operands.append(token) index += 1 continue if token == "--": options_done = True index += 1 continue if token in {"-n", "--quiet", "--silent"}: if quiet_seen: return False, 0 quiet_seen = True index += 1 continue if token == "-e": if index + 1 >= len(argv): return False, 0 expressions.append(argv[index + 1]) index += 2 continue if token.startswith("--expression="): expressions.append(token.split("=", 1)[1]) index += 1 continue if token.startswith("-") and token != "-": return False, 0 operands.append(token) index += 1 if not quiet_seen: return False, 0 if len(expressions) > 1: return False, 0 if expressions: script, files = expressions[0], operands elif operands: script, files = operands[0], operands[1:] else: return False, 0 if _SED_SCRIPT_RE.fullmatch(script) is None: return False, 0 if not all(_valid_n(number) for number in re.findall(r"\d+", script)): return False, 0 return True, len(files) def _sort_is_safe(argv: tuple[str, ...]) -> bool: index = 1 while index < len(argv): argument = argv[index] if argument == "--": return index == len(argv) - 1 if argument in {"-k", "-t"}: if index + 1 >= len(argv): return False value = argv[index + 1] if ( argument == "-k" and not _valid_key(value) ) or ( argument == "-t" and len(value.encode("utf-8")) != 1 ): return False index += 2 continue if argument in {"-r", "-u", "-n", "-f", "-s"}: index += 1 continue if argument.startswith(("-k", "-t")) and len(argument) > 2: value = argument[2:] if ( argument.startswith("-k") and not _valid_key(value) ) or ( argument.startswith("-t") and len(value.encode("utf-8")) != 1 ): return False index += 1 continue return False return True def _uniq_is_safe(argv: tuple[str, ...]) -> bool: index = 1 while index < len(argv): argument = argv[index] if argument == "--": return index == len(argv) - 1 if argument in {"-c", "-d", "-u", "-i"}: index += 1 continue if argument in {"-f", "-s", "-w"}: if index + 1 >= len(argv) or not _valid_n(argv[index + 1]): return False index += 2 continue if ( len(argument) > 2 and argument[:2] in {"-f", "-s", "-w"} and _valid_n(argument[2:]) ): index += 1 continue return False return True def _wc_is_safe(argv: tuple[str, ...], *, allow_files: bool) -> bool: """wc 인자가 안전한 라우팅 대상인지 판정한다. 플래그는 -c/-l/-m/-w 조합만 허용한다. 파일 피연산자는 `_cat_is_safe`(:1138)와 대칭으로 `allow_files`가 True일 때만 허용한다 — role이 "filter"(파이프 중간)면 stdin만 읽어야 하므로 파일 인자를 거부해야 한다. `--` 이후 토큰은 전부 피연산자로 취급한다(pathspec 구분자와 동일한 관례). """ operands = 0 options_done = False for argument in argv[1:]: if not options_done and argument == "--": options_done = True continue if not options_done and argument.startswith("-") and argument != "-": if ( argument.startswith("--") or not argument[1:] or not set(argument[1:]).issubset({"c", "l", "m", "w"}) ): return False continue operands += 1 return allow_files or operands == 0 def _tail_follows_forever(argv: tuple[str, ...]) -> bool: """tail 인자 어디에든 follow 옵션이 있는지 본다. GNU tail 은 옵션을 순열한다. `tail some.log -f` 도 follower 다. 옵션 훑기를 첫 위치인자에서 멈추면 그 형태가 안전으로 판정되어 래핑되고, 자식이 끝나지 않아 워치독까지 턴이 멈춘다. 같은 함정을 이 파일은 sed 와 shortlog 에서 이미 고쳤다. `--` 뒤는 전부 피연산자다. `tail -- -f` 의 `-f` 는 파일 이름이므로 보지 않는다. """ for argument in argv[1:]: if argument == "--": return False if argument.startswith("--"): if argument.split("=", 1)[0] in {"--follow", "--retry"}: return True continue if argument.startswith("-") and len(argument) > 1: # 짧은 옵션은 묶일 수 있다: `-fn 20`, `-qf`. 묶음 안의 f/F 도 follow 다. if any(letter in {"f", "F"} for letter in argument[1:]): return True return False def _head_tail_is_safe(argv: tuple[str, ...], *, allow_files: bool) -> bool: """head/tail 인자가 안전한 라우팅 대상인지 판정한다. `-n`/`--lines`(및 `-N`/`-nN`/`--lines=N` 축약형)는 최대 1회만 허용하며 유효한 양의 정수여야 한다. **`-n` 미지정도 허용한다** — bare `head`/`tail`은 기본 10줄 상한이 이미 적용되므로 무제한 출력 위험이 없다. `tail -f`/`-F`는 무제한 스트림이므로 allow_files 여부와 무관하게 항상 거부한다(`bash -c` 내부에서 프로세스가 종결되지 않는 것을 방지). `-c`(바이트 단위)는 지원하지 않는다 — trim 예산 단위는 줄(line)이라 바이트 상한과 섞일 수 없다. """ first = command_basename(argv[0]) if first == "tail" and _tail_follows_forever(argv): return False index = 1 count_seen = False while index < len(argv): argument = argv[index] if argument == "--": index += 1 break if first == "tail" and argument in {"-f", "-F"}: return False if argument in {"-n", "--lines"}: if count_seen or index + 1 >= len(argv) or not _valid_n(argv[index + 1]): return False count_seen = True index += 2 continue attached = re.fullmatch(r"(?:-|(?:-n)|(?:--lines=))([1-9]\d*)", argument) if attached is not None: if count_seen or not _valid_n(attached.group(1)): return False count_seen = True index += 1 continue if argument.startswith("-"): return False break return allow_files or index == len(argv) #: grep 긴 옵션(long flag) 중 이미 허용된 짧은 옵션과 동치인 것만 정확히 나열한 #: 화이트리스트. **접두사(startswith) 매칭 금지** — `--color`로 시작 매칭을 허용하면 #: `--color=always`(ANSI 이스케이프 주입)가, `--f`류 접두사 매칭을 허용하면 #: `--file=`(패턴을 파일에서 읽음, 예측 불가능한 I/O)이 함께 통과해버린다. #: #: 이 표는 **짧은 옵션 동치 규칙을 예외 없이** 지킨다. `--no-messages`는 그 짧은 #: 형태 `-s`가 `allowed_flags` 밖이라 표에서 뺐다 — `-s` 자체는 stderr 진단만 #: 억제해 위험하지 않지만, 규칙에 예외를 하나 두면 주석이 거짓이 되고 거짓 주석은 #: 이 저장소에서 결함이 전파되는 경로다. `-s`를 허용하기로 결정한다면 짧은 옵션 #: 쪽을 먼저 넓히고 그 다음 이 표에 롱 형태를 추가한다. #: #: 값 형태를 취하는 옵션(`--exclude=`, `--exclude-dir=`, `--devices=`, #: `--directories=`, `--label=`, `--binary-files=`, `-D/-U/-z/-Z/--null` 계열)은 #: 의도적으로 제외한다. 이유는 값이 동작을 바꾸기 때문이다(예: `--directories=read`). #: #: S010은 incidence gate를 통과한 bare recursive `grep`의 정확히 한 개 #: `--include=`만 아래 별도 값 문법으로 다룬다. 이 옵션은 정확 일치 이름, #: 제한된 ASCII basename grammar, recursive/file-operand 조건을 모두 만족해야 하며 #: 이 별칭 표에는 들어오지 않는다. 다른 값 옵션과 `--include*` 근접 철자는 계속 #: exact-match fail-closed 규칙을 따른다. _GREP_LONG_ALIASES = frozenset( { "--only-matching", "--count", "--files-with-matches", "--files-without-match", "--line-number", "--with-filename", "--no-filename", "--ignore-case", "--invert-match", "--word-regexp", "--line-regexp", "--extended-regexp", "--fixed-strings", "--basic-regexp", "--perl-regexp", "--quiet", "--silent", # `--recursive`는 표에 두지 않는다 — 조회보다 앞선 독립 분기가 이미 # 처리하므로 여기 넣으면 도달 불가능한 중복 항목이 되고, 모든 항목이 # 하중을 받아야 한다는 성질이 깨진다. "--dereference-recursive", "--color=never", "--color=auto", } ) _GREP_INCLUDE_GLOB_RE = re.compile(r"[A-Za-z0-9._*?-]+\Z", re.ASCII) def _grep_include_glob_is_safe(value: str) -> bool: return ( 1 <= len(value) <= 96 and not value.startswith("-") and _GREP_INCLUDE_GLOB_RE.fullmatch(value) is not None and re.search(r"[A-Za-z0-9._]", value, re.ASCII) is not None ) def _grep_is_safe( argv: tuple[str, ...], *, allow_files: bool, allow_include: bool = False, ) -> bool: pattern_seen = False files = 0 include_seen = False recursive_seen = False stdin_operand_seen = False allowed_flags = set("nHhivEFGPwxcolLrRq".replace(" ", "")) index = 1 while index < len(argv): argument = argv[index] if argument == "--": index += 1 break if argument.startswith("--include"): if ( not allow_include or include_seen or not argument.startswith("--include=") or not _grep_include_glob_is_safe(argument.split("=", 1)[1]) ): return False include_seen = True index += 1 continue if argument in {"-f", "--file"} or argument.startswith(("--file=", "--binary-files=")): return False if argument == "-e": if index + 1 >= len(argv): return False pattern_seen = True index += 2 continue if argument in {"-m", "--max-count", "-A", "-B", "-C"}: if index + 1 >= len(argv) or not _valid_n(argv[index + 1]): return False index += 2 continue if argument.startswith("--max-count="): if not _valid_n(argument.split("=", 1)[1]): return False index += 1 continue if re.fullmatch(r"-(?:m|A|B|C)([1-9]\d*)", argument): if not _valid_n(argument[2:]): return False index += 1 continue if argument == "--recursive": recursive_seen = True index += 1 continue if argument in _GREP_LONG_ALIASES: if argument == "--dereference-recursive": recursive_seen = True index += 1 continue if argument.startswith("-") and argument != "-": if ( argument.startswith("--") or not argument[1:] or not set(argument[1:]).issubset(allowed_flags) ): return False if "r" in argument[1:] or "R" in argument[1:]: recursive_seen = True index += 1 continue if not pattern_seen: pattern_seen = True else: files += 1 stdin_operand_seen = stdin_operand_seen or argument == "-" index += 1 while index < len(argv): if not pattern_seen: pattern_seen = True else: files += 1 stdin_operand_seen = stdin_operand_seen or argv[index] == "-" index += 1 if include_seen: return ( allow_files and recursive_seen and pattern_seen and files > 0 and not stdin_operand_seen ) return pattern_seen and (allow_files or files == 0) def _rg_is_safe(argv: tuple[str, ...]) -> bool: pattern_seen = False options_done = False index = 1 allowed_short = { "-n", "-H", "-h", "-i", "-S", "-F", "-w", "-x", "-l", "-c", } allowed_long = { "--line-number", "--with-filename", "--no-filename", "--ignore-case", "--smart-case", "--fixed-strings", "--word-regexp", "--line-regexp", "--files-with-matches", "--count", "--hidden", "--no-ignore", } while index < len(argv): argument = argv[index] if not options_done and argument == "--": options_done = True index += 1 continue if not options_done and argument in allowed_short | allowed_long: index += 1 continue if not options_done and argument in {"-g", "--glob"}: if index + 1 >= len(argv): return False index += 2 continue if not options_done and ( (argument.startswith("-g") and len(argument) > 2) or argument.startswith("--glob=") ): index += 1 continue if not options_done and argument.startswith("-"): return False pattern_seen = True index += 1 return pattern_seen GIT_TABLE_SUBCOMMANDS = frozenset({ "status", "log", "branch", "tag", "remote", "rev-parse", "describe", "ls-files", "shortlog", "blame", "stash", "diff", "show", "grep", }) """§6.1b 12행 쌍 화이트리스트가 다루는 git 서브커맨드 집합 — `diff`/`show`/`grep`은 한 표 행을 공유하므로 14개 서브커맨드가 12행이 된다(FIX-6이 `remote`행을 재도입해 11행 -> 12행). 오라클 `git-*` family 집합과의 동치 검증(AC-1b.3, R-11)이 이 상수를 그대로 참조한다 — 행을 늘리고 family를 빠뜨리면 그 테스트가 실패한다.""" def _git_flags_and_positionals( arguments: tuple[str, ...], *, long_flags: frozenset[str], short_flags: frozenset[str], ) -> int | None: """옵션을 소비하며 위치 인자 개수를 반환한다. 미지 플래그면 `None`. `--` 토큰 자체는 위치 인자로 계수하지 않되, 그 이후 토큰은 옵션 파싱을 끄고 전부 위치 인자로 계수한다(AC-1.10 — `git log a..b -- p1 p2 p3`는 `--`를 빼면 정확히 4개다. 과거 결함은 오버플로가 아니라 이 규칙의 부재였다). 묶음 단축 플래그(`-ad` 등)는 `-`로 시작하는 각 글자가 모두 `short_flags`에 속해야 허용된다(분해 없이 집합 매칭 — AC-1.9). `git branch -ad`는 `{a,d}`로 분해되고 `d`가 branch의 허용 집합에 없어 거부된다(D1 완화가 다시 쓰기를 재승인하지 않는지 확인하는 회귀 핀). """ positionals = 0 options_done = False for argument in arguments: if not options_done and argument == "--": options_done = True continue if options_done: positionals += 1 continue if argument in long_flags: continue if ( argument.startswith("-") and not argument.startswith("--") and argument != "-" and set(argument[1:]).issubset(short_flags) ): continue if argument.startswith("-"): return None positionals += 1 return positionals _GIT_STATUS_LONG_FLAGS = frozenset({ "--short", "--branch", "--porcelain", "--long", "--no-color", "--untracked-files", }) _GIT_STATUS_SHORT_FLAGS = frozenset("sb") def _git_status_is_safe(arguments: tuple[str, ...]) -> bool: """`git status`: 위치 인자 0개(§6.1b 표). `.git/index` stat-cache 갱신은 허용된 부작용이다(AC-1.4 각주) — 이 함수의 쓰기 판정 대상이 아니다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_STATUS_LONG_FLAGS, short_flags=_GIT_STATUS_SHORT_FLAGS, ) return positionals == 0 _GIT_BRANCH_LONG_FLAGS = frozenset({ "--all", "--remotes", "--verbose", "--list", "--show-current", "--no-color", "--sort", }) _GIT_BRANCH_SHORT_FLAGS = frozenset("arv") def _git_branch_is_safe(arguments: tuple[str, ...]) -> bool: """`git branch`: 위치 인자 0개 엄격 — arity가 조회(0개)를 생성(1개+)으로 뒤집는 서브커맨드다(D2 반증 사례, plan §6.1b). `--edit-description` 등 쓰기 플래그는 표에 없어 미지 플래그로 거부된다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_BRANCH_LONG_FLAGS, short_flags=_GIT_BRANCH_SHORT_FLAGS, ) return positionals == 0 _GIT_TAG_LONG_FLAGS = frozenset({"--list", "--sort", "--no-color"}) def _git_tag_is_safe(arguments: tuple[str, ...]) -> bool: """`git tag`: 위치 인자 0개 엄격 — branch와 동일하게 arity가 조회↔생성을 뒤집는다(§6.1b 표). `-n`은 부착형 주석 줄 수만 허용한다 — 분리형 `-n 5`는 다음 토큰 `5`가 미지 위치 인자로 남아 이미 안전하게 거부된다(subcommand별 `-n` 의미 차이, AC-1.9 — log는 분리형 값, tag는 부착형, shortlog는 순수 불리언).""" positionals = 0 for argument in arguments: if argument == "--": continue if argument in _GIT_TAG_LONG_FLAGS or argument in {"-l", "-n"}: continue if re.fullmatch(r"-n[1-9]\d*", argument): continue if argument.startswith("-"): return False positionals += 1 return positionals == 0 _GIT_REMOTE_LONG_FLAGS = frozenset({"--verbose"}) _GIT_REMOTE_SHORT_FLAGS = frozenset("v") def _git_remote_is_safe(arguments: tuple[str, ...]) -> bool: """`git remote`: 위치 인자 0개 엄격 — branch/tag와 동일하게 arity가 조회(0개)를 쓰기(`add`/`remove`/`rename`/`set-url`, 1개+)로 뒤집는다 (§6.1b 표, FIX-6 재도입). `add`/`remove`/`rename`/`set-url`/`get-url` 등 서브서브커맨드는 별도 목록 없이도 위치 인자로 잡혀 자동 거부된다(AC-1.4에 `remote add origin url` deny가 고정돼 있고, 이번 재도입 후에도 그대로다). `-v`/`--verbose`만 허용해 URL을 노출하는 유일한 조회 형태를 표에 올린다 — 이 URL이 자격증명을 담고 있어도 안전한 이유는 FIX-6에서 확장한 `credential_policy.py`의 토큰 전용(콜론 없는) userinfo 리댁션이 담보한다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_REMOTE_LONG_FLAGS, short_flags=_GIT_REMOTE_SHORT_FLAGS, ) return positionals == 0 _GIT_REV_PARSE_LONG_FLAGS = frozenset({ "--abbrev-ref", "--short", "--verify", "--show-toplevel", "--git-dir", "--is-inside-work-tree", "--quiet", }) def _git_rev_parse_is_safe(arguments: tuple[str, ...]) -> bool: """`git rev-parse`: 위치 인자 무제한(revision 문자열, §6.1b 표) — 쓰기가 되지 않는다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_REV_PARSE_LONG_FLAGS, short_flags=frozenset(), ) return positionals is not None _GIT_DESCRIBE_LONG_FLAGS = frozenset({ "--tags", "--always", "--dirty", "--long", "--abbrev", }) def _git_describe_is_safe(arguments: tuple[str, ...]) -> bool: """`git describe`: 위치 인자 무제한(§6.1b 표) — 쓰기가 되지 않는다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_DESCRIBE_LONG_FLAGS, short_flags=frozenset(), ) return positionals is not None _GIT_LS_FILES_LONG_FLAGS = frozenset({ "--cached", "--modified", "--others", "--exclude-standard", "--stage", "--deleted", }) _GIT_LS_FILES_SHORT_FLAGS = frozenset("cmos") def _git_ls_files_is_safe(arguments: tuple[str, ...]) -> bool: """`git ls-files`: 위치 인자 무제한(pathspec 필터, §6.1b 표) — 쓰기가 되지 않는다.""" positionals = _git_flags_and_positionals( arguments, long_flags=_GIT_LS_FILES_LONG_FLAGS, short_flags=_GIT_LS_FILES_SHORT_FLAGS, ) return positionals is not None _GIT_SHORTLOG_LONG_FLAGS = frozenset({ "--summary", "--numbered", "--email", "--no-color", }) _GIT_SHORTLOG_SHORT_FLAGS = frozenset("sne") def _git_shortlog_is_safe(arguments: tuple[str, ...]) -> bool: """`git shortlog`: 위치 인자 무제한이나 리비전 1개 이상 필수(§6.1b 표). `-n`은 여기서 `--numbered`(값을 취하지 않는 순수 불리언)다 — log의 max-count `-n`과 의미가 다르다(subcommand별 `-n` 의미 차이, AC-1.9). **리비전 1개 이상을 요구하는 이유(비종료 방지)**: git shortlog 는 리비전 피연산자가 없으면 커밋 로그를 stdin 에서 읽는다. 재작성 래퍼 (`sanitize_output.py:1052`)는 자식 프로세스에 `stdin=` 을 지정하지 않아 훅의 stdin 을 그대로 상속시키므로, 닫히지 않은 stdin 아래에서 `git shortlog -sn` 은 `DEFAULT_TIMEOUT_SECONDS`(600초) 워치독이 프로세스 그룹을 죽일 때까지 아무 것도 출력하지 않고 블록한다(실측). 이는 `_head_tail_is_safe` 가 `tail -f`/`-F` 를 거부하는 것과 동일한 불변식이며, 승인 범위를 좁히는 방향이므로 표의 보안 태세를 약화하지 않는다. `git shortlog -sn HEAD` 처럼 리비전을 주면 stdin 을 읽지 않고 즉시 끝난다. `--` 이후 토큰은 리비전이 아니라 pathspec 이므로 세지 않는다 — `git shortlog -sn -- README.md` 는 위치 인자가 1개로 보이지만 리비전이 없어 여전히 stdin 을 읽고 블록한다(실측). blame 의 `>=1 path` 규칙과 달리 여기서는 `--` 앞의 리비전만 요건을 충족시킨다. """ if _git_flags_and_positionals( arguments, long_flags=_GIT_SHORTLOG_LONG_FLAGS, short_flags=_GIT_SHORTLOG_SHORT_FLAGS, ) is None: return False revision_arguments = ( arguments[: arguments.index("--")] if "--" in arguments else arguments ) revisions = _git_flags_and_positionals( revision_arguments, long_flags=_GIT_SHORTLOG_LONG_FLAGS, short_flags=_GIT_SHORTLOG_SHORT_FLAGS, ) return revisions is not None and revisions >= 1 def _git_blame_is_safe(arguments: tuple[str, ...]) -> bool: """`git blame`: 위치 인자 무제한이나 경로 1개 이상 필수(§6.1b 표). `-L`은 값을 취한다(부착 `-L10,20` 또는 분리 `-L 10,20` 모두 허용 — 범위 문자열 자체를 검증하지 않아도 안전하다, sanitize 240줄 상한이 출력을 이미 유계화한다).""" positionals = 0 options_done = False index = 0 while index < len(arguments): argument = arguments[index] if not options_done and argument == "--": options_done = True index += 1 continue if not options_done and argument in {"--porcelain", "--line-porcelain", "-w"}: index += 1 continue if not options_done and argument == "-L": if index + 1 >= len(arguments): return False index += 2 continue if not options_done and argument.startswith("-L") and len(argument) > 2: index += 1 continue if not options_done and argument.startswith("-"): return False positionals += 1 index += 1 return positionals >= 1 def _git_stash_is_safe(arguments: tuple[str, ...]) -> bool: """`git stash`: `list`/`show`만 허용, 부가 인자 없는 정확히 그 형태만 — 맨 `git stash`(0-arity writer, D2 반증 사례)와 그 밖의 서브커맨드 (`push`/`pop`/`apply`/`drop`/`clear`/`branch`/`save`)는 표에 없어 거부된다(§6.1b 표).""" return len(arguments) == 1 and arguments[0] in {"list", "show"} _GIT_DIFF_SHOW_BOOLEAN_FLAGS = frozenset({ "-p", "--patch", "--stat", "--name-only", "--name-status", "--no-color", "--color=never", "--cached", "--staged", "--oneline", }) _GIT_CONFIG_EXECUTION_GUARD = ( "GIT_CONFIG_COUNT=1", "GIT_CONFIG_KEY_0=core.fsmonitor", "GIT_CONFIG_VALUE_0=false", ) _GIT_ORIGINAL_COMMAND_ENV = "CONTEXT_GUARD_ORIGINAL_COMMAND" _GIT_GUARD_MODE = "--context-guard-exec-git" _GIT_DIFF_EXECUTION_FLAGS = ("--no-ext-diff", "--no-textconv") _GIT_TEXTCONV_EXECUTION_FLAGS = ("--no-textconv",) _GIT_FILTER_CONFIG_KEY_RE = re.compile( r"^filter\..+\.(?:clean|smudge|process|required)$", re.IGNORECASE, ) _GIT_FILTER_CONFIG_QUERY = r"^filter\..*\.(clean|smudge|process|required)$" _GIT_FILTER_CONFIG_MAX_KEYS = 128 _GIT_FILTER_CONFIG_MAX_BYTES = 65_536 _GIT_FILTER_CONFIG_TIMEOUT_SECONDS = 5 def _git_diff_show_is_safe(arguments: tuple[str, ...]) -> bool: """`git diff`/`git show`: 기존 `_git_is_safe` 경로를 그대로 보존한다 (§6.1b 표 — "기존대로"). 개조 전 `patch_output`은 diff/show에서 항상 `True`로 시작해 끝까지 `False`로 바뀌는 경로가 없었으므로(오직 log에서만 `-p` 요구가 의미 있었다) 여기서는 제거했다 — 동작은 동일하다.""" index = 0 options_done = False while index < len(arguments): argument = arguments[index] if not options_done and argument == "--": options_done = True index += 1 continue if options_done: index += 1 continue if argument in _GIT_DIFF_SHOW_BOOLEAN_FLAGS: index += 1 continue if argument in {"-U", "--unified"}: if index + 1 >= len(arguments) or not _valid_n(arguments[index + 1]): return False index += 2 continue if re.fullmatch(r"-U[1-9]\d*", argument) or ( argument.startswith("--unified=") and _valid_n(argument.split("=", 1)[1]) ): index += 1 continue if argument.startswith("-"): return False index += 1 return True _GIT_LOG_BOOLEAN_FLAGS = frozenset({ "--oneline", "--stat", "--name-only", "--name-status", "--graph", "--decorate", "--no-color", "-p", "--patch", "--reverse", }) _GIT_LOG_VALUE_FLAGS = frozenset({ "--pretty", "--format", "--author", "--since", "--until", }) def _git_log_attached_value_ok(argument: str) -> bool: """`-`/`-U`/`--unified=`/`--max-count=`/`--=…` 부착형이 안전한지 판정한다(AC-1.9 — `git log --oneline -20` 같은 부착형이 거짓 거부되지 않도록 분해 전에 먼저 인식한다).""" if re.fullmatch(r"-[1-9]\d*", argument): return True if re.fullmatch(r"-U[1-9]\d*", argument): return True if argument.startswith("--unified=") and _valid_n(argument.split("=", 1)[1]): return True if argument.startswith("--max-count=") and _valid_n(argument.split("=", 1)[1]): return True return any(argument.startswith(f"{flag}=") for flag in _GIT_LOG_VALUE_FLAGS) def _git_log_is_safe(arguments: tuple[str, ...]) -> bool: """`git log`: 위치 인자 무제한(revision/pathspec, §6.1b 표) — arity가 쓰기로 뒤집히지 않으므로 상한이 불필요하다. 출력 증폭은 sanitize 240줄 상한(`sanitize_output.py:295`)으로 이미 유계다. 개조 전에는 `-p` 없이 `git log`/`git log --oneline`이 거부됐다(§0 정정 1) — 이 요구를 제거한 것이 이 함수의 핵심 완화다.""" index = 0 while index < len(arguments): argument = arguments[index] if argument == "--": return True if argument in _GIT_LOG_BOOLEAN_FLAGS: index += 1 continue if argument in {"-n", "--max-count", "-U", "--unified"}: if index + 1 >= len(arguments) or not _valid_n(arguments[index + 1]): return False index += 2 continue if argument in _GIT_LOG_VALUE_FLAGS: if index + 1 >= len(arguments): return False index += 2 continue if _git_log_attached_value_ok(argument): index += 1 continue if argument.startswith("-"): return False index += 1 return True def _git_is_safe(argv: tuple[str, ...]) -> bool: """git (서브커맨드, 인자 형태) 쌍 화이트리스트(D1, plan §6.1b, 12행). R-5 불변식(표 전체를 지탱하는 단일 지점) — `argv[1]`을 리터럴로만 서브커맨드로 인정한다. `-`로 시작하면 무조건 거부하고, 서브커맨드를 찾기 위해 선행 전역 옵션(`-c`/`-C`/`-p`/`--paginate`/`--no-pager`/ `--exec-path`/`--git-dir` 등)을 절대 건너뛰지 않는다. **경고**: `_package_script_route:1436`의 `while index < len(argv) and argv[index].startswith("-")` 패턴을 이 함수에 재사용하지 말 것 — 그 패턴을 쓰면 `git -c alias.zz='!echo pwned' zz` 가 임의 셸을 실행한다(3라운드 레드팀 실증, plan §4 시나리오 1). 현재 9개 전역 옵션 우회(AC-1b.2)가 전부 막히는 이유는 오직 이 리터럴 비교 하나다. R-1 불변식 — 서브커맨드 이름만으로도, "위치 인자 0개면 거부"만으로도 승인하지 않는다. 전자는 쓰기 6/6 누수, 후자는 0-arity 쓰기 8건 누수를 실증했다(`git stash`/`gc`/`prune`/`repack`/`clean -fd`/`reset --hard`/ `commit --amend --no-edit`/`branch --edit-description`; 뒤 둘은 데이터 손실이다). 반드시 (서브커맨드, 허용 플래그, 위치 인자 상한) 삼중으로 판정한다. 표에 없는 서브커맨드(`config`/`gc`/`prune`/`repack`/ `clean`/`reset`/`commit`/`push`/`pull`/`fetch`/`merge`/`rebase`/ `checkout`/`switch`/`restore` 등)는 아래 분기에 없어 자동으로 폴스루 거부된다 — never-list는 두지 않는다(이미 deny인 폴스루에 목록을 얹으면 "목록에 없으면 안전"이라는 오독만 유발할 뿐 방어를 강화하지 않는다, plan 결정 D1). `config`는 키 없이 값만 출력해 구조적으로 리댁션이 불가능하므로(원칙 6, R-13) 표에서 영구 삭제되었다 — `config`는 FIX-6의 범위 밖이다(FIX-6은 `remote`만 재도입 심사 대상이었다). `remote`는 FIX-6에서 재도입됐다. `git remote -v`가 자격증명이 임베드된 URL(`https://TOKEN@host/...`)을 출력해 구조적으로 위험했던 원인은 `credential_policy.py`의 URL 리댁션 정규식이 `user:pass@` 두 파트를 모두 요구해 콜론 없는 토큰 전용 URL(가장 흔한 PAT 임베딩 형태)을 통과시켰기 때문이다 — 그 정규식 자체의 결함이지, `remote` 행이 원천적으로 리댁션 불가능한 것은 아니었다(`config`와 다른 점). FIX-6이 그 정규식을 `scheme://TOKEN@` 형태까지 커버하도록 넓혔으므로(비밀번호 파트를 선택적으로 만듦) 지금은 안전하다 — `_git_remote_is_safe`가 `-v`/ `--verbose` 조회 형태만 허용하고 `add`/`remove`/`rename`/`set-url` 등 위치 인자가 있는 쓰기 형태는 branch/tag와 동일한 0-arity 규칙으로 거부한다(AC-1.4에 `remote add origin url` deny가 고정돼 있다). """ if len(argv) < 2 or argv[1].startswith("-"): return False subcommand = argv[1] arguments = argv[2:] if subcommand == "status": return _git_status_is_safe(arguments) if subcommand == "log": return _git_log_is_safe(arguments) if subcommand == "branch": return _git_branch_is_safe(arguments) if subcommand == "tag": return _git_tag_is_safe(arguments) if subcommand == "remote": return _git_remote_is_safe(arguments) if subcommand == "rev-parse": return _git_rev_parse_is_safe(arguments) if subcommand == "describe": return _git_describe_is_safe(arguments) if subcommand == "ls-files": return _git_ls_files_is_safe(arguments) if subcommand == "shortlog": return _git_shortlog_is_safe(arguments) if subcommand == "blame": return _git_blame_is_safe(arguments) if subcommand == "stash": return _git_stash_is_safe(arguments) if subcommand == "grep": return _grep_is_safe(("grep", *arguments), allow_files=True) if subcommand in {"diff", "show"}: return _git_diff_show_is_safe(arguments) return False def _package_script_route(argv: tuple[str, ...]) -> str: value_options = {"--prefix", "--workspace", "-w", "--filter", "--cwd", "-C"} long_value_options = {"--prefix", "--workspace", "--filter", "--cwd"} index = 1 while index < len(argv) and argv[index].startswith("-"): option = argv[index] if option in value_options and index + 1 < len(argv): index += 2 continue if any(option.startswith(name + "=") for name in long_value_options): index += 1 continue return "deny" if index >= len(argv): return "noop" command = argv[index] if command in {"test", "build", "lint"}: return ( "trim" if index + 1 == len(argv) or argv[index + 1] == "--" else "deny" ) if command in {"run", "run-script"} and index + 1 < len(argv): script = argv[index + 1] if script == "build" or script == "lint" or script.startswith("test"): return ( "trim" if index + 2 == len(argv) or argv[index + 2] == "--" else "deny" ) return "noop" def _npx_route(argv: tuple[str, ...]) -> str: index = 1 while index < len(argv) and argv[index].startswith("-"): option = argv[index] if option in {"--no-install", "--yes", "-y"}: index += 1 continue if option in {"-p", "--package"} and index + 1 < len(argv): index += 2 continue if option.startswith("--package="): index += 1 continue return "deny" if index >= len(argv): return "noop" delegated_command = argv[index] delegated_basename = command_basename(delegated_command) if delegated_basename != delegated_command: return "deny" if delegated_basename in {"jest", "vitest"}: return "trim" return "noop" def _make_route(argv: tuple[str, ...]) -> str: index = 1 while index < len(argv) and argv[index].startswith("-"): option = argv[index] if option == "-C" and index + 1 < len(argv): index += 2 continue if option.startswith("-C") and len(option) > 2: index += 1 continue if option == "--directory" and index + 1 < len(argv): index += 2 continue if option.startswith("--directory="): index += 1 continue if option in {"-s", "--silent", "--no-print-directory"}: index += 1 continue return "deny" if index < len(argv) and argv[index] in {"test", "build", "lint"}: return "trim" return "noop" def _is_explicit_noop_command(argv: tuple[str, ...]) -> bool: """Match only the pre-existing short-command controls kept by S011. Tool basenames are deliberately insufficient: e.g. `kubectl get secrets` and `docker run` still reach the fail-closed fallback. The one variable shape is a read-only pod description with a static Kubernetes-style name. """ if argv in MINISHELL_EXPLICIT_NOOP_ARGV: return True return ( len(argv) == 4 and argv[:3] == ("kubectl", "describe", "pod") and re.fullmatch( r"[a-z0-9](?:[a-z0-9.-]{0,251}[a-z0-9])?", argv[3], re.ASCII, ) is not None ) def _wclass_advisory_is_safe(argv: tuple[str, ...]) -> bool: # 토큰 형태(known flag/value pairing)만 검증한다 — `--repo`/`--task-file` # 필수 여부나 `--vendor`의 허용값 같은 CLI 자체의 필수 옵션·enum 검증은 # 의도적으로 위임한다(다운스트림 argparse가 이미 거부함). 이 predicate가 # 막는 건 "인식 못 하는 트레일링 토큰이 조용히 통과하는 것"이지 CLI # 문법 전체가 아니다. if len(argv) < 2: return False if argv[1] == "review": return len(argv) == 2 if argv[1] != "run": return False value_flags = {"--repo", "--task-file", "--vendor", "--workflow"} confirm_seen = False index = 2 while index < len(argv): token = argv[index] if token == "--confirm-task-egress": confirm_seen = True index += 1 continue if "=" in token: flag, value = token.split("=", 1) if flag not in value_flags or not value or value.startswith("-"): return False index += 1 continue if token not in value_flags or index + 1 >= len(argv): return False value = argv[index + 1] if not value or value.startswith("-"): return False index += 2 return confirm_seen # 심사된 정확 이름 확장 레지스트리 — 글롭/접두사 금지(R-12의 TERM*→TERMINFO # 실패 재현 방지). 각 값은 서브커맨드 토큰 하나만이 아니라 argv 전체 모양을 # 검증하는 predicate여야 한다. CGW_EXACT_NAME_EXTENSIONS = { "wclass-advisory": _wclass_advisory_is_safe, } def command_search_diff( argv: tuple[str, ...], *, role: str = "standalone", ) -> str: """Classify one boundary-checked simple command for the A1 route table. FIX-2: standalone `cat`도 `trim`으로 라우팅한다(과거에는 `noop`, 즉 무변형 통과였다). 48KB 초과 파일을 `cat `로 그대로 읽으면 Read 가드 (`guard_large_read.py`)가 `tool_name == "Read"`에서만 발동하므로 이 구멍을 그대로 우회했다 — standalone `cat`이 first/filter 역할과 동일하게 항상 `trim`을 받도록 통일해 막는다. `_cat_is_safe`의 안전성 판정 자체(허용 플래그, `allow_files`)는 바뀌지 않는다. """ if not argv: return "deny" first = command_basename(argv[0]) if _forbidden_command_basename(argv): return "deny" if first == "printf": if role == "filter" or not _printf_is_safe(argv): return "deny" return "trim" if role == "first" else ("noop" if role == "standalone" else "deny") if first == "ls": # standalone 은 오늘의 동작(`noop`)을 그대로 보존한다. `_ls_is_safe` 는 # producer(role == "first") 재승인의 게이트일 뿐이며, standalone 판정에 # 개입해서는 안 된다 — 개입하면 `ls -G`, `ls -x`, `ls --color=always` # 처럼 지금 통과하는 standalone 형태가 새로 거부되어 "이미 동작하는 것을 # 움직이지 않는다"는 설계 불변식을 깨뜨린다. if role == "standalone": return "noop" if role == "filter" or not _ls_is_safe(argv): return "deny" return "trim" if first == "cat": if not _cat_is_safe(argv, allow_files=role != "filter"): return "deny" return "trim" if first == "cut": if role == "first" or not _cut_is_safe(argv): return "deny" return "trim" if role == "filter" else "noop" if first == "sed": # design route-readmission-design-20260729.md §2.3 라우트 배선 — # 기존 filter/standalone 판정을 전혀 움직이지 않는다(둘 다 파일 # 피연산자가 없는 stdin 형태만 오늘 존재했으므로 files == 0 으로 # 수렴). role == "first" 만 새로 열린다 — 단, 파일 피연산자가 있을 # 때만이다. 파일 없는 producer sed 는 훅이 물려준 stdin 을 읽어 # 600초 워치독까지 블록한다(`_git_shortlog_is_safe` 와 동일한 # non-termination 불변식). safe, files = _sed_route_shape(argv) if not safe: return "deny" if role == "filter": return "deny" if files else "trim" if role == "first": return "trim" if files else "deny" return "trim" if files else "noop" if first == "sort": if role == "first" or not _sort_is_safe(argv): return "deny" return "trim" if role == "filter" else "noop" if first == "uniq": if role == "first" or not _uniq_is_safe(argv): return "deny" return "trim" if role == "filter" else "noop" if first == "wc": if role == "first" or not _wc_is_safe(argv, allow_files=role != "filter"): return "deny" return "trim" if role == "filter" else "noop" if first in {"head", "tail"}: return ( "trim" if _head_tail_is_safe(argv, allow_files=role != "filter") else "deny" ) if first in {"grep", "egrep", "fgrep"}: return ( "sanitize" if _grep_is_safe( argv, allow_files=role != "filter", allow_include=first == "grep" and role != "filter", ) else "deny" ) if first == "rg": return ( "sanitize" if role != "filter" and _rg_is_safe(argv) else "deny" ) if first == "git": return ( "sanitize" if role != "filter" and _git_is_safe(argv) else "deny" ) if first == "echo" or _is_explicit_noop_command(argv): # `echo` is the explicit side-effect-free noop used by the shell # contract and hook-envelope controls. The exact kubectl/docker rows # are the pre-existing short-command controls. Keep both distinct from # the unregistered-command fallback so closing F-1 does not turn a # broad tool basename into an allowlist. route = "noop" elif _wrapper_invocation(argv) is not None: # A direct ContextGuard helper CLI is not an incoming CGW1/v0 # execution envelope. `classify_incoming_wrapper` already denied the # exact envelope shapes before route classification; preserve the # established direct-CLI compatibility contract here explicitly. route = "noop" elif first in {"npm", "pnpm", "yarn", "bun"}: route = _package_script_route(argv) elif first == "npx": route = _npx_route(argv) elif first == "make": route = _make_route(argv) elif re.fullmatch(r"python(?:\d+(?:\.\d+)?)?", first): route = ( "trim" if len(argv) > 2 and argv[1] == "-m" and argv[2] in {"pytest", "unittest"} else "noop" ) elif first == "go": route = "trim" if len(argv) > 1 and argv[1] == "test" else "noop" elif first == "cargo": route = "trim" if len(argv) > 1 and argv[1] == "test" else "noop" elif first in {"mvn", "mvnw", "gradle", "gradlew"}: index = 1 if index < len(argv) and argv[index] in {"-q", "--quiet"}: index += 1 if index < len(argv) and argv[index] == "test": route = "trim" else: route = "noop" elif first in CGW_EXACT_NAME_EXTENSIONS: route = "noop" if CGW_EXACT_NAME_EXTENSIONS[first](argv) else "deny" elif first in {"pytest", "tox", "jest", "vitest"}: route = "trim" elif first in {"find", "tree", "fd"}: route = "trim" elif is_log_streaming_command(list(argv)): route = "sanitize" else: # F-1: an unregistered executable identity (including execution-prefix # wrappers such as nice/command/xargs/stdbuf/nohup) has no modeled # semantics. It must not inherit standalone `noop` merely because it # contains no pipeline. route = "deny" if role == "standalone": return route if role == "first": return route if route in {"trim", "sanitize"} else "deny" return "deny" def _find_command_is_side_effecting(argv: tuple[str, ...]) -> bool: if not argv or argv[0].rsplit("/", 1)[-1] != "find": return False return any(argument in _FIND_OUTPUT_RISK_ACTIONS for argument in argv[1:]) def _prefix_overrides_path( segment: tuple[MiniShellWord, ...], route_start: int, ) -> bool: return any( word.assignment_index == 4 and word.source_value.startswith("PATH=") for word in segment[:route_start] ) def _forbidden_command_basename(argv: tuple[str, ...]) -> bool: if not argv: return False # Forbidden identities may be recognized through a path because this gate # can only narrow behavior. Unlike positive route predicates, a basename # match here never grants a wrapper or noop route. basename = os.path.basename(argv[0]) if basename in MINISHELL_DENIED_COMMAND_BASENAMES: return True if basename not in MINISHELL_DENIED_SHELL_BASENAMES: return False return any( re.fullmatch(r"-[^-]*c[^-]*", argument) is not None for argument in argv[1:] ) def _reference_route_argv(parsed: MiniShellParse) -> tuple[str, ...] | None: """Recognize only the static standalone command emitted by the digest.""" if ( parsed.heredoc_delimiter is not None or len(parsed.segments) != 1 or len(parsed.segments[0]) not in {3, 5} ): return None words = parsed.segments[0] if any( word.source_value != word.value or not all(word.active) or word.barriers or word.assignment_index is not None for word in words ): return None argv = tuple(word.value for word in words) if ( argv[0] != BASH_REFERENCE_PUBLIC_COMMAND or argv[1] != "reference" or BASH_REFERENCE_HANDLE_RE.fullmatch(argv[2]) is None ): return None if len(argv) == 3: return argv offset = argv[4] if ( argv[3] != "--offset" or len(offset) > 20 or re.fullmatch(r"(?:0|[1-9][0-9]*)", offset, re.ASCII) is None ): return None return argv def classify_command(command: str, *, allow_cgw1: bool = True) -> CommandDecision: """Make a side-effect-free shell-boundary and routing decision.""" parsed = parse_minishell(command) # 파싱 결과보다 먼저 본다. `-exec` 형태는 `{}` 때문에 문법을 통과하지 # 못하므로, 파싱 이후에 검사하면 되돌릴 수 없는 삭제가 그대로 지나간다. if _raw_command_is_side_effecting_find(command): return _side_effecting_find_ask(parsed) if parsed.denial_reason is not None: # 문법을 끝까지 소비하지 못했다는 것은 "이 명령이 위험하다"가 아니라 # "우리가 안전하게 감쌀 만큼 이해하지 못했다"는 뜻이다. 원본을 그대로 # 통과시킨다. return _decline(parsed, parsed.denial_reason) if _reference_route_argv(parsed) is not None: return CommandDecision( action="reference", parsed=parsed, route_code="reference_expand", ) wrapper = classify_incoming_wrapper(parsed) if wrapper is not None: wrapper_status, _wrapper_kind_name, _payload = wrapper # 남은 단 하나의 거부. 사용자 명령에 대한 정책이 아니라 훅이 자기 # 실행 봉투가 재진입/위조되는 것을 막는 자기 보호다. return CommandDecision( action="deny", parsed=parsed, reason=( f"[context-guard] Refused a command carrying ContextGuard's own execution wrapper " f"({wrapper_status}) — recursion guard, not a policy about your command. " + bash_disable_hint() ), reason_code=wrapper_status, ) segment_routes: list[str] = [] for segment_index, segment in enumerate(parsed.segments): segment_argv = tuple(word.value for word in segment) route_start = _routing_start(segment, segment_argv) if route_start == -2: return _decline(parsed, "unsafe_env_name_denied") if route_start < 0: return _decline(parsed, "restricted_env_denied") if route_start < len(segment): command_word = segment[route_start] if ( command_word.source_value in MINISHELL_DENIED_COMMAND_WORDS and all(command_word.active) and not command_word.barriers ): return _decline(parsed, "reserved_word_denied") route_argv = segment_argv[route_start:] if not route_argv: return _decline(parsed, "assignment_only_denied") if _forbidden_command_basename(route_argv): # 네트워크/실행 계열 basename 목록은 실제 공격자를 막지 못하면서 # 정상 작업만 막아 왔다. 통과시키되 감싸지는 않는다. return _decline(parsed, "forbidden_command_denied") if ( command_basename(route_argv[0]) != route_argv[0] and not ( len(parsed.segments) == 1 and _is_expected_direct_wrapper_path(route_argv) ) ): return _decline(parsed, "command_identity_denied") if parsed.heredoc_delimiter is not None and ( len(parsed.segments) != 1 or command_basename(route_argv[0]) not in MINISHELL_HEREDOC_STDIN_CONSUMERS ): return _decline(parsed, "heredoc_consumer_denied") if ( len(parsed.segments) > 1 and _prefix_overrides_path(segment, route_start) ): return _decline(parsed, "route_operand_denied") if _find_command_is_side_effecting(route_argv): # 되돌릴 수 없는 로컬 변경이고 오탐 비용이 거의 없는 유일한 행이다. # 거부는 ContextGuard 의 권한이 아니지만, 마지막 제동을 말없이 # 걷어내는 것도 아니다 — 사람에게 넘긴다. return _side_effecting_find_ask(parsed) role = ( "standalone" if len(parsed.segments) == 1 else ("first" if segment_index == 0 else "filter") ) route = command_search_diff(route_argv, role=role) if route == "deny": # 라우팅 표에 없다는 것은 출력 형태를 모른다는 뜻일 뿐이다. 감싸지 # 않고 통과시킨다 — 미지의 명령을 trim 으로 감싸면 대화형/stdin # 명령이 워치독까지 멈춘다. return _decline(parsed, "route_policy_denied") segment_routes.append(route) if len(parsed.segments) == 1: action = segment_routes[0] route_code = { "noop": "noop", "trim": "rewrite_trim", "sanitize": "rewrite_sanitize", }[action] return CommandDecision(action=action, parsed=parsed, route_code=route_code) route = "sanitize" if "sanitize" in segment_routes else "trim" return CommandDecision( action=route, parsed=parsed, route_code=( "rewrite_sanitize" if route == "sanitize" else "rewrite_trim" ), ) _SHELL_SAFE_WORD_RE = re.compile(r"^[A-Za-z0-9_@%+=:,./-]+$") def shell_quote(value: str) -> str: if not value: return "''" if _SHELL_SAFE_WORD_RE.fullmatch(value): return value return "'" + value.replace("'", "'\"'\"'") + "'" def shell_join(argv: list[str] | tuple[str, ...]) -> str: return " ".join(shell_quote(value) for value in argv) def _render_minishell_word(word: MiniShellWord) -> str: assignment_name = _env_prefix_name(word) if assignment_name is None: return shell_quote(word.value) assignment_value = word.value[len(assignment_name) + 1 :] return f"{assignment_name}={shell_quote(assignment_value)}" def _git_execution_guard_spec(git_argv: tuple[str, ...]) -> tuple[int, tuple[str, ...]]: if len(git_argv) < 2: return (len(git_argv), ()) subcommand = git_argv[1] if subcommand in {"diff", "show", "log"}: return (2, _GIT_DIFF_EXECUTION_FLAGS) if subcommand in {"grep", "blame"}: return (2, _GIT_TEXTCONV_EXECUTION_FLAGS) if len(git_argv) >= 3 and git_argv[:3] == ("git", "stash", "show"): return (3, _GIT_DIFF_EXECUTION_FLAGS) return (2, ()) def _validated_guarded_git_argv(argv: tuple[str, ...]) -> tuple[str, ...] | None: """Accept only the exact guarded form of an independently safe Git command.""" if not argv or command_basename(argv[0]) != "git": return None flag_index, expected_flags = _git_execution_guard_spec(argv) if tuple(argv[flag_index : flag_index + len(expected_flags)]) != expected_flags: return None original_argv = ( argv[:flag_index] + argv[flag_index + len(expected_flags) :] ) if not _git_is_safe(original_argv): return None return argv def _clear_git_command_scope_config(environment: dict[str, str]) -> None: environment.pop("GIT_CONFIG_COUNT", None) environment.pop("GIT_CONFIG_PARAMETERS", None) for name in tuple(environment): if re.fullmatch(r"GIT_CONFIG_(?:KEY|VALUE)_\d+", name): environment.pop(name, None) def _discover_git_filter_config_keys(git_executable: str) -> tuple[str, ...]: discovery_env = os.environ.copy() _clear_git_command_scope_config(discovery_env) discovery_env.update( { "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "core.fsmonitor", "GIT_CONFIG_VALUE_0": "false", } ) result = subprocess.run( [ git_executable, "config", "--null", "--name-only", "--get-regexp", _GIT_FILTER_CONFIG_QUERY, ], env=discovery_env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=_GIT_FILTER_CONFIG_TIMEOUT_SECONDS, check=False, ) if result.returncode not in {0, 1}: raise RuntimeError("git config discovery failed") if len(result.stdout) > _GIT_FILTER_CONFIG_MAX_BYTES: raise RuntimeError("git filter config exceeded the discovery limit") keys: list[str] = [] seen: set[str] = set() for raw_key in result.stdout.split(b"\0"): if not raw_key: continue key = os.fsdecode(raw_key) if not _GIT_FILTER_CONFIG_KEY_RE.fullmatch(key): raise RuntimeError("git config discovery returned an unexpected key") if key in seen: continue seen.add(key) keys.append(key) if len(keys) > _GIT_FILTER_CONFIG_MAX_KEYS: raise RuntimeError("too many git filter config keys") return tuple(keys) def _guarded_git_environment(filter_keys: tuple[str, ...]) -> dict[str, str]: environment = os.environ.copy() _clear_git_command_scope_config(environment) environment.pop("GIT_EXTERNAL_DIFF", None) config_pairs: list[tuple[str, str]] = [("core.fsmonitor", "false")] for key in filter_keys: value = "false" if key.casefold().endswith(".required") else "" config_pairs.append((key, value)) environment["GIT_CONFIG_COUNT"] = str(len(config_pairs)) for index, (key, value) in enumerate(config_pairs): environment[f"GIT_CONFIG_KEY_{index}"] = key environment[f"GIT_CONFIG_VALUE_{index}"] = value return environment def run_guarded_git(argv: tuple[str, ...]) -> int: guarded_argv = _validated_guarded_git_argv(argv) if guarded_argv is None: print("ContextGuard denied an invalid guarded Git invocation.", file=sys.stderr) return 126 try: git_executable = _approved_runtime_executable("git") filter_keys = _discover_git_filter_config_keys(git_executable) environment = _guarded_git_environment(filter_keys) os.execve(git_executable, list(guarded_argv), environment) except (OSError, RuntimeError, subprocess.SubprocessError): print("ContextGuard could not neutralize Git execution configuration.", file=sys.stderr) return 126 raise AssertionError("os.execve returned unexpectedly") def neutralize_git_config_execution(command: str, parsed: MiniShellParse) -> str: """Disable config helpers while retaining the original command for inspection.""" guarded_segments: list[str] = [] changed = False for segment in parsed.segments: segment_argv = tuple(word.value for word in segment) route_start = _routing_start(segment, segment_argv) rendered_words = [_render_minishell_word(word) for word in segment] if ( route_start >= 0 and route_start + 1 < len(segment_argv) and command_basename(segment_argv[route_start]) == "git" ): git_argv = segment_argv[route_start:] relative_flag_index, flags = _git_execution_guard_spec(git_argv) flag_index = route_start + relative_flag_index rendered_words[flag_index:flag_index] = flags rendered_words[route_start : route_start + 1] = ( shell_quote(_approved_python_runtime()), "-I", shell_quote(os.path.realpath(__file__)), _GIT_GUARD_MODE, "--", "git", ) # Existing wrapper consumers inspect the rewritten string for the # admitted source command. Keep it as one quoted, namespaced # assignment; Git ignores the value and the shell cannot execute it. original_command_marker = ( f"{_GIT_ORIGINAL_COMMAND_ENV}={shell_quote(command)}" ) rendered_words[route_start:route_start] = ( original_command_marker, *_GIT_CONFIG_EXECUTION_GUARD, ) changed = True guarded_segments.append(" ".join(rendered_words)) return " | ".join(guarded_segments) if changed else command def build_wrapped_command(wrapper: str, command: str, *, bash_reference_v1: bool = False) -> str: prefix = _isolated_wrapper_prefix(wrapper) wrapped_argv = prefix + ["--max-lines", CGW1_MAX_LINES] if bash_reference_v1: wrapped_argv += ["--digest", "json", BASH_REFERENCE_FLAG] else: # 기본 경로 escrow. 예산을 넘긴 출력만 로컬 artifact 에 무손실로 넣고 모델에는 # digest + handle + 재조회 명령을 준다. 예산 안에 들어온 출력은 trim 쪽 escrow # 통과 규칙 덕분에 예전처럼 원본 그대로 지나간다. wrapped_argv += CGW1_ESCROW_FLAGS wrapped_argv += ["--", *_runtime_shell_argv(), command] return shell_join(wrapped_argv) def build_sanitized_command(wrapper: str, command: str) -> str: prefix = _isolated_wrapper_prefix(wrapper) wrapped_argv = prefix + [ CGW1_SENTINEL, CGW1_COMMAND_SEARCH_DIFF, "--", *_runtime_shell_argv(), command, ] return shell_join(wrapped_argv) def build_updated_input(tool_input: dict[str, object], wrapped: str) -> dict[str, object]: updated_input = copy.deepcopy(tool_input) updated_input["command"] = wrapped return updated_input def print_updated_command(wrapped: str, tool_input: dict[str, object]) -> None: response = { "hookSpecificOutput": { "hookEventName": "PreToolUse", "updatedInput": build_updated_input(tool_input, wrapped), } } print(json.dumps(response, ensure_ascii=False)) def hook_is_disabled() -> bool: # 같은 제품 안에서 두 플래그가 서로 다른 값 규약을 갖지 않도록 FAIL_OPEN 과 # 동일한 집합을 받는다. return os.environ.get(DISABLE_ENV, "").strip().lower() in FAIL_OPEN_VALUES def main() -> int: # git-guard 실행 모드에서는 stdout 이 훅 프로토콜이 아니라 명령 출력이다. # 거기에 `{}` 를 흘리거나 git 의 종료 코드를 0 으로 덮으면 안 되므로 # crash-open 을 적용하지 않고 그대로 죽게 둔다. if _GIT_GUARD_MODE in sys.argv[1:]: return _main() try: return _main() except SystemExit: raise except BaseException: # noqa: BLE001 - crash-open 은 의도된 계약이다 # 훅의 버그가 사용자의 Bash 를 멈추게 해서는 안 된다. 예외를 삼키고 # 개입을 포기한다. 진단과 응답은 서로를 막지 않도록 분리하고, # 마지막 수단은 print 기계 없이 fd 에 직접 쓴다. try: os.write( 2, b"context-guard-rewrite-bash: internal error; leaving the " b"command unchanged\n", ) except BaseException: # noqa: BLE001 - 진단 실패가 응답을 막으면 안 된다 pass try: os.write(1, b"{}\n") except BaseException: # noqa: BLE001 - 여기서 더 할 수 있는 일은 없다 pass return 0 def _main() -> int: if sys.argv[1:3] == [_GIT_GUARD_MODE, "--"]: return run_guarded_git(tuple(sys.argv[3:])) if _GIT_GUARD_MODE in sys.argv[1:]: print("ContextGuard denied a malformed guarded Git invocation.", file=sys.stderr) return 126 if any(arg in {"-h", "--help"} for arg in sys.argv[1:]): print("ContextGuard helper: context-guard-rewrite-bash") return 0 started = _journal_start() session_id: object = None def journal(intervened: bool, detail: str) -> int: """PreToolUse 훅 한 건을 저널에 남긴다. withheld 는 항상 0 이다 — PreToolUse 는 명령이 실행되기 전이라 출력 크기를 알 수 없고, 알지 못하는 값을 추정치로 적으면 저널이 거짓이 된다. """ _journal_record( started=started, intervened=intervened, session_id=session_id, input_bytes=_hook_input_bytes, withheld_bytes=0, detail=detail, ) return 0 if hook_is_disabled(): print_noop() return journal(False, "env off") if _switch_says_off(): print_noop() return journal(False, "switched off") bash_reference_v1 = BASH_REFERENCE_FLAG in sys.argv[1:] try: payload = load_hook_payload() session_id = payload.get("session_id") tool_input = select_tool_input(payload) except HookInputError as exc: decline_invalid_hook_input(exc.reason_code) return journal(False, f"declined {exc.reason_code}") except RecursionError: decline_invalid_hook_input("payload_nesting_too_deep") return journal(False, "declined payload_nesting_too_deep") except OSError: decline_invalid_hook_input("input_read_failed") return journal(False, "declined input_read_failed") command = tool_input["command"] if not isinstance(command, str): # assert 는 입력 검증 수단이 아니다 — `python -O` 에서 사라져 비문자열이 # 파서로 흘러 들어간다. decline_invalid_hook_input("command_not_string") return journal(False, "declined command_not_string") decision = classify_command(command) if decision.action == "deny": deny_self_protection( decision.reason or ("[context-guard] Refused a command carrying ContextGuard's own execution wrapper. " + bash_disable_hint()) ) return journal(True, "deny") if decision.action == "ask": print_ask_response( decision.reason or "ContextGuard asks you to confirm this command." ) return journal(True, "ask") if decision.action == "noop": print_noop() return journal(False, "noop") # 래퍼가 없다는 것은 ContextGuard 설치가 불완전하다는 뜻이지 명령이 # 위험하다는 뜻이 아니다. 예전에는 이 경우에도 실행을 막았다 — 부분 설치 # 하나로 시끄러운 명령이 전부 차단됐다. 이제는 경고만 남기고 통과시킨다. if decision.action == "trim": wrapper = find_wrapper("trim") if wrapper is None: decline_missing_wrapper("context-guard-trim-output", "untrimmed") return journal(False, "wrapper missing") wrapped = build_wrapped_command(wrapper, command, bash_reference_v1=bash_reference_v1) elif decision.action == "sanitize": wrapper = find_wrapper("sanitize") if wrapper is None: decline_missing_wrapper("context-guard-sanitize-output", "unsanitized") return journal(False, "wrapper missing") guarded_command = neutralize_git_config_execution(command, decision.parsed) wrapped = build_sanitized_command(wrapper, guarded_command) elif decision.action == "reference": wrapper = find_wrapper("trim") if wrapper is None: decline_missing_wrapper("context-guard-trim-output", "unexpanded") return journal(False, "wrapper missing") reference_argv = _reference_route_argv(decision.parsed) if reference_argv is None: raise AssertionError("reference route lost its closed grammar") prefix = ["python3", wrapper] if wrapper.endswith(".py") else [wrapper] wrapped = shell_join( [*prefix, "--expand-bash-reference", *reference_argv[2:]] ) else: raise AssertionError(f"unknown command action: {decision.action}") try: print_updated_command(wrapped, tool_input) except RecursionError: decline_invalid_hook_input("payload_copy_too_deep") return journal(False, "declined payload_copy_too_deep") # 래핑은 기본 경로의 일상 동작이지 개입이 아니다. deny/ask 만 intervened 로 센다(doctor 지표). return journal(False, f"wrapped {decision.action}") if __name__ == "__main__": raise SystemExit(main())