"""Deliberately constrained Git operations for the local checkout.""" from __future__ import annotations import hashlib import json import os from pathlib import Path import re import sys from typing import Any from .errors import LabyricornMcpError from .safety import RepositoryGuard PROHIBITED_COMMIT_PARTS = {".git", ".cache", ".lektor", "build", "dist"} SENSITIVE_NAMES = {".env", ".netrc", "id_rsa", "id_ed25519"} class GitOperations: def __init__(self, guard: RepositoryGuard): self.guard = guard scripts = str(guard.root / "scripts") if scripts not in sys.path: sys.path.insert(0, scripts) import git_repo self.module = git_repo self.repository = git_repo.GitRepository(guard.root) def _run(self, *args: str, timeout: int = 30, allow_prompt: bool = False) -> Any: try: result = self.module._run_git( self.guard.root, *args, check=False, timeout=timeout, allow_prompt=allow_prompt ) except Exception as exc: raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc if result.returncode != 0: detail = self.module._error_detail(result) detail = re.sub(r"(https?://)[^/\s@]+@", r"\1", detail) raise LabyricornMcpError("GIT_ERROR", detail) return result def _status_text(self) -> str: return self._run("status", "--porcelain=v1", "--untracked-files=normal").stdout.rstrip() def status(self) -> dict[str, Any]: text = self._status_text() branch = self._run("symbolic-ref", "--quiet", "--short", "HEAD").stdout.strip() commit = self._run("rev-parse", "HEAD").stdout.strip() upstream_result = self.module._run_git( self.guard.root, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}", check=False, ) upstream = upstream_result.stdout.strip() if upstream_result.returncode == 0 else None entries = self._parse_status(text) status_snapshot = json.dumps( { "branch": branch, "commit": commit, "upstream": upstream, "porcelain": text, "tracked_diff_sha256": hashlib.sha256( self._run("diff", "--binary", "HEAD").stdout.encode("utf-8") ).hexdigest(), "untracked_fingerprint": self._untracked_fingerprint(entries), }, sort_keys=True, separators=(",", ":"), ) return { "branch": branch, "commit": commit, "upstream": upstream, "clean": not entries, "changes": entries, "status_sha256": hashlib.sha256(status_snapshot.encode("utf-8")).hexdigest(), "main_push_warning": ( "Pushing main triggers the repository's external site update workflow. " "The MCP performs only a Git push." if branch == "main" else None ), } def _untracked_fingerprint(self, entries: list[dict[str, str]]) -> list[tuple[str, int, int]]: fingerprint: list[tuple[str, int, int]] = [] for entry in entries: if entry["index"] != "?": continue relative = entry["path"].rstrip("/") candidate = self.guard.resolve(relative) if candidate.is_dir() and (candidate / ".git").exists(): stat = candidate.stat() fingerprint.append((entry["path"], stat.st_size, stat.st_mtime_ns)) continue paths = [candidate] if candidate.is_dir(): paths = [] for directory, names, filenames in os.walk(candidate): for name in names: linked = Path(directory) / name if (linked.is_symlink() or getattr(linked, "is_junction", lambda: False)()) and not linked.resolve().is_relative_to(self.guard.root): raise LabyricornMcpError( "PATH_OUTSIDE_REPOSITORY", "An untracked linked path resolves outside the repository." ) names[:] = [name for name in names if name not in {".git", "__pycache__"}] paths.extend(Path(directory) / filename for filename in filenames) if len(paths) > 2_000: break for path in sorted(paths): if (path.is_symlink() or getattr(path, "is_junction", lambda: False)()) and not path.resolve().is_relative_to(self.guard.root): raise LabyricornMcpError( "PATH_OUTSIDE_REPOSITORY", "An untracked linked path resolves outside the repository." ) if not path.is_file(): continue stat = path.stat() fingerprint.append((self.guard.relative(path), stat.st_size, stat.st_mtime_ns)) return fingerprint @staticmethod def _parse_status(text: str) -> list[dict[str, str]]: entries = [] for line in text.splitlines(): if len(line) < 4: continue raw_path = line[3:] if " -> " in raw_path: original, path = raw_path.split(" -> ", 1) entries.append({"index": line[0], "worktree": line[1], "path": path.strip('"'), "original_path": original.strip('"')}) else: entries.append({"index": line[0], "worktree": line[1], "path": raw_path.strip('"')}) return entries def diff(self, *, staged: bool = False, path: str | None = None, max_chars: int = 100_000) -> dict[str, Any]: args = ["diff"] if staged: args.append("--cached") if path: safe = self.guard.relative(self.guard.resolve(path)) args.extend(("--", safe)) output = self._run(*args).stdout truncated = len(output) > max_chars return {"staged": staged, "path": path, "diff": output[:max_chars], "truncated": truncated} def log(self, limit: int = 20) -> dict[str, Any]: if not isinstance(limit, int) or not 1 <= limit <= 100: raise LabyricornMcpError("VALIDATION_FAILED", "Git log limit must be between 1 and 100.") separator = "\x1f" output = self._run( "log", f"-{limit}", f"--format=%H{separator}%h{separator}%aI{separator}%an{separator}%s" ).stdout commits = [] for line in output.splitlines(): parts = line.split(separator, 4) if len(parts) == 5: commits.append(dict(zip(("commit", "short_commit", "date", "author", "subject"), parts))) return {"count": len(commits), "commits": commits} def pull(self, expected_status_hash: str) -> dict[str, Any]: before = self.status() self._expect_status(before, expected_status_hash) if not before["clean"]: raise LabyricornMcpError("WORKTREE_CONFLICT", "A fast-forward pull requires a clean working tree.") try: state = self.repository.refresh() updated = self.repository.pull(state) except Exception as exc: raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc return {"pulled": True, "branch": updated.branch, "commit": self._run("rev-parse", "HEAD").stdout.strip(), "status": self.status()} def commit(self, message: str, paths: list[str], expected_status_hash: str) -> dict[str, Any]: before = self.status() self._expect_status(before, expected_status_hash) message = " ".join(str(message).splitlines()).strip() if not message: raise LabyricornMcpError("VALIDATION_FAILED", "A non-empty commit message is required.") if not isinstance(paths, list) or not paths: raise LabyricornMcpError("VALIDATION_FAILED", "At least one reviewed changed path is required.") changed = {entry["path"] for entry in before["changes"]} untracked_directories = { entry["path"] for entry in before["changes"] if entry["index"] == "?" and entry["path"].endswith("/") } selected: list[str] = [] for value in paths: path = self.guard.relative(self.guard.resolve(str(value), reject_links=True)) parts = set(Path(path).parts) if parts & PROHIBITED_COMMIT_PARTS or Path(path).name.casefold() in SENSITIVE_NAMES: raise LabyricornMcpError("PATH_NOT_ALLOWED", f"The path is not eligible for MCP commits: {path}") if self._inside_nested_repository(path): raise LabyricornMcpError("PATH_NOT_ALLOWED", f"Nested repositories are outside this MCP's commit scope: {path}") candidate = self.guard.root / path if candidate.exists() and not candidate.is_file(): raise LabyricornMcpError("PATH_NOT_ALLOWED", "Commit paths must identify individual files.") summarized = any(path.startswith(directory) for directory in untracked_directories) if path not in changed and not summarized: raise LabyricornMcpError("WORKTREE_CONFLICT", f"The reviewed path is not a current working-tree change: {path}") if path not in selected: selected.append(path) unrelated_staged = [ entry["path"] for entry in before["changes"] if entry["index"] not in {" ", "?"} and entry["path"] not in selected ] if unrelated_staged: raise LabyricornMcpError( "WORKTREE_CONFLICT", "Unrelated staged changes must be reviewed outside this commit.", {"staged_paths": unrelated_staged}, ) self._run("add", "--", *selected) staged = [line for line in self._run("diff", "--cached", "--name-only").stdout.splitlines() if line] if set(staged) != set(selected): raise LabyricornMcpError( "WORKTREE_CONFLICT", "Staging did not produce exactly the reviewed path set; no commit was created.", {"reviewed_paths": selected, "staged_paths": staged}, ) self._run("commit", "-m", message, "--", *selected, timeout=60) commit = self._run("rev-parse", "HEAD").stdout.strip() return {"committed": True, "commit": commit, "message": message, "paths": selected, "status": self.status()} def _inside_nested_repository(self, relative: str) -> bool: current = self.guard.root for part in Path(relative).parts[:-1]: current /= part if current != self.guard.root and (current / ".git").exists(): return True return False def push(self, expected_head: str, acknowledge_main_update: bool = False) -> dict[str, Any]: before = self.status() if before["commit"] != expected_head: raise LabyricornMcpError( "WORKTREE_CONFLICT", "HEAD changed after it was reviewed; inspect Git status again.", {"actual_head": before["commit"]}, ) if before["branch"] == "main" and not acknowledge_main_update: raise LabyricornMcpError( "MAIN_PUSH_ACK_REQUIRED", "Pushing main triggers the repository's external site update workflow. Set acknowledge_main_update=true to confirm the Git push.", ) try: state = self.repository.refresh() if state.branch != before["branch"]: raise LabyricornMcpError("WORKTREE_CONFLICT", "The branch changed while preparing the push.") updated = self.repository.push(state) except LabyricornMcpError: raise except Exception as exc: raise LabyricornMcpError("GIT_ERROR", str(exc)) from exc return { "pushed": True, "branch": updated.branch, "commit": expected_head, "external_update_may_run": updated.branch == "main", "production_contacted_by_mcp": False, } @staticmethod def _expect_status(status: dict[str, Any], expected: str) -> None: if status["status_sha256"] != expected: raise LabyricornMcpError( "WORKTREE_CONFLICT", "The working tree changed after it was reviewed; inspect Git status again.", {"actual_status_sha256": status["status_sha256"]}, )