205 lines
8.4 KiB
Python
205 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Pre-Push Gatekeeper for Git and MCP Memory.
|
|
Enforces workspace quality gates:
|
|
1. Allows tag pushes and branch deletions without verification.
|
|
2. Bypasses tests for docs/config-only modifications (.md, .txt, .png, justfile, etc.).
|
|
3. Queries mcp-memory gate API (/gate/verify?action=push&target=<branch>).
|
|
4. If no pre-authorized gate is found, runs project tests locally to verify 100% pass before allowing push.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
DOC_EXTENSIONS = {
|
|
".md", ".txt", ".png", ".jpg", ".jpeg", ".pdf", ".svg",
|
|
".drawio", ".gif", ".ico", ".csv"
|
|
}
|
|
|
|
DOC_FILENAMES = {
|
|
"justfile", "jenkinsfile", "dockerfile", "license",
|
|
"changelog", ".gitignore", ".gitattributes"
|
|
}
|
|
|
|
def is_doc_or_config(filepath: str) -> bool:
|
|
base = os.path.basename(filepath).lower()
|
|
if base in DOC_FILENAMES:
|
|
return True
|
|
ext = os.path.splitext(base)[1].lower()
|
|
return ext in DOC_EXTENSIONS
|
|
|
|
def run_git(cmd):
|
|
try:
|
|
res = subprocess.run(["git"] + cmd, capture_output=True, text=True, check=True)
|
|
return res.stdout.strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
def check_mcp_gate(port: str, branch: str):
|
|
"""
|
|
Returns (status_code, reason)
|
|
status_code: 200 (Authorized), 403 (Blocked), 404 (Not found), 0 (Server unreachable)
|
|
"""
|
|
url = f"http://127.0.0.1:{port}/gate/verify?action=push&target={branch}"
|
|
try:
|
|
req = urllib.request.Request(url, method="GET")
|
|
with urllib.request.urlopen(req, timeout=1.5) as resp:
|
|
return resp.status, "Authorized"
|
|
except urllib.error.HTTPError as e:
|
|
body = ""
|
|
try:
|
|
body = e.read().decode("utf-8")
|
|
except Exception:
|
|
pass
|
|
return e.code, body
|
|
except Exception:
|
|
return 0, "Server unreachable"
|
|
|
|
def set_mcp_gate(port: str, branch: str, authorize: bool, reason: str):
|
|
url = f"http://127.0.0.1:{port}/gate/set"
|
|
payload = {
|
|
"action": "push",
|
|
"target": branch,
|
|
"authorize": authorize,
|
|
"reason": reason
|
|
}
|
|
try:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
|
|
with urllib.request.urlopen(req, timeout=1.5):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
def post_mcp_event(port: str, topic: str, payload: dict):
|
|
url = f"http://127.0.0.1:{port}/api/events/post"
|
|
try:
|
|
data = json.dumps({"topic": topic, "payload": payload}).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
|
|
with urllib.request.urlopen(req, timeout=1.0):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
def wait_for_mcp_decision(port: str, topic: str, timeout_secs: int = 2):
|
|
"""Waits for an operator or auditor decision on the event bus."""
|
|
url = f"http://127.0.0.1:{port}/events/wait?topic={topic}&timeout={timeout_secs}"
|
|
try:
|
|
req = urllib.request.Request(url, method="GET")
|
|
with urllib.request.urlopen(req, timeout=timeout_secs + 0.5) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
payload = data.get("payload", {})
|
|
status = str(payload.get("status", "")).lower()
|
|
reason = payload.get("reason", "Decision received via event bus")
|
|
if status in ("authorized", "approved", "ok", "true"):
|
|
return True, reason
|
|
elif status in ("blocked", "denied", "rejected", "false"):
|
|
return False, reason
|
|
except Exception:
|
|
pass
|
|
return None, ""
|
|
|
|
def run_local_tests() -> bool:
|
|
# 1. Rust workspace
|
|
if os.path.exists("Cargo.toml"):
|
|
print("\033[36m[Pre-Push Gatekeeper]\033[0m Running cargo test...")
|
|
cmd = ["cargo", "test", "--workspace"]
|
|
res = subprocess.run(cmd)
|
|
return res.returncode == 0
|
|
# 2. Python workspace (uv / pytest)
|
|
elif os.path.exists("pyproject.toml") or os.path.exists("setup.py"):
|
|
print("\033[36m[Pre-Push Gatekeeper]\033[0m Running python tests...")
|
|
cmd = ["uv", "run", "pytest"] if os.system("uv --version >nul 2>&1") == 0 else ["pytest"]
|
|
res = subprocess.run(cmd)
|
|
return res.returncode == 0
|
|
return True
|
|
|
|
def main():
|
|
# Read stdin passed by git pre-push: <local ref> <local sha> <remote ref> <remote sha>
|
|
lines = sys.stdin.read().splitlines()
|
|
if not lines:
|
|
sys.exit(0)
|
|
|
|
port = os.environ.get("MCP_PORT", "3000")
|
|
for line in lines:
|
|
parts = line.strip().split()
|
|
if len(parts) < 4:
|
|
continue
|
|
local_ref, local_sha, remote_ref, remote_sha = parts[:4]
|
|
|
|
# 1. Skip deletions
|
|
if local_sha == "0000000000000000000000000000000000000000" or local_ref == "(delete)":
|
|
continue
|
|
|
|
# 2. Skip tags
|
|
if local_ref.startswith("refs/tags/"):
|
|
print(f"\033[33m[Pre-Push Gatekeeper]\033[0m Tag push detected ({local_ref}). Quality gate bypassed.")
|
|
continue
|
|
|
|
# Extract branch name
|
|
branch = local_ref.replace("refs/heads/", "")
|
|
|
|
# 3. Check for documentation / config-only changes
|
|
zero_sha = "0000000000000000000000000000000000000000"
|
|
diff_range = local_sha if remote_sha == zero_sha else f"{remote_sha}..{local_sha}"
|
|
changed_files_raw = run_git(["diff", "--name-only", diff_range])
|
|
changed_files = [f.strip() for f in changed_files_raw.splitlines() if f.strip()]
|
|
|
|
if changed_files and all(is_doc_or_config(f) for f in changed_files):
|
|
print(f"\033[33m[Pre-Push Gatekeeper]\033[0m All modified files are docs/config-only. Quality gate bypassed.")
|
|
continue
|
|
|
|
# 4. Query MCP Memory gate status
|
|
status, reason = check_mcp_gate(port, branch)
|
|
if status == 200:
|
|
print(f"\033[32m[Pre-Push Gatekeeper]\033[0m \u2714 Push AUTHORIZED by MCP Memory gate for branch '{branch}'.")
|
|
continue
|
|
elif status == 403:
|
|
print(f"\033[31m[Pre-Push Gatekeeper]\033[0m \u274c Push BLOCKED by MCP Memory gate for branch '{branch}'!")
|
|
if reason:
|
|
print(f" Reason: {reason}")
|
|
sys.exit(1)
|
|
|
|
# 5. No gate found (404 or server offline) -> Request interactive approval or verify unit tests
|
|
if status != 0:
|
|
post_mcp_event(port, "gate:approval_requested", {
|
|
"branch": branch,
|
|
"local_sha": local_sha,
|
|
"remote_sha": remote_sha,
|
|
"changed_files_count": len(changed_files)
|
|
})
|
|
# Fast check: Did dashboard / operator / auditor approve or reject via event bus?
|
|
decision, dec_reason = wait_for_mcp_decision(port, f"gate:decision:{branch}", timeout_secs=1)
|
|
if decision is True:
|
|
print(f"\033[32m[Pre-Push Gatekeeper]\033[0m \u2714 Push INTERACTIVELY AUTHORIZED via event bus: {dec_reason}")
|
|
set_mcp_gate(port, branch, True, dec_reason)
|
|
post_mcp_event(port, "gate:event", {"status": "authorized", "active_gate": "authorized", "branch": branch})
|
|
continue
|
|
elif decision is False:
|
|
print(f"\033[31m[Pre-Push Gatekeeper]\033[0m \u274c Push BLOCKED by operator: {dec_reason}")
|
|
set_mcp_gate(port, branch, False, dec_reason)
|
|
post_mcp_event(port, "gate:event", {"status": "blocked", "active_gate": "blocked", "branch": branch})
|
|
sys.exit(1)
|
|
|
|
print(f"\033[33m[Pre-Push Gatekeeper]\033[0m No pre-authorized gate record for '{branch}'. Verifying unit tests...")
|
|
if run_local_tests():
|
|
print(f"\033[32m[Pre-Push Gatekeeper]\033[0m \u2714 Unit tests passed 100%. Push authorized.")
|
|
if status != 0:
|
|
set_mcp_gate(port, branch, True, "Unit tests passed locally via pre-push gatekeeper")
|
|
post_mcp_event(port, "gate:event", {"status": "authorized", "active_gate": "authorized", "branch": branch})
|
|
else:
|
|
print(f"\033[31m[Pre-Push Gatekeeper]\033[0m \u274c Push BLOCKED! Unit tests failed locally. Fix failing tests before pushing.")
|
|
if status != 0:
|
|
set_mcp_gate(port, branch, False, "Unit tests failed during pre-push gatekeeper check")
|
|
post_mcp_event(port, "gate:event", {"status": "blocked", "active_gate": "blocked", "branch": branch})
|
|
sys.exit(1)
|
|
|
|
sys.exit(0)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|