From cce42c462a7c9de3158293c1c1d6a6aa93277e96 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 15:47:51 -0700 Subject: [PATCH 01/10] fix: allow placeholder authorization headers in proof staging --- src/mcp_audit/proof_observer.py | 22 +++++++++++++++- tests/test_proof_before_action.py | 42 +++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 1 deletion(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 8048546..30a03e0 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -73,6 +73,14 @@ r"://[^/@\s]+:[^/@\s]+@|" r"\b(?:AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{20,}|xox[baprs]-[A-Za-z0-9-]{10,})\b" ) +_LITERAL_CREDENTIAL_VALUE = re.compile( + r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+|" + r"://[^/@\s]+:[^/@\s]+@|" + r"\b(?:AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{20,}|xox[baprs]-[A-Za-z0-9-]{10,})\b" +) +_SENSITIVE_HEADER_ASSIGNMENT = re.compile( + r"(?i)(?:authorization|proxy-authorization|cookie|set-cookie)\s*[:=]\s*[\"']?([^\"'\r\n]+)" +) _HOME_PATH = re.compile(r"(? None: raise ObservationBlocked( f"repository input appears to contain private key material: {relative.as_posix()}" ) - if _SENSITIVE_VALUE.search(text): + if _contains_literal_credential_material(text): raise ObservationBlocked( f"repository input appears to contain credential material: {relative.as_posix()}" ) @@ -695,6 +703,18 @@ def _validate_staged_input(path: Path, relative: Path) -> None: ) +def _contains_literal_credential_material(text: str) -> bool: + if _LITERAL_CREDENTIAL_VALUE.search(text): + return True + for match in _SENSITIVE_HEADER_ASSIGNMENT.finditer(text): + value = match.group(1).strip().rstrip("\\").strip().strip("\"'") + if value.lower().startswith("bearer "): + value = value[7:].strip() + if not _is_placeholder(value): + return True + return False + + def _json_contains_literal_secret(value: Any) -> bool: if isinstance(value, dict): for key, nested in value.items(): diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 05e9f6a..2cc9c8f 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2239,6 +2239,48 @@ def test_sensitive_repository_input_is_blocked_before_execution(tmp_path: Path) ) +@pytest.mark.parametrize( + "placeholder", + ["$GH_TOKEN", "${{ secrets.GH_TOKEN }}"], +) +def test_placeholder_authorization_header_is_staged(tmp_path: Path, placeholder: str) -> None: + repo = _repo(tmp_path) + workflow = repo / ".github/workflows/publish.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text( + f'run: curl --header "Authorization: Bearer {placeholder}" https://example.test\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + _stage_repository(repo, staged) + assert (staged / ".github/workflows/publish.yml").read_text(encoding="utf-8") == ( + workflow.read_text(encoding="utf-8") + ) + + +@pytest.mark.parametrize( + "literal_header", + [ + "Authorization: Bearer private-value", + 'Authorization: "Basic private-value"', + "Cookie: session=private-value", + ], +) +def test_literal_authorization_and_cookie_headers_are_blocked( + tmp_path: Path, + literal_header: str, +) -> None: + repo = _repo(tmp_path) + workflow = repo / ".github/workflows/publish.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text(f"header: {literal_header}\n", encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="credential material"): + _stage_repository(repo, staged) + + @pytest.mark.parametrize( ("config_name", "secret_config"), [ From fb55bbeaaa6ef4ab32a69f705b2fc7c7a4666338 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 18:08:06 -0700 Subject: [PATCH 02/10] fix: close proof placeholder credential bypasses --- src/mcp_audit/proof_observer.py | 127 ++++++++++++++++++++++++++---- tests/test_connector.py | 2 +- tests/test_proof_before_action.py | 85 +++++++++++++++++++- 3 files changed, 195 insertions(+), 19 deletions(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 30a03e0..72eaf93 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -79,7 +79,8 @@ r"\b(?:AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{20,}|xox[baprs]-[A-Za-z0-9-]{10,})\b" ) _SENSITIVE_HEADER_ASSIGNMENT = re.compile( - r"(?i)(?:authorization|proxy-authorization|cookie|set-cookie)\s*[:=]\s*[\"']?([^\"'\r\n]+)" + r"(?i)(?]+>|changeme|dummy|example|fixture|" r"placeholder|redacted|sample|synthetic|test)$" ) +_LOCAL_ENV_PLACEHOLDER = re.compile(r"^\$\{?([A-Za-z0-9_]+)\}?$") _DATABASE_SUFFIXES = {".db", ".sqlite", ".sqlite3"} _REPO_CONFIG_NAMES = {".mcp.json", "server.json"} _TEXT_CONFIG_SUFFIXES = {".cfg", ".conf", ".ini", ".properties", ".toml", ".yaml", ".yml"} @@ -499,6 +501,7 @@ def _stage_repository(source: Path, destination: Path) -> None: os.close(descriptor) if traversed_directories != expected_directories: raise ObservationBlocked("repository input tree changed or could not be traversed completely") + _validate_staged_placeholder_sources(destination) def _raise_repository_walk_error(error: OSError) -> None: @@ -692,27 +695,64 @@ def _validate_staged_input(path: Path, relative: Path) -> None: if not _SENSITIVE_KEY.search(key): continue normalized_key = key.lower() - normalized_value = match_value.strip().lower() - if normalized_key == "id-token" and normalized_value in {"none", "read", "write"}: - continue - if normalized_key == "persist-credentials" and normalized_value == "false": - continue - if not _is_placeholder(match_value): + if _secret_assignment_is_literal(normalized_key, match_value): raise ObservationBlocked( f"repository text contains a literal credential assignment: {relative.as_posix()}" ) def _contains_literal_credential_material(text: str) -> bool: + literal, _references = _credential_material(text) + return literal + + +def _credential_material(text: str) -> tuple[bool, set[str]]: + references: set[str] = set() if _LITERAL_CREDENTIAL_VALUE.search(text): - return True - for match in _SENSITIVE_HEADER_ASSIGNMENT.finditer(text): - value = match.group(1).strip().rstrip("\\").strip().strip("\"'") - if value.lower().startswith("bearer "): - value = value[7:].strip() - if not _is_placeholder(value): - return True - return False + return True, references + for line in text.splitlines(): + for match in _SENSITIVE_HEADER_ASSIGNMENT.finditer(line): + value, shell_suffix_literal = _sensitive_header_value(line, match) + if shell_suffix_literal: + return True, references + normalized = _normalize_sensitive_header_value(value) + if not _is_placeholder(normalized): + return True, references + if reference := _local_placeholder_reference(normalized): + references.add(reference) + return False, references + + +def _sensitive_header_value(line: str, match: re.Match[str]) -> tuple[str, bool]: + start = match.end() + prefix_quote = ( + line[match.start() - 1] if match.start() > 0 and line[match.start() - 1] in {"'", '"'} else "" + ) + if start < len(line) and line[start] in {"'", '"'}: + value_quote = line[start] + end = line.find(value_quote, start + 1) + if end < 0: + return "", True + return line[start + 1 : end], False + if prefix_quote: + end = line.find(prefix_quote, start) + if end < 0: + return "", True + suffix = line[end + 1 :] + return line[start:end], bool(suffix and suffix[0] not in {" ", "\t"}) + return line[start:], False + + +def _normalize_sensitive_header_value(value: str) -> str: + normalized = value.strip().rstrip("\\").strip().strip("\"'") + if normalized.lower().startswith("bearer "): + return normalized[7:].strip() + return normalized + + +def _local_placeholder_reference(value: str) -> str | None: + match = _LOCAL_ENV_PLACEHOLDER.fullmatch(value.strip()) + return match.group(1) if match else None def _json_contains_literal_secret(value: Any) -> bool: @@ -720,8 +760,13 @@ def _json_contains_literal_secret(value: Any) -> bool: for key, nested in value.items(): key_text = str(key) if key_text in {"env", "headers"} and isinstance(nested, dict): + if key_text == "headers": + if any(_literal_header_config_value(header, item) for header, item in nested.items()): + return True + continue if any(_literal_secret_value(item) for item in nested.values()): return True + continue if _SENSITIVE_KEY.search(key_text) and _literal_secret_value(nested): return True if key_text == "args" and isinstance(nested, list): @@ -735,12 +780,60 @@ def _json_contains_literal_secret(value: Any) -> bool: return False +def _literal_header_config_value(header: Any, value: Any) -> bool: + if not isinstance(value, str): + return value is not None + header_text = str(header) + if _SENSITIVE_HEADER_ASSIGNMENT.fullmatch(f"{header_text}: "): + return not _is_placeholder(_normalize_sensitive_header_value(value)) + return _literal_secret_value(value) + + def _literal_secret_value(value: Any) -> bool: if isinstance(value, str): return bool(value.strip()) and not _is_placeholder(value) return value is not None +def _secret_assignment_is_literal(normalized_key: str, value: str) -> bool: + normalized_value = value.strip().lower() + if normalized_key == "id-token" and normalized_value in {"none", "read", "write"}: + return False + if normalized_key == "persist-credentials" and normalized_value == "false": + return False + return not _is_placeholder(value) + + +def _validate_staged_placeholder_sources(root: Path) -> None: + referenced_variables: dict[str, Path] = {} + assignments: dict[str, list[tuple[Path, str]]] = {} + for path in sorted(root.rglob("*")): + if not path.is_file() or path.suffix.lower() in _DATABASE_SUFFIXES: + continue + try: + raw_value = path.read_bytes() + if b"\0" in raw_value: + continue + text = raw_value.decode("utf-8") + except UnicodeDecodeError: + continue + relative = path.relative_to(root) + _literal, references = _credential_material(text) + for reference in references: + referenced_variables.setdefault(reference, relative) + for match in _TEXT_SECRET_ASSIGNMENT.finditer(text): + key, match_value = match.groups() + if _SENSITIVE_KEY.search(key): + assignments.setdefault(key, []).append((relative, match_value)) + for variable, header_relative in referenced_variables.items(): + for assignment_relative, assignment_value in assignments.get(variable, []): + if _secret_assignment_is_literal(variable.lower(), assignment_value): + raise ObservationBlocked( + "repository text defines a literal credential used by a staged placeholder: " + f"{assignment_relative.as_posix()} -> {header_relative.as_posix()}" + ) + + def _is_placeholder(value: str) -> bool: normalized = value.strip().strip("\"'") return bool(_PLACEHOLDER_VALUE.fullmatch(normalized)) diff --git a/tests/test_connector.py b/tests/test_connector.py index dcab269..754b157 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -359,7 +359,7 @@ def handle_stop(_signum: int, _frame: object) -> None: command=sys.executable, args=[str(server_script), str(pid_file), str(terminated_file)], ) - connector = ServerConnector(timeout=0.2) + connector = ServerConnector(timeout=1.0) try: audit = await connector.connect(config) diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 2cc9c8f..7667e3e 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2241,7 +2241,7 @@ def test_sensitive_repository_input_is_blocked_before_execution(tmp_path: Path) @pytest.mark.parametrize( "placeholder", - ["$GH_TOKEN", "${{ secrets.GH_TOKEN }}"], + ["$GH_TOKEN", "${{ secrets.GH_TOKEN }}", "${{ github.token }}"], ) def test_placeholder_authorization_header_is_staged(tmp_path: Path, placeholder: str) -> None: repo = _repo(tmp_path) @@ -2259,6 +2259,89 @@ def test_placeholder_authorization_header_is_staged(tmp_path: Path, placeholder: ) +def test_placeholder_authorization_header_with_github_token_assignment_is_staged(tmp_path: Path) -> None: + repo = _repo(tmp_path) + workflow = repo / ".github/workflows/publish.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text( + """env: + GH_TOKEN: ${{ github.token }} +jobs: + publish: + steps: + - run: | + curl --header "Authorization: Bearer $GH_TOKEN" \\ + https://example.test +""", + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + _stage_repository(repo, staged) + assert (staged / ".github/workflows/publish.yml").read_text(encoding="utf-8") == ( + workflow.read_text(encoding="utf-8") + ) + + +def test_placeholder_authorization_json_header_is_staged(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / ".mcp.json").write_text( + '{"mcpServers":{"safe":{"url":"https://example.test",' + '"headers":{"Authorization":"Bearer $GH_TOKEN"}}}}\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + _stage_repository(repo, staged) + assert (staged / ".mcp.json").read_text(encoding="utf-8") == (repo / ".mcp.json").read_text( + encoding="utf-8" + ) + + +def test_placeholder_authorization_header_rejects_adjacent_shell_literal(tmp_path: Path) -> None: + repo = _repo(tmp_path) + workflow = repo / ".github/workflows/publish.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text( + 'run: curl --header "Authorization: Bearer $GH_TOKEN""private-value" https://example.test\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="credential material"): + _stage_repository(repo, staged) + + +@pytest.mark.parametrize( + ("assignment_name", "assignment_text"), + [ + ("same.sh", 'GH_TOKEN="private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n'), + ("same.sh", 'export GH_TOKEN="private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n'), + ], +) +def test_placeholder_authorization_header_rejects_same_file_literal_assignment( + tmp_path: Path, + assignment_name: str, + assignment_text: str, +) -> None: + repo = _repo(tmp_path) + (repo / assignment_name).write_text(assignment_text, encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + +def test_placeholder_authorization_header_rejects_cross_file_literal_assignment(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / "secrets.sh").write_text('GH_TOKEN="private-value"\n', encoding="utf-8") + (repo / "publish.sh").write_text('curl --header "Authorization: Bearer $GH_TOKEN"\n', encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + @pytest.mark.parametrize( "literal_header", [ From c84528d8900ed393e789df2c5a5bc5048bbe302a Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 18:09:45 -0700 Subject: [PATCH 03/10] docs: note proof staging credential repair --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fca0ecb..ac035fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security + +- Tightened Proof Before Action staging so placeholder authorization headers + cannot be combined with adjacent literal material, and local placeholder + variables cannot be backed by staged literal credential assignments. + ## [2.5.0] - Unreleased ### Security From 89da45376fe75c99e536e8f601a3ecb2a111ce08 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:34:27 -0700 Subject: [PATCH 04/10] fix: close placeholder credential alias bypasses --- src/mcp_audit/proof_observer.py | 56 ++++++++++++++++++++++--------- tests/test_proof_before_action.py | 20 +++++++++++ 2 files changed, 60 insertions(+), 16 deletions(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 72eaf93..ecef75c 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -92,6 +92,10 @@ r"password|private[_-]?key|secret|token)[A-Za-z0-9._-]*)\s*[:=]\s*" r"[\"']?([^\"'#\r\n]+)" ) +_TEXT_VARIABLE_ASSIGNMENT = re.compile( + r"(?im)^\s*(?:(?:export|readonly|local|typeset)\s+|declare(?:\s+-[A-Za-z]+)*\s+)?" + r"([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*[\"']?([^\"'#\r\n]+)" +) _SAFE_DATABASE_NAME = re.compile(r"(?i)(?:fixture|sample|seed|synthetic|test)") _PLACEHOLDER_VALUE = re.compile( r"(?i)^(?:\$\{?[A-Z0-9_]+\}?|\$\{\{\s*(?:(?:env|secrets|vars)\.[A-Z0-9_.-]+|github\.token)\s*\}\}|" @@ -728,18 +732,28 @@ def _sensitive_header_value(line: str, match: re.Match[str]) -> tuple[str, bool] prefix_quote = ( line[match.start() - 1] if match.start() > 0 and line[match.start() - 1] in {"'", '"'} else "" ) + if prefix_quote and prefix_quote in line[match.start() : start]: + prefix_quote = "" + if prefix_quote: + outer_end = line.rfind(prefix_quote, start) + if outer_end < start: + return "", True + value = line[start:outer_end].strip() + if value.startswith(("'", '"')): + value_quote = value[0] + inner_end = value.find(value_quote, 1) + if inner_end < 0: + return "", True + suffix = value[inner_end + 1 :] + return value[1:inner_end], bool(suffix.strip()) + return value, False if start < len(line) and line[start] in {"'", '"'}: value_quote = line[start] end = line.find(value_quote, start + 1) if end < 0: return "", True - return line[start + 1 : end], False - if prefix_quote: - end = line.find(prefix_quote, start) - if end < 0: - return "", True - suffix = line[end + 1 :] - return line[start:end], bool(suffix and suffix[0] not in {" ", "\t"}) + suffix = line[end + 1 :].strip() + return line[start + 1 : end], bool(suffix and not re.fullmatch(r"[\]},]+", suffix)) return line[start:], False @@ -821,17 +835,27 @@ def _validate_staged_placeholder_sources(root: Path) -> None: _literal, references = _credential_material(text) for reference in references: referenced_variables.setdefault(reference, relative) - for match in _TEXT_SECRET_ASSIGNMENT.finditer(text): + for match in _TEXT_VARIABLE_ASSIGNMENT.finditer(text): key, match_value = match.groups() - if _SENSITIVE_KEY.search(key): - assignments.setdefault(key, []).append((relative, match_value)) + assignments.setdefault(key, []).append((relative, match_value)) for variable, header_relative in referenced_variables.items(): - for assignment_relative, assignment_value in assignments.get(variable, []): - if _secret_assignment_is_literal(variable.lower(), assignment_value): - raise ObservationBlocked( - "repository text defines a literal credential used by a staged placeholder: " - f"{assignment_relative.as_posix()} -> {header_relative.as_posix()}" - ) + pending = [variable] + visited: set[str] = set() + while pending: + current = pending.pop() + if current in visited: + continue + visited.add(current) + for assignment_relative, assignment_value in assignments.get(current, []): + normalized_value = assignment_value.strip() + if alias := _local_placeholder_reference(normalized_value): + pending.append(alias) + continue + if _secret_assignment_is_literal(current.lower(), normalized_value): + raise ObservationBlocked( + "repository text defines a literal credential used by a staged placeholder: " + f"{assignment_relative.as_posix()} -> {header_relative.as_posix()}" + ) def _is_placeholder(value: str) -> bool: diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 7667e3e..4bf63f1 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2312,11 +2312,31 @@ def test_placeholder_authorization_header_rejects_adjacent_shell_literal(tmp_pat _stage_repository(repo, staged) +def test_placeholder_authorization_header_rejects_nested_quote_literal(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / "publish.sh").write_text( + "curl -H 'Authorization: \"Bearer $GH_TOKEN\"private-value' https://example.test\n", + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="credential material"): + _stage_repository(repo, staged) + + @pytest.mark.parametrize( ("assignment_name", "assignment_text"), [ ("same.sh", 'GH_TOKEN="private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n'), ("same.sh", 'export GH_TOKEN="private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n'), + ("same.sh", 'declare -x GH_TOKEN="private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n'), + ("same.sh", 'VALUE="private-value"\ncurl --header "Authorization: Bearer $VALUE"\n'), + ( + "same.sh", + 'REAL_TOKEN="private-value"\n' + "GH_TOKEN=$REAL_TOKEN\n" + 'curl --header "Authorization: Bearer $GH_TOKEN"\n', + ), ], ) def test_placeholder_authorization_header_rejects_same_file_literal_assignment( From 60660e317d7288ba60afa2ca22d859492bf3fccc Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:37:51 -0700 Subject: [PATCH 05/10] fix: validate full placeholder header syntax --- src/mcp_audit/proof_observer.py | 17 ++++++++------ tests/test_proof_before_action.py | 37 +++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index ecef75c..f8d7bf5 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -682,17 +682,18 @@ def _validate_staged_input(path: Path, relative: Path) -> None: raise ObservationBlocked( f"repository input appears to contain private key material: {relative.as_posix()}" ) - if _contains_literal_credential_material(text): - raise ObservationBlocked( - f"repository input appears to contain credential material: {relative.as_posix()}" - ) + payload = None if path.suffix.lower() == ".json" or path.name in _REPO_CONFIG_NAMES: try: payload = json.loads(text) except json.JSONDecodeError: - payload = None + pass if payload is not None and _json_contains_literal_secret(payload): raise ObservationBlocked(f"repository JSON contains a literal credential: {relative.as_posix()}") + if payload is None and _contains_literal_credential_material(text): + raise ObservationBlocked( + f"repository input appears to contain credential material: {relative.as_posix()}" + ) if path.suffix.lower() in _TEXT_CONFIG_SUFFIXES: for match in _TEXT_SECRET_ASSIGNMENT.finditer(text): key, match_value = match.groups() @@ -738,6 +739,8 @@ def _sensitive_header_value(line: str, match: re.Match[str]) -> tuple[str, bool] outer_end = line.rfind(prefix_quote, start) if outer_end < start: return "", True + outer_suffix = line[outer_end + 1 :] + outer_suffix_literal = bool(outer_suffix and outer_suffix[0] not in {" ", "\t"}) value = line[start:outer_end].strip() if value.startswith(("'", '"')): value_quote = value[0] @@ -745,8 +748,8 @@ def _sensitive_header_value(line: str, match: re.Match[str]) -> tuple[str, bool] if inner_end < 0: return "", True suffix = value[inner_end + 1 :] - return value[1:inner_end], bool(suffix.strip()) - return value, False + return value[1:inner_end], bool(suffix.strip()) or outer_suffix_literal + return value, outer_suffix_literal if start < len(line) and line[start] in {"'", '"'}: value_quote = line[start] end = line.find(value_quote, start + 1) diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 4bf63f1..159858c 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2298,6 +2298,21 @@ def test_placeholder_authorization_json_header_is_staged(tmp_path: Path) -> None ) +def test_multiple_placeholder_json_headers_are_staged(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / ".mcp.json").write_text( + '{"mcpServers":{"safe":{"headers":{"Authorization":"Bearer $GH_TOKEN",' + '"X-Trace":"$TRACE"},"url":"https://example.test"}}}\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + _stage_repository(repo, staged) + assert (staged / ".mcp.json").read_text(encoding="utf-8") == (repo / ".mcp.json").read_text( + encoding="utf-8" + ) + + def test_placeholder_authorization_header_rejects_adjacent_shell_literal(tmp_path: Path) -> None: repo = _repo(tmp_path) workflow = repo / ".github/workflows/publish.yml" @@ -2312,6 +2327,18 @@ def test_placeholder_authorization_header_rejects_adjacent_shell_literal(tmp_pat _stage_repository(repo, staged) +def test_placeholder_authorization_header_rejects_outer_quote_literal_suffix(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / "publish.sh").write_text( + 'curl -H "Authorization: Bearer $GH_TOKEN"private-value https://example.test\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="credential material"): + _stage_repository(repo, staged) + + def test_placeholder_authorization_header_rejects_nested_quote_literal(tmp_path: Path) -> None: repo = _repo(tmp_path) (repo / "publish.sh").write_text( @@ -2362,6 +2389,16 @@ def test_placeholder_authorization_header_rejects_cross_file_literal_assignment( _stage_repository(repo, staged) +def test_placeholder_authorization_alias_cycle_is_staged(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / "environment.sh").write_text("GH_TOKEN=$REAL_TOKEN\nREAL_TOKEN=$GH_TOKEN\n", encoding="utf-8") + (repo / "publish.sh").write_text('curl --header "Authorization: Bearer $GH_TOKEN"\n', encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + _stage_repository(repo, staged) + assert (staged / "publish.sh").is_file() + + @pytest.mark.parametrize( "literal_header", [ From 9043625ee275ec821ab1907339a2f8f92e57f5c7 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:40:27 -0700 Subject: [PATCH 06/10] fix: scan credential material in JSON strings --- src/mcp_audit/proof_observer.py | 2 ++ tests/test_proof_before_action.py | 12 ++++++++++++ 2 files changed, 14 insertions(+) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index f8d7bf5..f6823ca 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -794,6 +794,8 @@ def _json_contains_literal_secret(value: Any) -> bool: return True elif isinstance(value, list): return any(_json_contains_literal_secret(item) for item in value) + elif isinstance(value, str): + return _contains_literal_credential_material(value) return False diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 159858c..a0dd0df 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2313,6 +2313,18 @@ def test_multiple_placeholder_json_headers_are_staged(tmp_path: Path) -> None: ) +def test_literal_credential_in_json_command_is_blocked(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / "package.json").write_text( + '{"scripts":{"publish":"curl -H Authorization: Bearer private-value"}}\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential"): + _stage_repository(repo, staged) + + def test_placeholder_authorization_header_rejects_adjacent_shell_literal(tmp_path: Path) -> None: repo = _repo(tmp_path) workflow = repo / ".github/workflows/publish.yml" From 25a77522416319e515594512c7d57126d22f636c Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:44:34 -0700 Subject: [PATCH 07/10] fix: retain JSON placeholder provenance --- src/mcp_audit/proof_observer.py | 31 ++++++++++++++++++++++++++++++- tests/test_proof_before_action.py | 16 ++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index f6823ca..4155981 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -799,6 +799,26 @@ def _json_contains_literal_secret(value: Any) -> bool: return False +def _json_local_placeholder_references(value: Any) -> set[str]: + references: set[str] = set() + if isinstance(value, dict): + for key, nested in value.items(): + if str(key) == "headers" and isinstance(nested, dict): + for header, item in nested.items(): + if not isinstance(item, str): + continue + if not _SENSITIVE_HEADER_ASSIGNMENT.fullmatch(f"{header}: "): + continue + normalized = _normalize_sensitive_header_value(item) + if reference := _local_placeholder_reference(normalized): + references.add(reference) + references.update(_json_local_placeholder_references(nested)) + elif isinstance(value, list): + for item in value: + references.update(_json_local_placeholder_references(item)) + return references + + def _literal_header_config_value(header: Any, value: Any) -> bool: if not isinstance(value, str): return value is not None @@ -837,7 +857,16 @@ def _validate_staged_placeholder_sources(root: Path) -> None: except UnicodeDecodeError: continue relative = path.relative_to(root) - _literal, references = _credential_material(text) + payload = None + if path.suffix.lower() == ".json" or path.name in _REPO_CONFIG_NAMES: + try: + payload = json.loads(text) + except json.JSONDecodeError: + pass + if payload is not None: + references = _json_local_placeholder_references(payload) + else: + _literal, references = _credential_material(text) for reference in references: referenced_variables.setdefault(reference, relative) for match in _TEXT_VARIABLE_ASSIGNMENT.finditer(text): diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index a0dd0df..5b02ce0 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2313,6 +2313,22 @@ def test_multiple_placeholder_json_headers_are_staged(tmp_path: Path) -> None: ) +def test_multiple_placeholder_json_headers_reject_cross_file_literal_assignment( + tmp_path: Path, +) -> None: + repo = _repo(tmp_path) + (repo / ".mcp.json").write_text( + '{"mcpServers":{"safe":{"headers":{"Authorization":"Bearer $GH_TOKEN",' + '"X-Trace":"$TRACE"},"url":"https://example.test"}}}\n', + encoding="utf-8", + ) + (repo / "secrets.sh").write_text('GH_TOKEN="private-value"\n', encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + def test_literal_credential_in_json_command_is_blocked(tmp_path: Path) -> None: repo = _repo(tmp_path) (repo / "package.json").write_text( From cf2bf519169f32771a150ba65f8f2647438a9bd1 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:49:25 -0700 Subject: [PATCH 08/10] fix: complete placeholder provenance graph --- src/mcp_audit/proof_observer.py | 39 +++++++++++++++++++++------- tests/test_proof_before_action.py | 43 +++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 9 deletions(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 4155981..ef5cd4b 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -96,6 +96,7 @@ r"(?im)^\s*(?:(?:export|readonly|local|typeset)\s+|declare(?:\s+-[A-Za-z]+)*\s+)?" r"([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*[\"']?([^\"'#\r\n]+)" ) +_INLINE_VARIABLE_ASSIGNMENT = re.compile(r"(?m)(?:^|\s)([A-Za-z_][A-Za-z0-9_]*)\s*=\s*[\"']?([^\s\"']+)") _SAFE_DATABASE_NAME = re.compile(r"(?i)(?:fixture|sample|seed|synthetic|test)") _PLACEHOLDER_VALUE = re.compile( r"(?i)^(?:\$\{?[A-Z0-9_]+\}?|\$\{\{\s*(?:(?:env|secrets|vars)\.[A-Z0-9_.-]+|github\.token)\s*\}\}|" @@ -799,24 +800,38 @@ def _json_contains_literal_secret(value: Any) -> bool: return False -def _json_local_placeholder_references(value: Any) -> set[str]: +def _json_placeholder_provenance(value: Any) -> tuple[set[str], dict[str, list[str]]]: references: set[str] = set() + assignments: dict[str, list[str]] = {} if isinstance(value, dict): for key, nested in value.items(): - if str(key) == "headers" and isinstance(nested, dict): + key_text = str(key) + if key_text == "headers" and isinstance(nested, dict): for header, item in nested.items(): if not isinstance(item, str): continue - if not _SENSITIVE_HEADER_ASSIGNMENT.fullmatch(f"{header}: "): - continue - normalized = _normalize_sensitive_header_value(item) + normalized = ( + _normalize_sensitive_header_value(item) + if _SENSITIVE_HEADER_ASSIGNMENT.fullmatch(f"{header}: ") + else item.strip() + ) if reference := _local_placeholder_reference(normalized): references.add(reference) - references.update(_json_local_placeholder_references(nested)) + elif key_text == "env" and isinstance(nested, dict): + for variable, item in nested.items(): + if isinstance(item, str): + assignments.setdefault(str(variable), []).append(item) + nested_references, nested_assignments = _json_placeholder_provenance(nested) + references.update(nested_references) + for variable, values in nested_assignments.items(): + assignments.setdefault(variable, []).extend(values) elif isinstance(value, list): for item in value: - references.update(_json_local_placeholder_references(item)) - return references + nested_references, nested_assignments = _json_placeholder_provenance(item) + references.update(nested_references) + for variable, values in nested_assignments.items(): + assignments.setdefault(variable, []).extend(values) + return references, assignments def _literal_header_config_value(header: Any, value: Any) -> bool: @@ -864,7 +879,10 @@ def _validate_staged_placeholder_sources(root: Path) -> None: except json.JSONDecodeError: pass if payload is not None: - references = _json_local_placeholder_references(payload) + references, json_assignments = _json_placeholder_provenance(payload) + for key, values in json_assignments.items(): + for value in values: + assignments.setdefault(key, []).append((relative, value)) else: _literal, references = _credential_material(text) for reference in references: @@ -872,6 +890,9 @@ def _validate_staged_placeholder_sources(root: Path) -> None: for match in _TEXT_VARIABLE_ASSIGNMENT.finditer(text): key, match_value = match.groups() assignments.setdefault(key, []).append((relative, match_value)) + for match in _INLINE_VARIABLE_ASSIGNMENT.finditer(text): + key, match_value = match.groups() + assignments.setdefault(key, []).append((relative, match_value)) for variable, header_relative in referenced_variables.items(): pending = [variable] visited: set[str] = set() diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index 5b02ce0..ec76a48 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2329,6 +2329,35 @@ def test_multiple_placeholder_json_headers_reject_cross_file_literal_assignment( _stage_repository(repo, staged) +def test_custom_placeholder_json_header_rejects_cross_file_literal_assignment( + tmp_path: Path, +) -> None: + repo = _repo(tmp_path) + (repo / ".mcp.json").write_text( + '{"mcpServers":{"safe":{"headers":{"X-API-Key":"$API_KEY"},"url":"https://example.test"}}}\n', + encoding="utf-8", + ) + (repo / "secrets.sh").write_text('API_KEY="private-value"\n', encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + +def test_json_env_alias_rejects_cross_file_literal_assignment(tmp_path: Path) -> None: + repo = _repo(tmp_path) + (repo / ".mcp.json").write_text( + '{"mcpServers":{"safe":{"env":{"GH_TOKEN":"$REAL_TOKEN"},' + '"headers":{"Authorization":"Bearer $GH_TOKEN"},"url":"https://example.test"}}}\n', + encoding="utf-8", + ) + (repo / "secrets.sh").write_text('REAL_TOKEN="private-value"\n', encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + def test_literal_credential_in_json_command_is_blocked(tmp_path: Path) -> None: repo = _repo(tmp_path) (repo / "package.json").write_text( @@ -2417,6 +2446,20 @@ def test_placeholder_authorization_header_rejects_cross_file_literal_assignment( _stage_repository(repo, staged) +def test_placeholder_authorization_header_rejects_inline_env_literal_assignment( + tmp_path: Path, +) -> None: + repo = _repo(tmp_path) + (repo / "publish.sh").write_text( + 'env GH_TOKEN="private-value" curl --header "Authorization: Bearer $GH_TOKEN"\n', + encoding="utf-8", + ) + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + def test_placeholder_authorization_alias_cycle_is_staged(tmp_path: Path) -> None: repo = _repo(tmp_path) (repo / "environment.sh").write_text("GH_TOKEN=$REAL_TOKEN\nREAL_TOKEN=$GH_TOKEN\n", encoding="utf-8") From 70d54a091502a9164833c63299c4a5b7f3d52ef8 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:53:23 -0700 Subject: [PATCH 09/10] fix: parse alternate placeholder assignments --- src/mcp_audit/proof_observer.py | 14 ++++++++++++++ tests/test_proof_before_action.py | 20 ++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index ef5cd4b..693b0a0 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -97,6 +97,13 @@ r"([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*[\"']?([^\"'#\r\n]+)" ) _INLINE_VARIABLE_ASSIGNMENT = re.compile(r"(?m)(?:^|\s)([A-Za-z_][A-Za-z0-9_]*)\s*=\s*[\"']?([^\s\"']+)") +_QUOTED_VARIABLE_ASSIGNMENT = re.compile( + r"""(?m)(?:^|\s)(?:env\s+|export\s+)?["']([A-Za-z_][A-Za-z0-9_]*)\s*=\s*([^"']+)["']""" +) +_YAML_INLINE_ENV = re.compile(r"(?im)^\s*env\s*:\s*\{([^}\r\n]*)\}") +_YAML_INLINE_ASSIGNMENT = re.compile( + r"""(?:^|,)\s*["']?([A-Za-z_][A-Za-z0-9_]*)["']?\s*:\s*["']?([^,"'}\s]+)""" +) _SAFE_DATABASE_NAME = re.compile(r"(?i)(?:fixture|sample|seed|synthetic|test)") _PLACEHOLDER_VALUE = re.compile( r"(?i)^(?:\$\{?[A-Z0-9_]+\}?|\$\{\{\s*(?:(?:env|secrets|vars)\.[A-Z0-9_.-]+|github\.token)\s*\}\}|" @@ -893,6 +900,13 @@ def _validate_staged_placeholder_sources(root: Path) -> None: for match in _INLINE_VARIABLE_ASSIGNMENT.finditer(text): key, match_value = match.groups() assignments.setdefault(key, []).append((relative, match_value)) + for match in _QUOTED_VARIABLE_ASSIGNMENT.finditer(text): + key, match_value = match.groups() + assignments.setdefault(key, []).append((relative, match_value)) + for env_match in _YAML_INLINE_ENV.finditer(text): + for match in _YAML_INLINE_ASSIGNMENT.finditer(env_match.group(1)): + key, match_value = match.groups() + assignments.setdefault(key, []).append((relative, match_value)) for variable, header_relative in referenced_variables.items(): pending = [variable] visited: set[str] = set() diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index ec76a48..bc1388a 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2460,6 +2460,26 @@ def test_placeholder_authorization_header_rejects_inline_env_literal_assignment( _stage_repository(repo, staged) +@pytest.mark.parametrize( + "assignment", + [ + """env 'GH_TOKEN=private-value' curl --header "Authorization: Bearer $GH_TOKEN"\n""", + """export "GH_TOKEN=private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n""", + """env: { GH_TOKEN: private-value }\nrun: curl --header "Authorization: Bearer $GH_TOKEN"\n""", + ], +) +def test_placeholder_authorization_header_rejects_alternate_literal_assignment_syntax( + tmp_path: Path, + assignment: str, +) -> None: + repo = _repo(tmp_path) + (repo / "publish.yml").write_text(assignment, encoding="utf-8") + staged = tmp_path / "staged" + staged.mkdir() + with pytest.raises(ObservationBlocked, match="literal credential used by a staged placeholder"): + _stage_repository(repo, staged) + + def test_placeholder_authorization_alias_cycle_is_staged(tmp_path: Path) -> None: repo = _repo(tmp_path) (repo / "environment.sh").write_text("GH_TOKEN=$REAL_TOKEN\nREAL_TOKEN=$GH_TOKEN\n", encoding="utf-8") From 65bd13a6f73fee0b9de7c22a3289beb79b5053f1 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Sun, 19 Jul 2026 21:58:07 -0700 Subject: [PATCH 10/10] fix: parse quoted placeholder assignment keys --- src/mcp_audit/proof_observer.py | 8 +++++++- tests/test_proof_before_action.py | 2 ++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 693b0a0..5bb6362 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -94,12 +94,15 @@ ) _TEXT_VARIABLE_ASSIGNMENT = re.compile( r"(?im)^\s*(?:(?:export|readonly|local|typeset)\s+|declare(?:\s+-[A-Za-z]+)*\s+)?" - r"([A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*[\"']?([^\"'#\r\n]+)" + r"[\"']?([A-Za-z_][A-Za-z0-9_]*)[\"']?[ \t]*[:=][ \t]*[\"']?([^\"'#\r\n]+)" ) _INLINE_VARIABLE_ASSIGNMENT = re.compile(r"(?m)(?:^|\s)([A-Za-z_][A-Za-z0-9_]*)\s*=\s*[\"']?([^\s\"']+)") _QUOTED_VARIABLE_ASSIGNMENT = re.compile( r"""(?m)(?:^|\s)(?:env\s+|export\s+)?["']([A-Za-z_][A-Za-z0-9_]*)\s*=\s*([^"']+)["']""" ) +_QUOTED_KEY_VARIABLE_ASSIGNMENT = re.compile( + r"""(?m)(?:^|\s)(?:env\s+|export\s+)?["']([A-Za-z_][A-Za-z0-9_]*)["']\s*=\s*["']?([^\s"']+)""" +) _YAML_INLINE_ENV = re.compile(r"(?im)^\s*env\s*:\s*\{([^}\r\n]*)\}") _YAML_INLINE_ASSIGNMENT = re.compile( r"""(?:^|,)\s*["']?([A-Za-z_][A-Za-z0-9_]*)["']?\s*:\s*["']?([^,"'}\s]+)""" @@ -903,6 +906,9 @@ def _validate_staged_placeholder_sources(root: Path) -> None: for match in _QUOTED_VARIABLE_ASSIGNMENT.finditer(text): key, match_value = match.groups() assignments.setdefault(key, []).append((relative, match_value)) + for match in _QUOTED_KEY_VARIABLE_ASSIGNMENT.finditer(text): + key, match_value = match.groups() + assignments.setdefault(key, []).append((relative, match_value)) for env_match in _YAML_INLINE_ENV.finditer(text): for match in _YAML_INLINE_ASSIGNMENT.finditer(env_match.group(1)): key, match_value = match.groups() diff --git a/tests/test_proof_before_action.py b/tests/test_proof_before_action.py index bc1388a..1386d01 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2464,8 +2464,10 @@ def test_placeholder_authorization_header_rejects_inline_env_literal_assignment( "assignment", [ """env 'GH_TOKEN=private-value' curl --header "Authorization: Bearer $GH_TOKEN"\n""", + """env 'GH_TOKEN'=private-value curl --header "Authorization: Bearer $GH_TOKEN"\n""", """export "GH_TOKEN=private-value"\ncurl --header "Authorization: Bearer $GH_TOKEN"\n""", """env: { GH_TOKEN: private-value }\nrun: curl --header "Authorization: Bearer $GH_TOKEN"\n""", + """env:\n "GH_TOKEN": private-value\nrun: curl --header "Authorization: Bearer $GH_TOKEN"\n""", ], ) def test_placeholder_authorization_header_rejects_alternate_literal_assignment_syntax(