vrr/wirenboard-agent-vm
2
0
Fork 0
mirror of https://github.com/wirenboard/agent-vm.git synced 2026-07-09 16:00:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Add GitHub MCP proxy with security hardening

Browse source 
Defense-in-depth repo scope enforcement for the GitHub MCP proxy:
- Default-deny for unknown tools (allowlist of known tools)
- Search query scoping: auto-inject repo:OWNER/REPO, block org:/user: qualifiers
- Block non-repo-scoped tools (search_users, search_orgs, get_teams, etc.)
- Inject owner/repo when tools omit them
- Server-side tool filtering via X-MCP-Toolsets/Tools/Readonly/Lockdown headers
- Sensible default toolsets: repos,issues,pull_requests,git,labels
- Lockdown mode enabled by default
- VM cannot override X-MCP-* headers
- Skip body re-serialization when unmodified

Also adds GitHub App token generation scripts and comprehensive tests
(unit + end-to-end with mock upstream server).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Evgeny Boger 2026-02-16 02:33:46 +03:00
parent 76f652f831
commit 967bc51256
5 changed files with 1807 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

402
github-mcp-proxy.py Normal file
View file

@ -0,0 +1,402 @@
#!/usr/bin/env python3
"""
Host-side GitHub MCP proxy for Claude Code VMs.
Accepts unauthenticated MCP requests from the VM, injects the GitHub token
as a Bearer header, enforces repo scope, and forwards to GitHub's hosted
MCP endpoint at api.githubcopilot.com/mcp/.
The VM never sees the GitHub token.
Usage:
GITHUB_MCP_TOKEN=ghu_... GITHUB_MCP_OWNER=wirenboard GITHUB_MCP_REPO=agent-vm \
python3 github-mcp-proxy.py
# Prints the listening port to stdout, then serves until SIGTERM.
Optional env vars for tool filtering (server-side, via GitHub MCP headers):
GITHUB_MCP_TOOLSETS Comma-separated toolsets (default: "repos,issues,pull_requests,git,labels")
GITHUB_MCP_TOOLS Comma-separated tool names (e.g. "get_file_contents,issue_read")
GITHUB_MCP_READONLY Set to "1" for read-only mode
GITHUB_MCP_LOCKDOWN Set to "0" to disable lockdown mode (enabled by default)
"""
import http.client
import http.server
import json
import os
import re
import signal
import ssl
import sys
import socketserver
import time
MCP_HOST = "api.githubcopilot.com"
MCP_PORT = 443
MCP_PATH = "/mcp/"
TOKEN = os.environ.get("GITHUB_MCP_TOKEN", "")
OWNER = os.environ.get("GITHUB_MCP_OWNER", "")
REPO = os.environ.get("GITHUB_MCP_REPO", "")
DEBUG = os.environ.get("GITHUB_MCP_PROXY_DEBUG", "0") == "1"
# Comma-separated toolsets to expose via X-MCP-Toolsets header.
# Default is limited to toolsets that make sense for a single-repo-scoped token.
# Excluded by default: actions, code_security, dependabot, discussions, gists,
# notifications, orgs, projects, secret_protection, security_advisories,
# stargazers, users, copilot, copilot_spaces.
# See https://github.com/github/github-mcp-server/blob/main/docs/remote-server.md
DEFAULT_TOOLSETS = "repos,issues,pull_requests,git,labels"
TOOLSETS = os.environ.get("GITHUB_MCP_TOOLSETS", DEFAULT_TOOLSETS)
# Optional: comma-separated list of individual tool names to expose (e.g.
# "get_file_contents,issue_read"). Sent as X-MCP-Tools header.
# When empty, all tools within the allowed toolsets are available.
TOOLS = os.environ.get("GITHUB_MCP_TOOLS", "")
# Optional: set to "1" to restrict to read-only operations.
READONLY = os.environ.get("GITHUB_MCP_READONLY", "0") == "1"
# Lockdown mode: hides public issue details from users without push access.
# Enabled by default for security.
LOCKDOWN = os.environ.get("GITHUB_MCP_LOCKDOWN", "1") != "0"
# MCP tool argument fields that specify a repo target
REPO_FIELDS = {
"owner": OWNER,
"repo": REPO,
}
# Tools that are safe without repo scoping (read-only user metadata)
UNSCOPED_TOOLS = {"get_me"}
# Search tools that use a "query" parameter instead of owner/repo.
# We inject "repo:OWNER/REPO" to enforce scope.
SEARCH_TOOLS = {
"search_code",
"search_repositories",
"search_issues",
"search_pull_requests",
}
# Search tools that are blocked entirely because repo: scoping doesn't
# apply to their domain (users aren't scoped to a repo).
BLOCKED_SEARCH_TOOLS = {"search_users", "search_orgs"}
# Tools that use org-level fields instead of owner/repo
ORG_TOOLS = {"get_teams", "get_team_members", "list_issue_types"}
# All known tools and their categories. Unknown tools are blocked by default
# to prevent new upstream tools from bypassing scope checks.
KNOWN_TOOLS = (
UNSCOPED_TOOLS
| SEARCH_TOOLS
| BLOCKED_SEARCH_TOOLS
| ORG_TOOLS
# Tools with owner/repo fields (checked by REPO_FIELDS loop)
| {
# repos toolset
"create_branch", "create_or_update_file", "create_repository",
"delete_file", "fork_repository", "get_commit", "get_file_contents",
"get_latest_release", "get_release_by_tag", "get_tag",
"list_branches", "list_commits", "list_releases", "list_tags",
"push_files",
# issues toolset
"add_issue_comment", "assign_copilot_to_issue", "get_label",
"issue_read", "issue_write", "list_issues", "sub_issue_write",
# pull_requests toolset
"add_comment_to_pending_review", "add_reply_to_pull_request_comment",
"create_pull_request", "list_pull_requests", "merge_pull_request",
"pull_request_read", "pull_request_review_write",
"request_copilot_review", "update_pull_request",
"update_pull_request_branch",
# git toolset
"get_repository_tree",
# labels toolset
"label_write", "list_label",
}
)
def debug(msg):
if DEBUG:
print(f"[github-mcp-proxy {time.strftime('%H:%M:%S')}] {msg}", file=sys.stderr)
def _enforce_search_scope(tool, args):
"""For search tools, inject repo scope into the query string.
Returns (modified_args, error_message).
"""
if not OWNER or not REPO:
return args, None
query = args.get("query", "")
scope = f"repo:{OWNER}/{REPO}"
# Reject repo: qualifiers that point to a different repo.
for match in re.finditer(r'\brepo:(\S+)', query):
if match.group(1) != f"{OWNER}/{REPO}":
msg = (f"Repo scope violation: {tool} query contains "
f"repo:{match.group(1)}, expected repo:{OWNER}/{REPO}")
debug(f" BLOCKED: {msg}")
return None, msg
# Reject org: and user: qualifiers — these widen search beyond the
# scoped repo and could leak cross-repo data.
for match in re.finditer(r'\b(org|user):(\S+)', query):
qualifier, value = match.group(1), match.group(2)
msg = (f"Repo scope violation: {tool} query contains "
f"{qualifier}:{value} (not allowed, use repo: scope)")
debug(f" BLOCKED: {msg}")
return None, msg
# If no repo: qualifier present, inject one
if f"repo:{OWNER}/{REPO}" not in query:
args = dict(args)
args["query"] = f"{scope} {query}".strip()
debug(f" injected scope: {args['query']}")
return args, None
def enforce_repo_scope(body_bytes):
"""Parse MCP request body, enforce owner/repo in tool call arguments.
Returns (modified_body_bytes, error_message). error_message is None on success.
"""
if not body_bytes:
return body_bytes, None
try:
req = json.loads(body_bytes)
except (json.JSONDecodeError, UnicodeDecodeError):
return body_bytes, None
method = req.get("method")
if method != "tools/call":
return body_bytes, None
params = req.get("params", {})
tool = params.get("name", "unknown")
args = params.get("arguments", {})
if not isinstance(args, dict):
return body_bytes, None
# Block unknown tools (default-deny). New upstream tools won't
# silently bypass scope checks.
if tool not in KNOWN_TOOLS:
msg = f"Repo scope violation: unknown tool {tool!r} is not allowed"
debug(f" BLOCKED: {msg}")
return None, msg
# Allow safe tools that don't need repo scoping
if tool in UNSCOPED_TOOLS:
debug(f" allowed unscoped tool: {tool}")
return body_bytes, None
# Block org-level tools and user/org search tools
if tool in ORG_TOOLS or tool in BLOCKED_SEARCH_TOOLS:
msg = f"Repo scope violation: {tool} is not allowed (not repo-scoped)"
debug(f" BLOCKED: {msg}")
return None, msg
# For search tools, inject repo scope into the query
if tool in SEARCH_TOOLS:
args, err = _enforce_search_scope(tool, args)
if err:
return None, err
req["params"]["arguments"] = args
return json.dumps(req).encode(), None
# For tools with owner/repo, enforce scoped values and inject if missing
modified = False
for field, enforced_value in REPO_FIELDS.items():
if enforced_value:
if field in args:
if args[field] != enforced_value:
msg = f"Repo scope violation: {tool} called with {field}={args[field]!r}, expected {enforced_value!r}"
debug(f" BLOCKED: {msg}")
return None, msg
else:
args[field] = enforced_value
modified = True
debug(f" injected {field}={enforced_value!r} for {tool}")
if modified:
req["params"]["arguments"] = args
return json.dumps(req).encode(), None
return body_bytes, None
class GitHubMCPProxyHandler(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, format, *args):
pass
def do_request(self):
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length) if content_length else None
debug(f">>> {self.command} {self.path} ({content_length} bytes)")
# Enforce repo scope on tool calls
if body and self.command == "POST":
body, err = enforce_repo_scope(body)
if err:
self.send_error_response(403, err)
return
# Build upstream headers: copy all, replace auth and host
headers = {}
for key, value in self.headers.items():
lower = key.lower()
if lower in ("authorization", "host",
"x-mcp-toolsets", "x-mcp-tools",
"x-mcp-readonly", "x-mcp-lockdown"):
continue
headers[key] = value
headers["Authorization"] = f"Bearer {TOKEN}"
headers["Host"] = MCP_HOST
if body:
headers["Content-Length"] = str(len(body))
# Inject tool-filtering headers (set by host, never from VM)
if TOOLSETS:
headers["X-MCP-Toolsets"] = TOOLSETS
if TOOLS:
headers["X-MCP-Tools"] = TOOLS
if READONLY:
headers["X-MCP-Readonly"] = "true"
if LOCKDOWN:
headers["X-MCP-Lockdown"] = "true"
# Forward to GitHub's MCP endpoint
ctx = ssl.create_default_context()
try:
t0 = time.monotonic()
conn = http.client.HTTPSConnection(MCP_HOST, MCP_PORT, context=ctx, timeout=300)
# Rewrite path to the MCP endpoint
upstream_path = MCP_PATH + self.path.lstrip("/")
# Normalize double slashes
upstream_path = upstream_path.replace("//", "/")
conn.request(self.command, upstream_path, body=body, headers=headers)
upstream = conn.getresponse()
latency_ms = (time.monotonic() - t0) * 1000
except Exception as e:
debug(f" UPSTREAM ERROR: {e}")
self.send_error_response(502, f"Failed to connect to GitHub MCP: {e}")
return
upstream_headers = upstream.getheaders()
is_streaming = any(
k.lower() == "transfer-encoding" for k, v in upstream_headers
)
debug(f"<<< {upstream.status} {'streaming' if is_streaming else 'complete'} ({latency_ms:.0f}ms)")
if is_streaming:
self.send_response(upstream.status)
for key, value in upstream_headers:
lower = key.lower()
if lower in ("transfer-encoding", "content-length",
"connection", "keep-alive"):
continue
self.send_header(key, value)
self.send_header("Transfer-Encoding", "chunked")
self.end_headers()
total_bytes = 0
try:
while True:
data = upstream.read(8192)
if not data:
break
total_bytes += len(data)
self.wfile.write(f"{len(data):x}\r\n".encode())
self.wfile.write(data)
self.wfile.write(b"\r\n")
self.wfile.flush()
self.wfile.write(b"0\r\n\r\n")
self.wfile.flush()
debug(f" streamed {total_bytes} bytes")
except (BrokenPipeError, ConnectionResetError) as e:
debug(f" stream broken: {e} after {total_bytes} bytes")
finally:
conn.close()
else:
body_data = upstream.read()
conn.close()
debug(f" body: {len(body_data)} bytes")
if DEBUG and len(body_data) < 4096:
try:
debug(f" {body_data.decode()}")
except UnicodeDecodeError:
pass
self.send_response(upstream.status)
for key, value in upstream_headers:
lower = key.lower()
if lower in ("transfer-encoding", "content-length",
"connection", "keep-alive"):
continue
self.send_header(key, value)
self.send_header("Content-Length", str(len(body_data)))
self.end_headers()
self.wfile.write(body_data)
self.wfile.flush()
def send_error_response(self, code, message):
body = json.dumps({"error": {"type": "proxy_error", "message": message}}).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
do_GET = do_request
do_POST = do_request
do_PUT = do_request
do_DELETE = do_request
class QuietServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = True
daemon_threads = True
def main():
if not TOKEN:
print("Error: GITHUB_MCP_TOKEN is required", file=sys.stderr)
sys.exit(1)
if not OWNER or not REPO:
print("Error: GITHUB_MCP_OWNER and GITHUB_MCP_REPO are required", file=sys.stderr)
sys.exit(1)
server = QuietServer(("127.0.0.1", 0), GitHubMCPProxyHandler)
port = server.server_address[1]
print(port, flush=True)
def handle_signal(signum, frame):
server.shutdown()
signal.signal(signal.SIGTERM, handle_signal)
signal.signal(signal.SIGINT, handle_signal)
debug(f"Listening on port {port}, forwarding to {MCP_HOST}{MCP_PATH}")
debug(f"Repo scope: {OWNER}/{REPO}")
if TOOLSETS:
debug(f"Toolsets: {TOOLSETS}")
if TOOLS:
debug(f"Tools: {TOOLS}")
if READONLY:
debug(f"Read-only mode: enabled")
if LOCKDOWN:
debug(f"Lockdown mode: enabled")
server.serve_forever()
if __name__ == "__main__":
main()

