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 diff --git a/src/mcp_audit/proof_observer.py b/src/mcp_audit/proof_observer.py index 8048546..5bb6362 100644 --- a/src/mcp_audit/proof_observer.py +++ b/src/mcp_audit/proof_observer.py @@ -73,22 +73,47 @@ 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)(?]+>|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"} @@ -491,6 +516,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: @@ -667,41 +693,108 @@ def _validate_staged_input(path: Path, relative: Path) -> None: raise ObservationBlocked( f"repository input appears to contain private key material: {relative.as_posix()}" ) - if _SENSITIVE_VALUE.search(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() 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, 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 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 + 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] + 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()) 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) + if end < 0: + return "", True + suffix = line[end + 1 :].strip() + return line[start + 1 : end], bool(suffix and not re.fullmatch(r"[\]},]+", suffix)) + 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: if isinstance(value, dict): 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): @@ -712,15 +805,134 @@ 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 +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(): + key_text = str(key) + if key_text == "headers" and isinstance(nested, dict): + for header, item in nested.items(): + if not isinstance(item, str): + continue + 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) + 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: + 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: + 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) + 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_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: + referenced_variables.setdefault(reference, relative) + 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 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() + assignments.setdefault(key, []).append((relative, match_value)) + for variable, header_relative in referenced_variables.items(): + 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: 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 05e9f6a..1386d01 100644 --- a/tests/test_proof_before_action.py +++ b/tests/test_proof_before_action.py @@ -2239,6 +2239,281 @@ def test_sensitive_repository_input_is_blocked_before_execution(tmp_path: Path) ) +@pytest.mark.parametrize( + "placeholder", + ["$GH_TOKEN", "${{ secrets.GH_TOKEN }}", "${{ github.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") + ) + + +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_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_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_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( + '{"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" + 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) + + +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( + "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( + 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) + + +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) + + +@pytest.mark.parametrize( + "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( + 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") + (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", + [ + "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"), [