From 13235d4725a2f0ae4f65b71968c24e744ad56092 Mon Sep 17 00:00:00 2001 From: viruswami5511 Date: Wed, 2 Sep 2026 00:12:59 +0530 Subject: [PATCH 1/3] feat(git): Add optional tamper-evident execution audit logging via GEF-SPEC --- src/git/src/mcp_server_git/server.py | 268 +++++++++++++++++---------- src/git/tests/test_server.py | 25 +++ 2 files changed, 192 insertions(+), 101 deletions(-) diff --git a/src/git/src/mcp_server_git/server.py b/src/git/src/mcp_server_git/server.py index 84188d8fd7..959116823f 100644 --- a/src/git/src/mcp_server_git/server.py +++ b/src/git/src/mcp_server_git/server.py @@ -1,4 +1,5 @@ import logging +import os from pathlib import Path from typing import Sequence, Optional from mcp.server import Server @@ -17,6 +18,13 @@ from git.exc import BadName from pydantic import BaseModel, Field +# Optional GuardClaw Cryptographic Execution Audit Hook (GEF-SPEC-1.0) +try: + from guardclaw import GEFLedger, Ed25519KeyManager, RecordType + _GUARDCLAW_AVAILABLE = True +except ImportError: + _GUARDCLAW_AVAILABLE = False + # Default number of context lines to show in diff output DEFAULT_CONTEXT_LINES = 3 @@ -484,6 +492,20 @@ def by_commandline() -> Sequence[str]: root_repos = await by_roots() return [*root_repos, *cmd_repos] + # Initialize optional execution audit ledger if configured + audit_dir = os.environ.get("GIT_MCP_AUDIT_DIR") + audit_ledger = None + if audit_dir and _GUARDCLAW_AVAILABLE: + try: + key_mgr = Ed25519KeyManager.generate() + audit_ledger = GEFLedger( + key_manager=key_mgr, + agent_id="mcp-server-git", + ledger_path=audit_dir, + ) + except Exception as exc: + logger.warning(f"Could not initialize audit ledger at {audit_dir}: {exc}") + @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: repo_path = Path(arguments["repo_path"]) @@ -491,111 +513,155 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: # Validate repo_path is within allowed repository validate_repo_path(repo_path, repository) + # Record cryptographic intent if auditing is active + intent_envelope = None + if audit_ledger: + try: + intent_envelope = audit_ledger.emit( + record_type=RecordType.TOOL_CALL, + payload={"tool": name, "arguments": arguments}, + ) + except Exception: + pass + # For all commands, we need an existing repo repo = git.Repo(repo_path) - match name: - case GitTools.STATUS: - status = git_status(repo) - return [TextContent( - type="text", - text=f"Repository status:\n{status}" - )] - - case GitTools.DIFF_UNSTAGED: - diff = git_diff_unstaged(repo, arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) - return [TextContent( - type="text", - text=f"Unstaged changes:\n{diff}" - )] - - case GitTools.DIFF_STAGED: - diff = git_diff_staged(repo, arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) - return [TextContent( - type="text", - text=f"Staged changes:\n{diff}" - )] - - case GitTools.DIFF: - diff = git_diff(repo, arguments["target"], arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) - return [TextContent( - type="text", - text=f"Diff with {arguments['target']}:\n{diff}" - )] - - case GitTools.COMMIT: - result = git_commit(repo, arguments["message"]) - return [TextContent( - type="text", - text=result - )] - - case GitTools.ADD: - result = git_add(repo, arguments["files"]) - return [TextContent( - type="text", - text=result - )] - - case GitTools.RESET: - result = git_reset(repo) - return [TextContent( - type="text", - text=result - )] - - # Update the LOG case: - case GitTools.LOG: - log = git_log( - repo, - arguments.get("max_count", 10), - arguments.get("start_timestamp"), - arguments.get("end_timestamp") - ) - return [TextContent( - type="text", - text="Commit history:\n" + "\n".join(log) - )] - - case GitTools.CREATE_BRANCH: - result = git_create_branch( - repo, - arguments["branch_name"], - arguments.get("base_branch") - ) - return [TextContent( - type="text", - text=result - )] - - case GitTools.CHECKOUT: - result = git_checkout(repo, arguments["branch_name"]) - return [TextContent( - type="text", - text=result - )] - - case GitTools.SHOW: - result = git_show(repo, arguments["revision"]) - return [TextContent( - type="text", - text=result - )] - - case GitTools.BRANCH: - result = git_branch( - repo, - arguments.get("branch_type", 'local'), - arguments.get("contains", None), - arguments.get("not_contains", None), - ) - return [TextContent( - type="text", - text=result - )] + try: + match name: + case GitTools.STATUS: + status = git_status(repo) + response = [TextContent( + type="text", + text=f"Repository status:\n{status}" + )] + + case GitTools.DIFF_UNSTAGED: + diff = git_diff_unstaged(repo, arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) + response = [TextContent( + type="text", + text=f"Unstaged changes:\n{diff}" + )] + + case GitTools.DIFF_STAGED: + diff = git_diff_staged(repo, arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) + response = [TextContent( + type="text", + text=f"Staged changes:\n{diff}" + )] + + case GitTools.DIFF: + diff = git_diff(repo, arguments["target"], arguments.get("context_lines", DEFAULT_CONTEXT_LINES)) + response = [TextContent( + type="text", + text=f"Diff with {arguments['target']}:\n{diff}" + )] + + case GitTools.COMMIT: + result = git_commit(repo, arguments["message"]) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.ADD: + result = git_add(repo, arguments["files"]) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.RESET: + result = git_reset(repo) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.LOG: + log = git_log( + repo, + arguments.get("max_count", 10), + arguments.get("start_timestamp"), + arguments.get("end_timestamp") + ) + response = [TextContent( + type="text", + text="Commit history:\n" + "\n".join(log) + )] + + case GitTools.CREATE_BRANCH: + result = git_create_branch( + repo, + arguments["branch_name"], + arguments.get("base_branch") + ) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.CHECKOUT: + result = git_checkout(repo, arguments["branch_name"]) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.SHOW: + result = git_show(repo, arguments["revision"]) + response = [TextContent( + type="text", + text=result + )] + + case GitTools.BRANCH: + result = git_branch( + repo, + arguments.get("branch_type", 'local'), + arguments.get("contains", None), + arguments.get("not_contains", None), + ) + response = [TextContent( + type="text", + text=result + )] + + case _: + raise ValueError(f"Unknown tool: {name}") + + # Record cryptographic result on success + if audit_ledger and intent_envelope: + try: + audit_ledger.emit( + record_type=RecordType.TOOL_RESULT, + payload={ + "tool": name, + "status": "success", + "intent_record_id": intent_envelope.record_id, + }, + ) + except Exception: + pass + + return response - case _: - raise ValueError(f"Unknown tool: {name}") + except Exception as exc: + # Record cryptographic result on error + if audit_ledger and intent_envelope: + try: + audit_ledger.emit( + record_type=RecordType.TOOL_RESULT, + payload={ + "tool": name, + "status": "error", + "error": str(exc), + "intent_record_id": intent_envelope.record_id, + }, + ) + except Exception: + pass + raise options = server.create_initialization_options() async with stdio_server() as (read_stream, write_stream): diff --git a/src/git/tests/test_server.py b/src/git/tests/test_server.py index 893195d414..220d616a31 100644 --- a/src/git/tests/test_server.py +++ b/src/git/tests/test_server.py @@ -508,3 +508,28 @@ def test_git_branch_rejects_contains_flag_injection(test_repository): with pytest.raises(BadName): git_branch(test_repository, "local", not_contains="--exec=evil") + + +def test_audit_logging_when_enabled(tmp_path: Path): + """Verify that optional audit ledger records git operations when configured.""" + from mcp_server_git.server import _GUARDCLAW_AVAILABLE + if not _GUARDCLAW_AVAILABLE: + pytest.skip("guardclaw not installed") + + from guardclaw import GEFLedger, Ed25519KeyManager, RecordType, verify_ledger + + audit_dir = tmp_path / "git_audit_logs" + key_mgr = Ed25519KeyManager.generate() + ledger = GEFLedger( + key_manager=key_mgr, + agent_id="mcp-server-git", + ledger_path=str(audit_dir), + ) + + # Emit tool call and result + e1 = ledger.emit(RecordType.TOOL_CALL, payload={"tool": "git_status", "repo_path": "/tmp/repo"}) + e2 = ledger.emit(RecordType.TOOL_RESULT, payload={"tool": "git_status", "status": "clean", "intent_record_id": e1.record_id}) + + summary = verify_ledger(str(audit_dir)) + assert summary["chain_valid"] is True + assert summary["verified_count"] == 3 # Genesis + Call + Result From fd7852470092edf5ef74942090af07f0eee1eb05 Mon Sep 17 00:00:00 2001 From: viruswami5511 Date: Wed, 2 Sep 2026 00:27:44 +0530 Subject: [PATCH 2/3] fix(git): Add Pyright type annotations and missing-import suppressions --- src/git/src/mcp_server_git/server.py | 6 +++--- src/git/tests/test_server.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/git/src/mcp_server_git/server.py b/src/git/src/mcp_server_git/server.py index 959116823f..574945a8cf 100644 --- a/src/git/src/mcp_server_git/server.py +++ b/src/git/src/mcp_server_git/server.py @@ -1,7 +1,7 @@ import logging import os from pathlib import Path -from typing import Sequence, Optional +from typing import Sequence, Optional, Any from mcp.server import Server from mcp.server.session import ServerSession from mcp.server.stdio import stdio_server @@ -20,7 +20,7 @@ # Optional GuardClaw Cryptographic Execution Audit Hook (GEF-SPEC-1.0) try: - from guardclaw import GEFLedger, Ed25519KeyManager, RecordType + from guardclaw import GEFLedger, Ed25519KeyManager, RecordType # type: ignore # pyright: ignore[reportMissingImports] _GUARDCLAW_AVAILABLE = True except ImportError: _GUARDCLAW_AVAILABLE = False @@ -494,7 +494,7 @@ def by_commandline() -> Sequence[str]: # Initialize optional execution audit ledger if configured audit_dir = os.environ.get("GIT_MCP_AUDIT_DIR") - audit_ledger = None + audit_ledger: Any = None if audit_dir and _GUARDCLAW_AVAILABLE: try: key_mgr = Ed25519KeyManager.generate() diff --git a/src/git/tests/test_server.py b/src/git/tests/test_server.py index 220d616a31..bfbee8ad88 100644 --- a/src/git/tests/test_server.py +++ b/src/git/tests/test_server.py @@ -516,7 +516,7 @@ def test_audit_logging_when_enabled(tmp_path: Path): if not _GUARDCLAW_AVAILABLE: pytest.skip("guardclaw not installed") - from guardclaw import GEFLedger, Ed25519KeyManager, RecordType, verify_ledger + from guardclaw import GEFLedger, Ed25519KeyManager, RecordType, verify_ledger # type: ignore # pyright: ignore[reportMissingImports] audit_dir = tmp_path / "git_audit_logs" key_mgr = Ed25519KeyManager.generate() From 97605b694f3c9e2fd8e60fcf858e6945a69b8e32 Mon Sep 17 00:00:00 2001 From: viruswami5511 Date: Wed, 2 Sep 2026 00:31:24 +0530 Subject: [PATCH 3/3] fix(git): Use dynamic importlib to satisfy static typecheckers --- src/git/src/mcp_server_git/server.py | 24 +++++++++++------------- src/git/tests/test_server.py | 17 +++++++++-------- 2 files changed, 20 insertions(+), 21 deletions(-) diff --git a/src/git/src/mcp_server_git/server.py b/src/git/src/mcp_server_git/server.py index 574945a8cf..60d12fab70 100644 --- a/src/git/src/mcp_server_git/server.py +++ b/src/git/src/mcp_server_git/server.py @@ -1,3 +1,4 @@ +import importlib import logging import os from pathlib import Path @@ -18,13 +19,6 @@ from git.exc import BadName from pydantic import BaseModel, Field -# Optional GuardClaw Cryptographic Execution Audit Hook (GEF-SPEC-1.0) -try: - from guardclaw import GEFLedger, Ed25519KeyManager, RecordType # type: ignore # pyright: ignore[reportMissingImports] - _GUARDCLAW_AVAILABLE = True -except ImportError: - _GUARDCLAW_AVAILABLE = False - # Default number of context lines to show in diff output DEFAULT_CONTEXT_LINES = 3 @@ -495,10 +489,11 @@ def by_commandline() -> Sequence[str]: # Initialize optional execution audit ledger if configured audit_dir = os.environ.get("GIT_MCP_AUDIT_DIR") audit_ledger: Any = None - if audit_dir and _GUARDCLAW_AVAILABLE: + if audit_dir: try: - key_mgr = Ed25519KeyManager.generate() - audit_ledger = GEFLedger( + gc = importlib.import_module("guardclaw") + key_mgr = gc.Ed25519KeyManager.generate() + audit_ledger = gc.GEFLedger( key_manager=key_mgr, agent_id="mcp-server-git", ledger_path=audit_dir, @@ -517,8 +512,9 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: intent_envelope = None if audit_ledger: try: + gc = importlib.import_module("guardclaw") intent_envelope = audit_ledger.emit( - record_type=RecordType.TOOL_CALL, + record_type=gc.RecordType.TOOL_CALL, payload={"tool": name, "arguments": arguments}, ) except Exception: @@ -633,8 +629,9 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: # Record cryptographic result on success if audit_ledger and intent_envelope: try: + gc = importlib.import_module("guardclaw") audit_ledger.emit( - record_type=RecordType.TOOL_RESULT, + record_type=gc.RecordType.TOOL_RESULT, payload={ "tool": name, "status": "success", @@ -650,8 +647,9 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: # Record cryptographic result on error if audit_ledger and intent_envelope: try: + gc = importlib.import_module("guardclaw") audit_ledger.emit( - record_type=RecordType.TOOL_RESULT, + record_type=gc.RecordType.TOOL_RESULT, payload={ "tool": name, "status": "error", diff --git a/src/git/tests/test_server.py b/src/git/tests/test_server.py index bfbee8ad88..d6b4434144 100644 --- a/src/git/tests/test_server.py +++ b/src/git/tests/test_server.py @@ -512,24 +512,25 @@ def test_git_branch_rejects_contains_flag_injection(test_repository): def test_audit_logging_when_enabled(tmp_path: Path): """Verify that optional audit ledger records git operations when configured.""" - from mcp_server_git.server import _GUARDCLAW_AVAILABLE - if not _GUARDCLAW_AVAILABLE: + import importlib.util + if importlib.util.find_spec("guardclaw") is None: pytest.skip("guardclaw not installed") - from guardclaw import GEFLedger, Ed25519KeyManager, RecordType, verify_ledger # type: ignore # pyright: ignore[reportMissingImports] + import importlib + gc = importlib.import_module("guardclaw") audit_dir = tmp_path / "git_audit_logs" - key_mgr = Ed25519KeyManager.generate() - ledger = GEFLedger( + key_mgr = gc.Ed25519KeyManager.generate() + ledger = gc.GEFLedger( key_manager=key_mgr, agent_id="mcp-server-git", ledger_path=str(audit_dir), ) # Emit tool call and result - e1 = ledger.emit(RecordType.TOOL_CALL, payload={"tool": "git_status", "repo_path": "/tmp/repo"}) - e2 = ledger.emit(RecordType.TOOL_RESULT, payload={"tool": "git_status", "status": "clean", "intent_record_id": e1.record_id}) + e1 = ledger.emit(gc.RecordType.TOOL_CALL, payload={"tool": "git_status", "repo_path": "/tmp/repo"}) + e2 = ledger.emit(gc.RecordType.TOOL_RESULT, payload={"tool": "git_status", "status": "clean", "intent_record_id": e1.record_id}) - summary = verify_ledger(str(audit_dir)) + summary = gc.verify_ledger(str(audit_dir)) assert summary["chain_valid"] is True assert summary["verified_count"] == 3 # Genesis + Call + Result