336
github_app_token.sh Executable file
View file

@ -0,0 +1,336 @@
#!/usr/bin/env bash
#
# GitHub App Scoped Token Generator (pure bash, no Python needed)
#
# Generates repo-scoped GitHub tokens for wirenboard/agent-vm using a GitHub App
# with read/write contents permission.
#
# Dependencies: openssl, curl, jq (optional, for pretty output)
#
# Usage:
# # Print instructions to create the GitHub App
# ./github_app_token.sh create-app
#
# # Generate a scoped token
# ./github_app_token.sh get-token --app-id 123456 --pem-file key.pem
set -euo pipefail
TARGET_OWNER="${TARGET_OWNER:-wirenboard}"
TARGET_REPO="${TARGET_REPO:-agent-vm}"
API="https://api.github.com"
# --- Helpers ---
die() { echo "Error: $*" >&2; exit 1; }
base64url() {
openssl base64 -e -A | tr '+/' '-_' | tr -d '='
}
# --- App creation helper ---
cmd_create_app() {
cat <<'EOF'
============================================================
GitHub App Creation Instructions
============================================================
1. Go to: https://github.com/organizations/wirenboard/settings/apps/new
(Or for personal account: https://github.com/settings/apps/new)
2. Fill in the following settings:
- GitHub App name: WB Agent VM Token Generator
- Homepage URL: https://github.com/wirenboard/agent-vm
- Uncheck 'Active' under Webhook
- Under 'Repository permissions':
- Contents: Read & write
- Under 'Where can this GitHub App be installed?':
- Select 'Only on this account'
3. Click 'Create GitHub App'
4. Note the App ID shown on the next page
5. Scroll down to 'Private keys' and click 'Generate a private key'
Save the downloaded .pem file
6. Install the app:
- Click 'Install App' in the left sidebar
- Choose the wirenboard organization
- Select 'Only select repositories' and pick 'agent-vm'
7. Generate a token:
EOF
echo " $0 get-token --app-id <APP_ID> --pem-file <path/to/key.pem>"
echo
}
# --- JWT generation ---
generate_jwt() {
local app_id="$1"
local pem_file="$2"
local now
now=$(date +%s)
local iat=$((now - 60))
local exp=$((now + 600))
local header
header=$(printf '{"alg":"RS256","typ":"JWT"}' | base64url)
local payload
payload=$(printf '{"iat":%d,"exp":%d,"iss":"%s"}' "$iat" "$exp" "$app_id" | base64url)
local signature
signature=$(printf '%s.%s' "$header" "$payload" \
| openssl dgst -sha256 -sign "$pem_file" -binary \
| base64url)
printf '%s.%s.%s' "$header" "$payload" "$signature"
}
# --- GitHub API ---
github_api() {
local method="$1"
local path="$2"
local token="$3"
local token_type="${4:-Bearer}"
local body="${5:-}"
local curl_args=(
-s -f
-X "$method"
-H "Accept: application/vnd.github+json"
-H "Authorization: ${token_type} ${token}"
-H "X-GitHub-Api-Version: 2022-11-28"
)
if [[ -n "$body" ]]; then
curl_args+=(-H "Content-Type: application/json" -d "$body")
fi
local response http_code
response=$(curl -w '\n%{http_code}' "${curl_args[@]}" "${API}${path}" 2>&1) || {
echo "API request failed: ${method} ${API}${path}" >&2
echo "$response" >&2
exit 1
}
# Extract body (all lines except last) and http_code (last line)
http_code=$(echo "$response" | tail -1)
echo "$response" | sed '$d'
}
# --- Token generation ---
cmd_get_token() {
local app_id=""
local pem_file=""
while [[ $# -gt 0 ]]; do
case "$1" in
--app-id) app_id="$2"; shift 2 ;;
--pem-file) pem_file="$2"; shift 2 ;;
*) die "Unknown option: $1" ;;
esac
done
[[ -n "$app_id" ]] || die "Missing --app-id"
[[ -n "$pem_file" ]] || die "Missing --pem-file"
[[ -f "$pem_file" ]] || die "PEM file not found: $pem_file"
echo "Generating JWT for App ID ${app_id}..."
local jwt
jwt=$(generate_jwt "$app_id" "$pem_file")
echo "Looking up installation..."
local installations
installations=$(github_api GET /app/installations "$jwt")
# Find installation for target owner
local installation_id=""
if command -v jq &>/dev/null; then
installation_id=$(echo "$installations" \
| jq -r --arg owner "$TARGET_OWNER" \
'.[] | select(.account.login == $owner) | .id // empty' \
| head -1)
else
# Fallback: grep for the installation id near the target owner
# This is fragile but works for simple cases
installation_id=$(echo "$installations" \
| grep -B5 "\"login\":.*\"${TARGET_OWNER}\"" \
| grep '"id":' | head -1 \
| sed 's/.*"id": *\([0-9]*\).*/\1/')
fi
[[ -n "$installation_id" ]] || die "No installation found for ${TARGET_OWNER}/${TARGET_REPO}. Make sure the app is installed."
echo "Found installation ID: ${installation_id}"
echo "Requesting scoped token for ${TARGET_OWNER}/${TARGET_REPO} (contents: write)..."
local token_body
token_body=$(printf '{"repositories":["%s"],"permissions":{"contents":"write"}}' "$TARGET_REPO")
local token_resp
token_resp=$(github_api POST "/app/installations/${installation_id}/access_tokens" "$jwt" "Bearer" "$token_body")
local token expires_at
if command -v jq &>/dev/null; then
token=$(echo "$token_resp" | jq -r '.token')
expires_at=$(echo "$token_resp" | jq -r '.expires_at')
else
token=$(echo "$token_resp" | grep '"token":' | head -1 | sed 's/.*"token": *"\([^"]*\)".*/\1/')
expires_at=$(echo "$token_resp" | grep '"expires_at":' | head -1 | sed 's/.*"expires_at": *"\([^"]*\)".*/\1/')
fi
[[ -n "$token" && "$token" != "null" ]] || die "Failed to get token from response: ${token_resp}"
echo "Token expires at: ${expires_at}"
echo
echo "Verifying token..."
local repo_resp
repo_resp=$(github_api GET "/repos/${TARGET_OWNER}/${TARGET_REPO}" "$token" "token")
local full_name
if command -v jq &>/dev/null; then
full_name=$(echo "$repo_resp" | jq -r '.full_name')
else
full_name=$(echo "$repo_resp" | grep '"full_name":' | head -1 | sed 's/.*"full_name": *"\([^"]*\)".*/\1/')
fi
echo "Verified: can access ${full_name}"
echo
echo "============================================================"
echo "GITHUB_TOKEN=${token}"
echo "============================================================"
}
# --- User access token via device flow ---
json_val() {
# Extract a JSON string value by key (simple, no jq fallback)
local json="$1" key="$2"
if command -v jq &>/dev/null; then
echo "$json" | jq -r ".$key // empty"
else
echo "$json" | sed -n 's/.*"'"$key"'": *"\{0,1\}\([^",}]*\)"\{0,1\}.*/\1/p' | head -1
fi
}
cmd_user_token() {
local client_id=""
while [[ $# -gt 0 ]]; do
case "$1" in
--client-id) client_id="$2"; shift 2 ;;
*) die "Unknown option: $1" ;;
esac
done
[[ -n "$client_id" ]] || die "Missing --client-id"
echo "Requesting device code..."
local device_resp
device_resp=$(curl -s -X POST \
-H "Accept: application/json" \
-d "client_id=${client_id}" \
"https://github.com/login/device/code")
local device_code user_code verification_uri expires_in interval
device_code=$(json_val "$device_resp" "device_code")
user_code=$(json_val "$device_resp" "user_code")
verification_uri=$(json_val "$device_resp" "verification_uri")
expires_in=$(json_val "$device_resp" "expires_in")
interval=$(json_val "$device_resp" "interval")
interval=${interval:-5}
[[ -n "$device_code" ]] || die "Failed to get device code: ${device_resp}"
echo
echo "============================================================"
echo " Open: ${verification_uri}"
echo " Enter: ${user_code}"
echo "============================================================"
echo
echo "Waiting for authorization (expires in ${expires_in}s)..."
# Poll for the token
while true; do
sleep "$interval"
local token_resp
token_resp=$(curl -s -X POST \
-H "Accept: application/json" \
-d "client_id=${client_id}&device_code=${device_code}&grant_type=urn:ietf:params:oauth:grant-type:device_code" \
"https://github.com/login/oauth/access_token")
local error
error=$(json_val "$token_resp" "error")
case "$error" in
authorization_pending) continue ;;
slow_down)
interval=$(json_val "$token_resp" "interval")
interval=${interval:-$((interval + 5))}
continue
;;
expired_token) die "Device code expired. Please try again." ;;
access_denied) die "Authorization was denied by the user." ;;
"") ;; # No error — success
*) die "${error}: $(json_val "$token_resp" "error_description")" ;;
esac
# Success
local token expires_in_tok refresh_token
token=$(json_val "$token_resp" "access_token")
expires_in_tok=$(json_val "$token_resp" "expires_in")
refresh_token=$(json_val "$token_resp" "refresh_token")
[[ -n "$token" ]] || die "No access_token in response: ${token_resp}"
echo "Authorization successful!"
echo
[[ -z "$expires_in_tok" ]] || echo "Token expires in: ${expires_in_tok}s"
[[ -z "$refresh_token" ]] || echo "Refresh token: ${refresh_token}"
echo
echo "Verifying token..."
local repo_resp
repo_resp=$(curl -s \
-H "Accept: application/vnd.github+json" \
-H "Authorization: Bearer ${token}" \
"${API}/repos/${TARGET_OWNER}/${TARGET_REPO}")
local full_name
full_name=$(json_val "$repo_resp" "full_name")
echo "Verified: can access ${full_name}"
echo
echo "============================================================"
echo "GITHUB_TOKEN=${token}"
echo "============================================================"
break
done
}
# --- Main ---
case "${1:-}" in
create-app) cmd_create_app ;;
get-token) shift; cmd_get_token "$@" ;;
user-token) shift; cmd_user_token "$@" ;;
*)
echo "Usage: $0 {create-app|get-token|user-token}"
echo
echo "Commands:"
echo " create-app Print instructions to create the GitHub App"
echo " get-token Generate a scoped installation token"
echo " --app-id <ID> GitHub App ID"
echo " --pem-file <path> Path to private key PEM"
echo " user-token Generate a user access token via device flow"
echo " --client-id <ID> GitHub App Client ID"
exit 1
;;
esac

