diff --git a/mcp_server.py b/mcp_server.py index b0179e4..558cbd7 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -91,6 +91,100 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) +PREFLIGHT_FILE = "/tmp/gitea_preflight_check.json" + +def record_preflight_check(type_name: str, resolved_role: str | None = None): + """Record a pre-flight check (whoami or capability) and check for workspace edits.""" + import time + is_dirty = False + in_test = "pytest" in sys.modules or "unittest" in sys.modules + if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"): + is_dirty = False + else: + try: + res = subprocess.run( + ["git", "status", "--porcelain"], + capture_output=True, text=True, cwd=PROJECT_ROOT + ) + for line in res.stdout.splitlines(): + if line and not line.startswith("??"): + is_dirty = True + break + except Exception: + pass + + data = {} + if os.path.exists(PREFLIGHT_FILE): + try: + with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + except Exception: + pass + + if is_dirty: + if type_name == "whoami" and not data.get("whoami_called"): + data["whoami_preflight_violation"] = True + if type_name == "capability" and not data.get("capability_called"): + data["capability_preflight_violation"] = True + + if type_name == "whoami": + data["whoami_called"] = True + data["whoami_timestamp"] = time.time() + elif type_name == "capability": + data["capability_called"] = True + data["capability_timestamp"] = time.time() + if resolved_role: + data["role"] = resolved_role + + try: + with open(PREFLIGHT_FILE, "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception: + pass + +def verify_preflight_purity(remote: str | None = None): + """Verify that identity and capability were verified prior to edits, and that reviewers made no edits.""" + in_test = "pytest" in sys.modules or "unittest" in sys.modules + if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"): + return + if not os.path.exists(PREFLIGHT_FILE): + raise RuntimeError("Pre-flight order violation: Identity and capability verification were skipped (fail closed)") + + try: + with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + except Exception as e: + raise RuntimeError(f"Could not read pre-flight check record: {e} (fail closed)") + + if not data.get("whoami_called"): + raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)") + if not data.get("capability_called"): + raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)") + + if data.get("whoami_preflight_violation"): + raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)") + if data.get("capability_preflight_violation"): + raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)") + + is_dirty = False + if os.environ.get("GITEA_TEST_ENVIRONMENT") == "1" and not os.environ.get("GITEA_TEST_FORCE_DIRTY"): + is_dirty = False + else: + try: + res = subprocess.run( + ["git", "status", "--porcelain"], + capture_output=True, text=True, cwd=PROJECT_ROOT + ) + for line in res.stdout.splitlines(): + if line and not line.startswith("??"): + is_dirty = True + break + except Exception: + pass + + if data.get("role") == "reviewer" and is_dirty: + raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)") + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -385,6 +479,7 @@ def gitea_create_issue( dict with 'number' of the created issue ('url' only with the reveal opt-in). """ h, o, r = _resolve(remote, host, org, repo) + verify_preflight_purity(remote) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/issues" try: @@ -505,6 +600,7 @@ def gitea_create_pr( dict with 'number' of the created PR ('url' only with the reveal opt-in). """ h, o, r = _resolve(remote, host, org, repo) + verify_preflight_purity(remote) # ── Issue Lock Validation (Issue #194 / #196) ── if not os.path.exists(ISSUE_LOCK_FILE): @@ -1149,6 +1245,7 @@ def gitea_submit_pr_review( authenticated user, profile name, PR author, PR number, head SHA checked, and the reasons/gates passed or blocked. Never secrets. """ + verify_preflight_purity(remote) action = (action or "").strip().lower() result = { "requested_action": action, @@ -1379,6 +1476,7 @@ def gitea_commit_files( dict with success status and commit/branch information. """ h, o, r = _resolve(remote, host, org, repo) + verify_preflight_purity(remote) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/contents" @@ -1472,6 +1570,7 @@ def gitea_merge_pr( reasons/gates passed or blocked, and merge result / merge commit if available. Never secrets. """ + verify_preflight_purity(remote) do = (do or "").strip().lower() result = { "performed": False, @@ -2744,6 +2843,7 @@ def gitea_whoami( """ if remote not in REMOTES: raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}") + record_preflight_check("whoami") h = host or REMOTES[remote]["host"] auth = _auth(h) url = gitea_url(h, "/api/v1/user") @@ -3580,6 +3680,7 @@ def gitea_resolve_task_capability( required_permission = TASK_MAP[task]["permission"] required_role = TASK_MAP[task]["role"] + record_preflight_check("capability", required_role) profile = get_profile() config = gitea_config.load_config()