453
github_app_token_demo.py Normal file
View file

@ -0,0 +1,453 @@
#!/usr/bin/env python3
"""
GitHub App User Token Generator
Generates repo-scoped GitHub user access tokens via the device flow.
Only requires the GitHub App's Client ID (no secrets needed).
Usage:
# Print instructions to create the GitHub App (one-time setup)
python3 github_app_token_demo.py create-app
# Generate a user access token via device flow
python3 github_app_token_demo.py user-token --client-id Iv23liXXXXXX
# Scope the token to a single repo (writes only)
python3 github_app_token_demo.py user-token --client-id Iv23liXXXXXX --repo wirenboard/agent-vm
Add -v/--verbose to any command for detailed HTTP logging.
"""
import argparse
import json
import os
import sys
import re
import time
import urllib.parse
import urllib.request
import urllib.error
TARGET_OWNER = os.environ.get("TARGET_OWNER", "wirenboard")
TARGET_REPO = os.environ.get("TARGET_REPO", "agent-vm")
VERBOSE = False
def verbose(msg):
if VERBOSE:
print(f" [verbose] {msg}", file=sys.stderr)
def parse_repo(repo_str):
"""Parse a repo URL/slug into (owner, name). Accepts:
- owner/name
- https://github.com/owner/name
- git@github.com:owner/name.git
"""
# git@github.com:owner/name.git
m = re.match(r'git@github\.com:([^/]+)/([^/]+?)(?:\.git)?$', repo_str)
if m:
return m.group(1), m.group(2)
# https://github.com/owner/name[.git]
m = re.match(r'https?://github\.com/([^/]+)/([^/]+?)(?:\.git)?(?:/.*)?$', repo_str)
if m:
return m.group(1), m.group(2)
# owner/name
m = re.match(r'^([^/]+)/([^/]+)$', repo_str)
if m:
return m.group(1), m.group(2)
print(f"Error: cannot parse repo: {repo_str}", file=sys.stderr)
sys.exit(1)
def resolve_repo_id(owner, name):
"""Resolve a repo's numeric ID. Tries public API first, falls back to gh CLI."""
import subprocess
url = f"https://api.github.com/repos/{owner}/{name}"
req = urllib.request.Request(url)
req.add_header("Accept", "application/vnd.github+json")
verbose(f"Resolving repo ID: {owner}/{name}")
try:
with urllib.request.urlopen(req) as resp:
data = json.loads(resp.read().decode())
repo_id = data["id"]
verbose(f"Resolved {owner}/{name} -> {repo_id}")
return str(repo_id)
except urllib.error.HTTPError:
verbose(f"Public API failed, trying gh CLI...")
try:
result = subprocess.run(
["gh", "api", f"repos/{owner}/{name}", "--jq", ".id"],
capture_output=True, text=True, check=True,
)
repo_id = result.stdout.strip()
verbose(f"Resolved {owner}/{name} -> {repo_id} (via gh)")
return repo_id
except (subprocess.CalledProcessError, FileNotFoundError) as e:
print(f"Error: cannot resolve repo {owner}/{name}", file=sys.stderr)
print(f" Public API returned 404 and gh CLI failed: {e}", file=sys.stderr)
print(f" Use --repository-id to pass the numeric ID directly", file=sys.stderr)
sys.exit(1)
# --- App creation helper ---
APP_MANIFEST = {
"name": "WB Agent VM Token Generator",
"url": "https://github.com/wirenboard/agent-vm",
"hook_attributes": {"active": False},
"redirect_url": "",
"public": False,
"default_permissions": {"contents": "write"},
"default_events": [],
}
def cmd_create_app():
"""Print instructions for creating the GitHub App via the UI."""
print("=" * 60)
print("GitHub App Creation Instructions")
print("=" * 60)
print()
print("1. Go to: https://github.com/organizations/wirenboard/settings/apps/new")
print(" (Or for personal account: https://github.com/settings/apps/new)")
print()
print("2. Fill in the following settings:")
print(f" - GitHub App name: {APP_MANIFEST['name']}")
print(f" - Homepage URL: {APP_MANIFEST['url']}")
print(" - Uncheck 'Active' under Webhook")
print(" - Under 'Repository permissions':")
print(" - Contents: Read & write")
print(" - Under 'Where can this GitHub App be installed?':")
print(" - Select 'Only on this account'")
print()
print("3. Click 'Create GitHub App'")
print()
print("4. Note the Client ID (starts with Iv...)")
print()
print("5. Install the app:")
print(" - Click 'Install App' in the left sidebar")
print(" - Choose the target organization/account")
print(" - Select repositories to grant access to")
print()
print("6. Generate a token:")
print(f" python3 {sys.argv[0]} user-token --client-id <CLIENT_ID> --repo <owner/repo>")
print()
# --- GitHub API helper ---
def github_api(method, path, token, token_type="Bearer", body=None):
"""Make a GitHub API request."""
url = f"https://api.github.com{path}"
data = json.dumps(body).encode() if body else None
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Accept", "application/vnd.github+json")
req.add_header("Authorization", f"{token_type} {token}")
req.add_header("X-GitHub-Api-Version", "2022-11-28")
if data:
req.add_header("Content-Type", "application/json")
verbose(f">>> {method} {url}")
if body:
verbose(f">>> Body: {json.dumps(body)}")
try:
with urllib.request.urlopen(req) as resp:
resp_body = resp.read().decode()
verbose(f"<<< {resp.status} {resp.reason}")
verbose(f"<<< Body: {resp_body[:500]}{'...' if len(resp_body) > 500 else ''}")
return json.loads(resp_body)
except urllib.error.HTTPError as e:
error_body = e.read().decode()
verbose(f"<<< {e.code} {e.reason}")
verbose(f"<<< Body: {error_body}")
print(f"GitHub API error: {e.code} {e.reason}", file=sys.stderr)
print(f"URL: {url}", file=sys.stderr)
print(f"Response: {error_body}", file=sys.stderr)
sys.exit(1)
def verify_token(token, token_type="token"):
"""Verify the token works by reading the target repo."""
repo = github_api(
"GET",
f"/repos/{TARGET_OWNER}/{TARGET_REPO}",
token,
token_type=token_type,
)
return repo
# --- Token cache ---
DEFAULT_CACHE_DIR = os.path.expanduser("~/.cache/claude-vm")
def _cache_path(cache_dir, client_id, owner, repo):
"""Return the cache file path for a given client/repo combo."""
safe = f"{owner}_{repo}".replace("/", "_")
return os.path.join(cache_dir, f"github-token-{safe}.json")
def load_cached_token(cache_dir, client_id, owner, repo):
"""Try to load a valid token from cache. Returns token string or None."""
path = _cache_path(cache_dir, client_id, owner, repo)
try:
with open(path, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return None
expires_at = data.get("expires_at", 0)
# 5 minute margin
if time.time() < expires_at - 300:
verbose(f"Using cached token (expires in {int(expires_at - time.time())}s)")
return data.get("access_token")
# Token expired — try refresh
refresh_token = data.get("refresh_token")
if not refresh_token:
verbose("Cached token expired, no refresh token")
return None
verbose("Cached token expired, attempting refresh...")
try:
resp = device_flow_request(
"https://github.com/login/oauth/access_token",
{
"client_id": client_id,
"grant_type": "refresh_token",
"refresh_token": refresh_token,
},
)
except SystemExit:
verbose("Refresh failed, will re-authenticate")
return None
if resp.get("error"):
verbose(f"Refresh error: {resp.get('error')}")
return None
token = resp.get("access_token")
if not token:
return None
verbose("Token refreshed successfully")
save_cached_token(
cache_dir, client_id, owner, repo,
token,
resp.get("refresh_token", refresh_token),
resp.get("expires_in", 28800),
)
return token
def save_cached_token(cache_dir, client_id, owner, repo,
access_token, refresh_token, expires_in):
"""Save token to cache file."""
path = _cache_path(cache_dir, client_id, owner, repo)
os.makedirs(os.path.dirname(path), exist_ok=True)
data = {
"access_token": access_token,
"refresh_token": refresh_token,
"expires_at": time.time() + (int(expires_in) if expires_in != "never" else 86400),
}
with open(path, "w") as f:
json.dump(data, f)
# Restrict permissions — token file
os.chmod(path, 0o600)
verbose(f"Token cached to {path}")
# --- Device flow (user access token) ---
def device_flow_request(url, params):
"""POST to GitHub's OAuth endpoints (form-encoded, JSON response)."""
data = urllib.parse.urlencode(params).encode()
req = urllib.request.Request(url, data=data, method="POST")
req.add_header("Accept", "application/json")
verbose(f">>> POST {url}")
verbose(f">>> Body: {urllib.parse.urlencode(params)}")
try:
with urllib.request.urlopen(req) as resp:
resp_body = resp.read().decode()
verbose(f"<<< {resp.status} {resp.reason}")
verbose(f"<<< Body: {resp_body[:500]}{'...' if len(resp_body) > 500 else ''}")
return json.loads(resp_body)
except urllib.error.HTTPError as e:
error_body = e.read().decode()
verbose(f"<<< {e.code} {e.reason}")
verbose(f"<<< Body: {error_body}")
print(f"OAuth error: {e.code} {e.reason}", file=sys.stderr)
print(f"Response: {error_body}", file=sys.stderr)
sys.exit(1)
def cmd_user_token(client_id, repository_id=None, token_only=False,
cache_dir=None):
"""Generate a user access token via the device flow."""
out = sys.stderr if token_only else sys.stdout
# Try cached token first
if cache_dir:
cached = load_cached_token(cache_dir, client_id, TARGET_OWNER, TARGET_REPO)
if cached:
print(f"Using cached token for {TARGET_OWNER}/{TARGET_REPO}", file=out)
if token_only:
print(cached)
else:
print("=" * 60)
print(f"GITHUB_TOKEN={cached}")
print("=" * 60)
return
print("Requesting device code...", file=out)
resp = device_flow_request(
"https://github.com/login/device/code",
{"client_id": client_id},
)
device_code = resp["device_code"]
user_code = resp["user_code"]
verification_uri = resp["verification_uri"]
expires_in = resp["expires_in"]
interval = resp.get("interval", 5)
verbose(f"Device code: {device_code}")
verbose(f"Polling interval: {interval}s, expires in: {expires_in}s")
print(file=out)
print("=" * 60, file=out)
print(f" Open: {verification_uri}", file=out)
print(f" Enter: {user_code}", file=out)
print("=" * 60, file=out)
print(file=out)
print(f"Waiting for authorization (expires in {expires_in}s)...", file=out)
# Poll for the token
poll_count = 0
while True:
time.sleep(interval)
poll_count += 1
verbose(f"Polling attempt #{poll_count}...")
poll_params = {
"client_id": client_id,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
}
if repository_id:
poll_params["repository_id"] = repository_id
token_resp = device_flow_request(
"https://github.com/login/oauth/access_token",
poll_params,
)
error = token_resp.get("error")
if error == "authorization_pending":
verbose("Status: authorization_pending")
continue
elif error == "slow_down":
interval = token_resp.get("interval", interval + 5)
verbose(f"Status: slow_down, new interval: {interval}s")
continue
elif error == "expired_token":
print("Error: Device code expired. Please try again.", file=sys.stderr)
sys.exit(1)
elif error == "access_denied":
print("Error: Authorization was denied by the user.", file=sys.stderr)
sys.exit(1)
elif error:
print(f"Error: {error} - {token_resp.get('error_description', '')}", file=sys.stderr)
sys.exit(1)
# Success
token = token_resp["access_token"]
token_type = token_resp.get("token_type", "bearer")
expires = token_resp.get("expires_in", "never")
refresh = token_resp.get("refresh_token", "")
verbose(f"Token type: {token_type}")
verbose(f"Full token response keys: {list(token_resp.keys())}")
break
print("Authorization successful!", file=out)
print(file=out)
if expires != "never":
print(f"Token expires in: {expires}s", file=out)
if refresh:
print(f"Refresh token: {refresh}", file=out)
print(file=out)
print("Verifying token...", file=out)
repo = verify_token(token, token_type="Bearer")
print(f"Verified: can access {repo['full_name']}", file=out)
print(file=out)
# Cache the token
if cache_dir:
save_cached_token(cache_dir, client_id, TARGET_OWNER, TARGET_REPO,
token, refresh, expires)
if token_only:
print(token)
else:
print("=" * 60)
print(f"GITHUB_TOKEN={token}")
print("=" * 60)
# --- CLI ---
def main():
global VERBOSE
parser = argparse.ArgumentParser(
description="Generate repo-scoped GitHub user tokens via a GitHub App"
)
parser.add_argument("-v", "--verbose", action="store_true",
help="Enable verbose HTTP logging to stderr")
sub = parser.add_subparsers(dest="command")
sub.add_parser("create-app", help="Print instructions to create the GitHub App")
user_parser = sub.add_parser("user-token", help="Generate a user access token via device flow")
user_parser.add_argument("--client-id", required=True, help="GitHub App Client ID (Iv23li...)")
user_parser.add_argument("--repo", type=str, default=None,
help="Repository to scope the token to (owner/name, URL, or git@github.com:owner/name.git)")
user_parser.add_argument("--repository-id", type=str, default=None,
help="Numeric repo ID (use for private repos where --repo can't resolve)")
user_parser.add_argument("--token-only", action="store_true",
help="Print only the token to stdout (status messages go to stderr)")
user_parser.add_argument("--cache-dir", type=str, default=None,
help="Directory to cache tokens (default: no caching)")
args = parser.parse_args()
VERBOSE = args.verbose
if args.command == "create-app":
cmd_create_app()
elif args.command == "user-token":
global TARGET_OWNER, TARGET_REPO
repository_id = args.repository_id
if args.repo:
TARGET_OWNER, TARGET_REPO = parse_repo(args.repo)
if not repository_id:
repository_id = resolve_repo_id(TARGET_OWNER, TARGET_REPO)
cmd_user_token(args.client_id, repository_id=repository_id,
token_only=args.token_only,
cache_dir=args.cache_dir)
else:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()

445
test_proxy_e2e.py Normal file
View file

@ -0,0 +1,445 @@
#!/usr/bin/env python3
"""End-to-end test for the proxy scope enforcement.
Starts a mock MCP server (upstream) and the proxy in front of it,
then sends requests through the proxy and verifies scoping.
"""
import http.server
import json
import os
import socketserver
import sys
import threading
import time
import urllib.request
import urllib.error
# --- Mock upstream MCP server ---
# Records what it receives so we can verify the proxy modified the request.
received_requests = []
received_headers = []
class MockMCPHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length) if length else b""
req = json.loads(body) if body else {}
received_requests.append(req)
# Capture headers as a dict for inspection
received_headers.append(dict(self.headers))
# Return a mock MCP response
result = {"content": [{"type": "text", "text": json.dumps({"mock": True})}]}
resp = json.dumps({"jsonrpc": "2.0", "id": req.get("id"), "result": result})
resp_bytes = resp.encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(resp_bytes)))
self.end_headers()
self.wfile.write(resp_bytes)
def do_GET(self):
self.do_POST()
def start_mock_server():
server = socketserver.TCPServer(("127.0.0.1", 0), MockMCPHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, port
# --- Proxy setup ---
def start_proxy(upstream_host, upstream_port, extra_env=None):
"""Start the proxy pointing at our mock upstream."""
# We need to patch the proxy module to use HTTP instead of HTTPS
# and point at our mock server
import importlib.util
spec = importlib.util.spec_from_file_location(
"proxy_mod", os.path.join(os.path.dirname(__file__), "github-mcp-proxy.py"))
proxy = importlib.util.module_from_spec(spec)
# Override env before loading
os.environ["GITHUB_MCP_TOKEN"] = "test-token-broader-scope"
os.environ["GITHUB_MCP_OWNER"] = "wirenboard"
os.environ["GITHUB_MCP_REPO"] = "agent-vm"
os.environ["GITHUB_MCP_PROXY_DEBUG"] = "1"
# Clear optional env vars by default
for key in ("GITHUB_MCP_TOOLSETS", "GITHUB_MCP_TOOLS", "GITHUB_MCP_READONLY"):
os.environ.pop(key, None)
if extra_env:
os.environ.update(extra_env)
spec.loader.exec_module(proxy)
# Monkey-patch to use our mock server instead of GitHub
proxy.MCP_HOST = upstream_host
proxy.MCP_PORT = upstream_port
proxy.MCP_PATH = "/"
# Patch do_request to use HTTP instead of HTTPS
original_do_request = proxy.GitHubMCPProxyHandler.do_request
def patched_do_request(self):
import http.client as hc
# Store original HTTPSConnection and replace with HTTPConnection
orig_https = hc.HTTPSConnection
hc.HTTPSConnection = lambda host, port, **kw: hc.HTTPConnection(host, port, timeout=kw.get("timeout", 30))
try:
original_do_request(self)
finally:
hc.HTTPSConnection = orig_https
proxy.GitHubMCPProxyHandler.do_request = patched_do_request
proxy.GitHubMCPProxyHandler.do_GET = patched_do_request
proxy.GitHubMCPProxyHandler.do_POST = patched_do_request
server = socketserver.TCPServer(("127.0.0.1", 0), proxy.GitHubMCPProxyHandler)
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
return server, port
def mcp_request(proxy_port, tool_name, arguments):
"""Send an MCP tools/call through the proxy."""
body = json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": tool_name, "arguments": arguments}
}).encode()
req = urllib.request.Request(
f"http://127.0.0.1:{proxy_port}/mcp",
data=body,
headers={
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
)
try:
with urllib.request.urlopen(req) as resp:
return resp.status, json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read().decode())
# --- Tests ---
def main():
print("Starting mock upstream MCP server...")
mock_server, mock_port = start_mock_server()
print(f" Mock server on port {mock_port}")
print("Starting proxy...")
proxy_server, proxy_port = start_proxy("127.0.0.1", mock_port)
print(f" Proxy on port {proxy_port}")
time.sleep(0.2)
passed = 0
failed = 0
def check(label, status, resp, expect_blocked=False, check_upstream_query=None):
nonlocal passed, failed
if expect_blocked:
if status == 403 and "error" in resp:
print(f" PASS [{label}]: blocked - {resp['error']['message']}")
passed += 1
else:
print(f" FAIL [{label}]: expected 403 block, got {status}")
failed += 1
else:
if status == 200:
if check_upstream_query is not None:
last_req = received_requests[-1]
actual_query = last_req["params"]["arguments"]["query"]
if actual_query == check_upstream_query:
print(f" PASS [{label}]: upstream saw query={actual_query!r}")
passed += 1
else:
print(f" FAIL [{label}]: expected query={check_upstream_query!r}, got {actual_query!r}")
failed += 1
else:
print(f" PASS [{label}]: allowed (status {status})")
passed += 1
else:
print(f" FAIL [{label}]: expected 200, got {status}: {resp}")
failed += 1
print()
print("=== E2E: search_code with no scope gets repo injected ===")
received_requests.clear()
status, resp = mcp_request(proxy_port, "search_code", {"query": "def main"})
check("search_code unscoped", status, resp,
check_upstream_query="repo:wirenboard/agent-vm def main")
print()
print("=== E2E: search_code targeting wrong repo is BLOCKED ===")
status, resp = mcp_request(proxy_port, "search_code",
{"query": "repo:wirenboard/exposure-probe secret"})
check("search_code wrong repo", status, resp, expect_blocked=True)
print()
print("=== E2E: search_repositories gets repo scope injected ===")
received_requests.clear()
status, resp = mcp_request(proxy_port, "search_repositories",
{"query": "language:python"})
check("search_repos unscoped", status, resp,
check_upstream_query="repo:wirenboard/agent-vm language:python")
print()
print("=== E2E: create_branch on wrong repo is BLOCKED ===")
status, resp = mcp_request(proxy_port, "create_branch",
{"owner": "wirenboard", "repo": "exposure-probe",
"branch": "evil-branch"})
check("create_branch wrong repo", status, resp, expect_blocked=True)
print()
print("=== E2E: create_branch on correct repo passes through ===")
status, resp = mcp_request(proxy_port, "create_branch",
{"owner": "wirenboard", "repo": "agent-vm",
"branch": "legit-branch"})
check("create_branch correct repo", status, resp)
print()
print("=== E2E: get_me (unscoped tool) passes through ===")
status, resp = mcp_request(proxy_port, "get_me", {})
check("get_me", status, resp)
print()
print("=== E2E: get_teams (org tool) is BLOCKED ===")
status, resp = mcp_request(proxy_port, "get_teams", {"user": "someone"})
check("get_teams", status, resp, expect_blocked=True)
print()
print("=== E2E: search_users (not repo-scoped) is BLOCKED ===")
status, resp = mcp_request(proxy_port, "search_users", {"query": "john"})
check("search_users", status, resp, expect_blocked=True)
print()
print("=== E2E: org:/user: qualifiers in search queries BLOCKED ===")
status, resp = mcp_request(proxy_port, "search_code",
{"query": "org:wirenboard def main"})
check("search_code with org:", status, resp, expect_blocked=True)
status, resp = mcp_request(proxy_port, "search_issues",
{"query": "user:someone is:open"})
check("search_issues with user:", status, resp, expect_blocked=True)
print()
print("=== E2E: unknown tools BLOCKED (default-deny) ===")
status, resp = mcp_request(proxy_port, "totally_new_tool",
{"owner": "wirenboard", "repo": "agent-vm"})
check("unknown tool", status, resp, expect_blocked=True)
print()
print("=== E2E: list_issues without owner/repo gets them injected ===")
received_requests.clear()
status, resp = mcp_request(proxy_port, "list_issues", {"state": "OPEN"})
check("list_issues no owner/repo", status, resp)
# Verify upstream received the injected values
last_req = received_requests[-1]
args = last_req["params"]["arguments"]
assert args["owner"] == "wirenboard", f"owner not injected: {args}"
assert args["repo"] == "agent-vm", f"repo not injected: {args}"
print(f" PASS [list_issues injection verified]: owner={args['owner']}, repo={args['repo']}")
passed += 1
# --- Test tool-filtering headers ---
# Stop first proxy, start a new one with TOOLSETS/TOOLS/READONLY
proxy_server.shutdown()
print()
print("=== E2E: tool-filtering headers (X-MCP-Toolsets, X-MCP-Tools, X-MCP-Readonly) ===")
proxy_server2, proxy_port2 = start_proxy("127.0.0.1", mock_port, extra_env={
"GITHUB_MCP_TOOLSETS": "repos,issues",
"GITHUB_MCP_TOOLS": "get_file_contents,issue_read",
"GITHUB_MCP_READONLY": "1",
})
time.sleep(0.2)
received_headers.clear()
status, resp = mcp_request(proxy_port2, "get_file_contents",
{"owner": "wirenboard", "repo": "agent-vm", "path": "/"})
if status == 200 and received_headers:
h = received_headers[-1]
checks = [
("X-MCP-Toolsets" in h and h["X-MCP-Toolsets"] == "repos,issues",
"X-MCP-Toolsets"),
("X-MCP-Tools" in h and h["X-MCP-Tools"] == "get_file_contents,issue_read",
"X-MCP-Tools"),
("X-MCP-Readonly" in h and h["X-MCP-Readonly"] == "true",
"X-MCP-Readonly"),
]
for ok, name in checks:
if ok:
print(f" PASS [{name} header]: upstream received {name}={h.get(name)!r}")
passed += 1
else:
print(f" FAIL [{name} header]: not found or wrong value in upstream headers: {h}")
failed += 1
else:
print(f" FAIL [header injection]: request failed with status {status}")
failed += 3
# Verify VM can't override these headers
received_headers.clear()
body = json.dumps({
"jsonrpc": "2.0", "id": 1,
"method": "tools/call",
"params": {"name": "get_file_contents",
"arguments": {"owner": "wirenboard", "repo": "agent-vm", "path": "/"}}
}).encode()
req = urllib.request.Request(
f"http://127.0.0.1:{proxy_port2}/mcp",
data=body,
headers={
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"X-MCP-Toolsets": "repos,issues,pull_requests,actions,users,orgs",
"X-MCP-Tools": "create_branch,push_files,delete_file",
"X-MCP-Readonly": "false",
},
)
try:
with urllib.request.urlopen(req) as resp_obj:
pass
except urllib.error.HTTPError:
pass
if received_headers:
h = received_headers[-1]
# The proxy should strip the VM's values and inject its own
vm_override_checks = [
(h.get("X-MCP-Toolsets") == "repos,issues",
"X-MCP-Toolsets not overridden by VM"),
(h.get("X-MCP-Tools") == "get_file_contents,issue_read",
"X-MCP-Tools not overridden by VM"),
(h.get("X-MCP-Readonly") == "true",
"X-MCP-Readonly not overridden by VM"),
]
for ok, label in vm_override_checks:
if ok:
print(f" PASS [{label}]")
passed += 1
else:
print(f" FAIL [{label}]: VM override leaked through: {h}")
failed += 1
else:
print(" FAIL [VM override test]: no headers captured")
failed += 3
proxy_server2.shutdown()
# --- Test defaults (toolsets + lockdown enabled, tools/readonly absent) ---
print()
print("=== E2E: default headers (toolsets + lockdown on, tools/readonly off) ===")
proxy_server3, proxy_port3 = start_proxy("127.0.0.1", mock_port)
time.sleep(0.2)
received_headers.clear()
status, resp = mcp_request(proxy_port3, "get_file_contents",
{"owner": "wirenboard", "repo": "agent-vm", "path": "/"})
if status == 200 and received_headers:
h = received_headers[-1]
default_checks = [
(h.get("X-MCP-Toolsets") == "repos,issues,pull_requests,git,labels",
"X-MCP-Toolsets default"),
("X-MCP-Tools" not in h,
"X-MCP-Tools absent by default"),
("X-MCP-Readonly" not in h,
"X-MCP-Readonly absent by default"),
(h.get("X-MCP-Lockdown") == "true",
"X-MCP-Lockdown enabled by default"),
]
for ok, label in default_checks:
if ok:
print(f" PASS [{label}]: {h.get('X-MCP-Toolsets', '(absent)')}, lockdown={h.get('X-MCP-Lockdown', '(absent)')}")
passed += 1
else:
print(f" FAIL [{label}]: headers={h}")
failed += 1
else:
print(f" FAIL [defaults test]: request failed with status {status}")
failed += 4
# Verify VM can't inject X-MCP-Lockdown
received_headers.clear()
body = json.dumps({
"jsonrpc": "2.0", "id": 1,
"method": "tools/call",
"params": {"name": "get_file_contents",
"arguments": {"owner": "wirenboard", "repo": "agent-vm", "path": "/"}}
}).encode()
req = urllib.request.Request(
f"http://127.0.0.1:{proxy_port3}/mcp",
data=body,
headers={
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"X-MCP-Lockdown": "false",
},
)
try:
with urllib.request.urlopen(req) as resp_obj:
pass
except urllib.error.HTTPError:
pass
if received_headers:
h = received_headers[-1]
if h.get("X-MCP-Lockdown") == "true":
print(f" PASS [X-MCP-Lockdown not overridden by VM]")
passed += 1
else:
print(f" FAIL [X-MCP-Lockdown overridden by VM]: {h.get('X-MCP-Lockdown')}")
failed += 1
else:
print(" FAIL [lockdown override test]: no headers captured")
failed += 1
proxy_server3.shutdown()
# --- Test lockdown disabled ---
print()
print("=== E2E: lockdown disabled when GITHUB_MCP_LOCKDOWN=0 ===")
proxy_server4, proxy_port4 = start_proxy("127.0.0.1", mock_port, extra_env={
"GITHUB_MCP_LOCKDOWN": "0",
})
time.sleep(0.2)
received_headers.clear()
status, resp = mcp_request(proxy_port4, "get_file_contents",
{"owner": "wirenboard", "repo": "agent-vm", "path": "/"})
if status == 200 and received_headers:
h = received_headers[-1]
if "X-MCP-Lockdown" not in h:
print(f" PASS [X-MCP-Lockdown absent when disabled]")
passed += 1
else:
print(f" FAIL [X-MCP-Lockdown should be absent]: {h.get('X-MCP-Lockdown')}")
failed += 1
else:
print(f" FAIL [lockdown disabled test]: request failed")
failed += 1
proxy_server4.shutdown()
print()
print(f"{'=' * 50}")
print(f"Results: {passed} passed, {failed} failed")
if failed:
sys.exit(1)
else:
print("All end-to-end tests passed!")
if __name__ == "__main__":
main()

171
test_proxy_scope.py Normal file
View file

@ -0,0 +1,171 @@
#!/usr/bin/env python3
"""Tests for the repo scope enforcement in github-mcp-proxy.py."""
import json
import os
import sys
# Set env before importing
os.environ["GITHUB_MCP_TOKEN"] = "fake-token"
os.environ["GITHUB_MCP_OWNER"] = "wirenboard"
os.environ["GITHUB_MCP_REPO"] = "agent-vm"
os.environ["GITHUB_MCP_PROXY_DEBUG"] = "1"
# Reload module to pick up env
if "github-mcp-proxy" in sys.modules:
del sys.modules["github-mcp-proxy"]
# Import the proxy module
import importlib.util
spec = importlib.util.spec_from_file_location(
"proxy", os.path.join(os.path.dirname(__file__), "github-mcp-proxy.py"))
proxy = importlib.util.module_from_spec(spec)
spec.loader.exec_module(proxy)
def make_tool_call(tool_name, arguments):
"""Build an MCP tools/call request body."""
return json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments,
}
}).encode()
def check(label, body_bytes, expect_blocked=False, expect_query=None,
expect_unchanged=False):
"""Run enforce_repo_scope and check the result."""
result, err = proxy.enforce_repo_scope(body_bytes)
if expect_blocked:
assert err is not None, f"FAIL [{label}]: expected BLOCKED but got through"
print(f" PASS [{label}]: blocked with: {err}")
else:
assert err is None, f"FAIL [{label}]: expected ALLOWED but blocked: {err}"
if expect_query is not None:
req = json.loads(result)
actual_query = req["params"]["arguments"]["query"]
assert actual_query == expect_query, \
f"FAIL [{label}]: expected query={expect_query!r}, got {actual_query!r}"
print(f" PASS [{label}]: allowed, query={actual_query!r}")
elif expect_unchanged:
assert result == body_bytes, \
f"FAIL [{label}]: body was re-serialized when it shouldn't have been"
print(f" PASS [{label}]: allowed, body unchanged")
else:
print(f" PASS [{label}]: allowed")
print("=== Test: existing owner/repo check still works ===")
check("correct repo",
make_tool_call("create_branch", {"owner": "wirenboard", "repo": "agent-vm", "branch": "test"}),
expect_unchanged=True)
check("wrong repo",
make_tool_call("create_branch", {"owner": "wirenboard", "repo": "exposure-probe", "branch": "test"}),
expect_blocked=True)
check("wrong owner",
make_tool_call("create_branch", {"owner": "evil-org", "repo": "agent-vm", "branch": "test"}),
expect_blocked=True)
print()
print("=== Test: search tools get repo scope injected ===")
check("search_code no scope",
make_tool_call("search_code", {"query": "def main"}),
expect_query="repo:wirenboard/agent-vm def main")
check("search_code correct scope",
make_tool_call("search_code", {"query": "repo:wirenboard/agent-vm def main"}),
expect_query="repo:wirenboard/agent-vm def main")
check("search_code wrong scope",
make_tool_call("search_code", {"query": "repo:wirenboard/exposure-probe def main"}),
expect_blocked=True)
check("search_repositories no scope",
make_tool_call("search_repositories", {"query": "language:python"}),
expect_query="repo:wirenboard/agent-vm language:python")
check("search_issues no scope",
make_tool_call("search_issues", {"query": "is:open bug"}),
expect_query="repo:wirenboard/agent-vm is:open bug")
check("search_pull_requests wrong scope",
make_tool_call("search_pull_requests", {"query": "repo:evil/repo is:open"}),
expect_blocked=True)
print()
print("=== Test: org:/user: qualifiers blocked in search queries ===")
check("search_code with org:",
make_tool_call("search_code", {"query": "org:wirenboard def main"}),
expect_blocked=True)
check("search_code with user:",
make_tool_call("search_code", {"query": "user:evgeny-boger password"}),
expect_blocked=True)
check("search_issues with org:",
make_tool_call("search_issues", {"query": "org:evil-corp is:open"}),
expect_blocked=True)
print()
print("=== Test: search_users and search_orgs blocked ===")
check("search_users",
make_tool_call("search_users", {"query": "john"}),
expect_blocked=True)
check("search_orgs",
make_tool_call("search_orgs", {"query": "wirenboard"}),
expect_blocked=True)
print()
print("=== Test: unscoped tools allowed ===")
check("get_me",
make_tool_call("get_me", {}),
expect_unchanged=True)
print()
print("=== Test: org-level tools blocked ===")
check("get_teams",
make_tool_call("get_teams", {"user": "someone"}),
expect_blocked=True)
check("get_team_members",
make_tool_call("get_team_members", {"org": "wirenboard", "team_slug": "devs"}),
expect_blocked=True)
check("list_issue_types",
make_tool_call("list_issue_types", {"owner": "wirenboard"}),
expect_blocked=True)
print()
print("=== Test: unknown tools blocked (default-deny) ===")
check("totally_fake_tool",
make_tool_call("totally_fake_tool", {"owner": "wirenboard", "repo": "agent-vm"}),
expect_blocked=True)
check("future_dangerous_tool",
make_tool_call("future_dangerous_tool", {"query": "secrets"}),
expect_blocked=True)
print()
print("=== Test: optional owner/repo gets injected ===")
check("list_issues missing owner/repo",
make_tool_call("list_issues", {"state": "OPEN"}))
# Verify the injected values
result, _ = proxy.enforce_repo_scope(
make_tool_call("list_issues", {"state": "OPEN"}))
req = json.loads(result)
assert req["params"]["arguments"]["owner"] == "wirenboard", "owner not injected"
assert req["params"]["arguments"]["repo"] == "agent-vm", "repo not injected"
print(" PASS [list_issues injection]: owner/repo injected correctly")
print()
print("=== Test: body not re-serialized when unmodified ===")
# When owner/repo are already correct, body should pass through as-is
check("get_file_contents unchanged",
make_tool_call("get_file_contents",
{"owner": "wirenboard", "repo": "agent-vm", "path": "/"}),
expect_unchanged=True)
print()
print("=== Test: non-tool-call methods pass through ===")
init_body = json.dumps({"jsonrpc": "2.0", "id": 0, "method": "initialize",
"params": {"protocolVersion": "2024-11-05"}}).encode()
result, err = proxy.enforce_repo_scope(init_body)
assert err is None and result == init_body
print(" PASS [initialize]: passed through unchanged")
print()
print("All tests passed!")