Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  jcwalker3andClaude Opus 4.8
|
0104a76eea | ||
|
jcwalker3andClaude Opus 4.8
|
a143cd065b | ||
|
jcwalker3andClaude Opus 4.8
|
2fb835a1aa | ||
|
jcwalker3andClaude Opus 4.8
|
c162608175 | ||
 sysadmin
|
9b80e75ca3 | ||
|
sysadminandClaude Opus 4.8
|
b3de9c941c | ||
|
sysadmin
|
aad5c8b423 |
@@ -0,0 +1,80 @@
|
||||
{
|
||||
"_comment": [
|
||||
"Machine-checkable anchor table for docs/remote-mcp/threat-model.md (#956).",
|
||||
"Every file:line anchor cited in the threat model must appear here, and the",
|
||||
"source line at that anchor must contain the 'expect' substring.",
|
||||
"tests/test_issue_956_threat_model.py enforces both directions, so a refactor",
|
||||
"that shifts a line number fails the suite instead of silently rotting the",
|
||||
"document. #930's inventory had no such guard and its gitea_mcp_server.py",
|
||||
"anchors drifted between 7bf4f125 and aad5c8b4."
|
||||
],
|
||||
"generated_against_commit": "a143cd065ba06e1a2bdc5143a19ec156e53650ef",
|
||||
"anchors": [
|
||||
{"anchor": "gitea_mcp_server.py:24728", "expect": "mcp_daemon_guard.bind_native_mcp_transport()"},
|
||||
{"anchor": "mcp_daemon_guard.py:49", "expect": "_PRODUCTION_TRANSPORTS = mcp_transport_config.SUPPORTED_TRANSPORTS"},
|
||||
{"anchor": "mcp_daemon_guard.py:195", "expect": "def bind_native_mcp_transport"},
|
||||
{"anchor": "irrecoverable_provenance.py:497", "expect": "def assess_transport_for_auth_mint"},
|
||||
{"anchor": "gitea_mcp_server.py:9132", "expect": "assess_transport_for_auth_mint()"},
|
||||
{"anchor": "gitea_mcp_server.py:9381", "expect": "assess_transport_for_auth_mint()"},
|
||||
{"anchor": "mcp_server.py:4", "expect": "The transport is selected by deployment configuration"},
|
||||
|
||||
{"anchor": "gitea_mcp_server.py:15415", "expect": "def _is_client_managed_process"},
|
||||
{"anchor": "gitea_mcp_server.py:15445", "expect": "def _provenance_mutation_block"},
|
||||
{"anchor": "gitea_mcp_server.py:15453", "expect": "unsupported_manual_launch"},
|
||||
{"anchor": "gitea_mcp_server.py:19004", "expect": "server_provenance"},
|
||||
{"anchor": "gitea_mcp_server.py:21445", "expect": "def _check_mcp_runtimes_diagnostics"},
|
||||
{"anchor": "gitea_mcp_server.py:21465", "expect": "\"ps\", \"-o\", \"pid,lstart,command\""},
|
||||
{"anchor": "gitea_mcp_server.py:21509", "expect": "\"ps\", \"eww\""},
|
||||
{"anchor": "gitea_config.py:1172", "expect": "RECOGNIZED_GITEA_ENV_KEYS"},
|
||||
{"anchor": "gitea_config.py:1233", "expect": "GITEA_CLIENT_MANAGED"},
|
||||
|
||||
{"anchor": "gitea_config.py:54", "expect": "ENV_PROFILE = \"GITEA_MCP_PROFILE\""},
|
||||
{"anchor": "gitea_config.py:97", "expect": "_REVIEW_MERGE_OPS"},
|
||||
{"anchor": "gitea_config.py:499", "expect": "repository authorization scope"},
|
||||
|
||||
{"anchor": "gitea_config.py:956", "expect": "def _keychain_token"},
|
||||
{"anchor": "gitea_config.py:974", "expect": "def resolve_token"},
|
||||
{"anchor": "gitea_config.py:1015", "expect": "def keychain_auth"},
|
||||
{"anchor": "gitea_config.py:294", "expect": "def _validate_identity_auth"},
|
||||
{"anchor": "mcp_daemon_guard.py:583", "expect": "def assert_keychain_access_allowed"},
|
||||
{"anchor": "gitea_mcp_server.py:19261", "expect": "def gitea_list_profiles"},
|
||||
{"anchor": "gitea_mcp_server.py:19312", "expect": "gitea_config.resolve_token(p)"},
|
||||
{"anchor": "gitea_mcp_server.py:19555", "expect": "def gitea_audit_config"},
|
||||
{"anchor": "gitea_mcp_server.py:19577", "expect": "service_summaries(config)"},
|
||||
|
||||
{"anchor": "gitea_config.py:704", "expect": "def resolve_service"},
|
||||
{"anchor": "gitea_config.py:837", "expect": "def service_summaries"},
|
||||
{"anchor": "gitea_config.py:851", "expect": "_keychain_token(auth.get(\"id\"))"},
|
||||
{"anchor": "gitea_mcp_server.py:17710", "expect": "\"jenkins-mcp\""},
|
||||
{"anchor": "gitea_mcp_server.py:17716", "expect": "external-mcp"},
|
||||
{"anchor": "gitea_mcp_server.py:17737", "expect": "\"glitchtip-mcp\""},
|
||||
{"anchor": "gitea_mcp_server.py:17742", "expect": "external-mcp"},
|
||||
{"anchor": "mcp_discoverability.py:9", "expect": "EXPECTED_JENKINS_TOOLS"},
|
||||
{"anchor": "mcp_discoverability.py:17", "expect": "EXPECTED_GLITCHTIP_TOOLS"},
|
||||
|
||||
{"anchor": "sentry_incident_bridge.py:36", "expect": "SENTRY_AUTH_TOKEN"},
|
||||
{"anchor": "sentry_incident_bridge.py:190", "expect": "def resolve_token"},
|
||||
{"anchor": "sentry_incident_bridge.py:289", "expect": "Authorization"},
|
||||
{"anchor": "sentry_observability.py:55", "expect": "SENTRY_DSN"},
|
||||
|
||||
{"anchor": "master_parity_gate.py:168", "expect": "def capture_startup_parity"},
|
||||
{"anchor": "master_parity_gate.py:255", "expect": "mutation_safe"},
|
||||
{"anchor": "gitea_mcp_server.py:19105", "expect": "def gitea_assess_master_parity"},
|
||||
|
||||
{"anchor": "gitea_mcp_server.py:193", "expect": "ACTIVE_WORKTREE_ENV"},
|
||||
{"anchor": "gitea_mcp_server.py:194", "expect": "AUTHOR_WORKTREE_ENV"},
|
||||
{"anchor": "gitea_mcp_server.py:2351", "expect": "/tmp/gitea_issue_lock.json"},
|
||||
{"anchor": "gitea_mcp_server.py:10897", "expect": "def gitea_bootstrap_author_issue_worktree"},
|
||||
{"anchor": "mcp_server.py:13", "expect": "/tmp/mcp_server_stderr.log"},
|
||||
|
||||
{"anchor": "issue_lock_store.py:26", "expect": "DEFAULT_LOCK_DIR"},
|
||||
{"anchor": "issue_lock_store.py:83", "expect": "def session_pointer_path"},
|
||||
{"anchor": "issue_lock_store.py:98", "expect": "def is_process_alive"},
|
||||
{"anchor": "mcp_session_state.py:27", "expect": "DEFAULT_STATE_DIR"},
|
||||
{"anchor": "control_plane_db.py:47", "expect": "DEFAULT_DB_PATH"},
|
||||
{"anchor": "control_plane_db.py:380", "expect": "mode=0o700"},
|
||||
{"anchor": "control_plane_db.py:386", "expect": "sqlite3.connect"},
|
||||
{"anchor": "control_plane_db.py:1145", "expect": "os.getpid()"},
|
||||
{"anchor": "gitea_mcp_server.py:12804", "expect": "owner_pid_alive"}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,407 @@
|
||||
# Remote-MCP threat model, trust boundaries, and service decomposition
|
||||
|
||||
What the adversary is, what each boundary protects, and which services may share a process.
|
||||
|
||||
- **Issue:** #956 (Remote-MCP threat model), child of epic #929, cross-linked to #955.
|
||||
- **Depends on:** #930 (closed) — `docs/remote-mcp/coupling-inventory.md`.
|
||||
- **Blocks:** #932, #933, #934, #938.
|
||||
- **Generated against commit:** `a143cd065ba06e1a2bdc5143a19ec156e53650ef` (#931's transport
|
||||
bind seam). Originally generated against `aad5c8b42361d380a8eeb07b94b90815e594c2c5`
|
||||
(`master`) and re-anchored when #931 shifted the cited lines.
|
||||
- **Scope:** documentation only. This child changes no server behavior. It adds one
|
||||
document, one anchor fixture, and the test that enforces them.
|
||||
|
||||
## Relationship to #930
|
||||
|
||||
#930 asked *what breaks when the process stops being local*. This document asks *what an
|
||||
attacker gets, and where we stop them*. The two are deliberately different axes: #930
|
||||
classifies each coupling as portable, seam, replacement, or cannot-be-remote; this document
|
||||
classifies each **credential** by blast radius and each **boundary** by what crossing it
|
||||
requires. An entry can be perfectly portable and still be a trust disaster —
|
||||
`gitea_config.py:851` is portable Python that reads a CI secret from inside the Gitea server.
|
||||
|
||||
### Anchors are enforced, not asserted
|
||||
|
||||
Every `file:line` in this document is declared in `docs/remote-mcp/threat-model-anchors.json`
|
||||
with the substring that must appear at that line, and
|
||||
`tests/test_issue_956_threat_model.py` fails if any anchor does not resolve or if the
|
||||
document cites an anchor the fixture does not cover.
|
||||
|
||||
This guard exists because #930 did not have one. Its inventory was generated at
|
||||
`7bf4f125`; by `aad5c8b4` its `gitea_mcp_server.py` anchors had drifted — the transport
|
||||
bind it cited at line 23750 now lives at `gitea_mcp_server.py:24728`, and its
|
||||
client-managed provenance anchor at 14588 now lands in an unrelated function. Nothing
|
||||
failed, because nothing checked. Anchors into a ~24,700-line module rot silently, and a
|
||||
security document that cannot prove its own citations is worse than none, because it is
|
||||
trusted.
|
||||
|
||||
---
|
||||
|
||||
## 1. Assets
|
||||
|
||||
What an adversary wants. Ordered by consequence, not by likelihood.
|
||||
|
||||
| ID | Asset | Why it matters |
|
||||
| -- | ----- | -------------- |
|
||||
| A1 | Merge authority on `Scaled-Tech-Consulting/Gitea-Tools` | This repository *is* the control plane. Code merged here becomes the gate that authorizes every future mutation, so merge authority is self-amplifying: one merge can disable every other control in this document. |
|
||||
| A2 | Write authority on the `mdcps` tenant | A second, unrelated organization reachable from the same configuration. Compromise here is a cross-organization incident, not an internal one. |
|
||||
| A3 | The eight Gitea role credentials | Long-lived bearer tokens. Possession is authority; there is no second factor at the API. |
|
||||
| A4 | Jenkins read access (`mdcps`, enabled) | Build logs routinely carry deployment topology, internal hostnames, and accidentally-echoed secrets. |
|
||||
| A5 | Error-tracking read access (GlitchTip / Sentry) | Event payloads carry stack frames, request context, and production user data. |
|
||||
| A6 | Coordination-state integrity | The locks, leases, and review-decision records that make "exactly one owner" true. Corrupting them needs no Gitea credential and produces duplicate or lost work. |
|
||||
| A7 | The operator's checkout and worktrees | Unmerged code, branch state, and the filesystem the author tools write to. |
|
||||
| A8 | The macOS login keychain | The meta-credential. Everything in A3, A4, and A5 resolves from it. |
|
||||
| A9 | Separation of duty between review and merge | The property that no single actor both approves and lands a change. An *asset*, not a control, because it is what the controls exist to produce. |
|
||||
| A10 | Audit and provenance records | Determine whether an incident is reconstructable. An attacker who can forge provenance makes an intrusion indistinguishable from normal work. |
|
||||
|
||||
## 2. Adversaries
|
||||
|
||||
| ID | Adversary | Capability assumed | Not assumed |
|
||||
| -- | --------- | ------------------ | ----------- |
|
||||
| ADV1 | **Compromised LLM client** | Full control of one MCP client. Issues arbitrary tool calls, in any order, with any arguments, at machine speed. Sees every tool result. | Cannot read the operator's disk except through tools; cannot execute arbitrary local code outside the tool surface. |
|
||||
| ADV2 | **Prompt injection** via repository content | Controls text the model reads and treats as instruction — issue bodies, PR descriptions, review comments, commit messages, file contents. Reaches the model on any read of untrusted content. | Holds no credential and issues no call directly. Its entire power is causing an *authorized* client to act. |
|
||||
| ADV3 | **Malicious tool arguments** | Supplies hostile values to any parameter — paths, branch names, session identifiers, worktree paths, issue numbers — including traversal, injection, and confusion between look-alike identifiers. | Cannot bypass a gate that actually validates its input. |
|
||||
| ADV4 | **Network attacker** | Observes and modifies traffic between client, server, and Gitea. Attempts downgrade, replay, and endpoint impersonation. | Does not hold a valid credential at the start. |
|
||||
| ADV5 | **Curious operator** | Legitimate local access to the workstation: process table, `/tmp`, home directory, keychain prompts. Not malicious, but not authorized for every role either. | Does not defeat the OS keychain's own access control without a prompt. |
|
||||
|
||||
ADV2 is the adversary this architecture most under-models. Every other adversary must first
|
||||
obtain something. Prompt injection obtains nothing: it borrows authority the client already
|
||||
holds and is indistinguishable at the tool boundary from legitimate work. Each boundary
|
||||
below therefore states whether it constrains ADV2 at all — and most do not, because they
|
||||
authenticate the *caller*, not the *intent*.
|
||||
|
||||
## 3. Trust boundaries
|
||||
|
||||
"Crossing requires today" is what the code actually enforces at
|
||||
`a143cd065ba06e1a2bdc5143a19ec156e53650ef`, not what the design intends.
|
||||
|
||||
| ID | Boundary | Protects | Crossing requires today | Crossing must require remotely |
|
||||
| -- | -------- | -------- | ----------------------- | ------------------------------ |
|
||||
| B1 | LLM client ↔ MCP server session | A1, A3, A10 — that a mutating session was established through the sanctioned client path | A single configured bind (`gitea_mcp_server.py:24728`) validated against one closed allowlist (`mcp_daemon_guard.py:49`, `mcp_daemon_guard.py:195`) — since #931 the identifier comes from deployment configuration and defaults to the local transport, so the boundary no longer rests on a literal, but it still rests on the *bind* rather than on an authenticated caller; client-managed provenance (`gitea_mcp_server.py:15415`) or a refusal (`gitea_mcp_server.py:15453`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9132` and `gitea_mcp_server.py:9381`) | An authenticated handshake issuing a server-side session identity bound to a principal, with the transport recorded in provenance. The physical proof (a pipe) must become a cryptographic one. |
|
||||
| B2 | Role ↔ role | A9 — that author, reviewer, merger, and reconciler are distinct authorities | **The process boundary only.** The role is a property of the process, read once from `GITEA_MCP_PROFILE` (`gitea_config.py:54`). A caller gets author permissions by connecting to the author process. Review and merge are the operations singled out for extra care (`gitea_config.py:97`) | A per-request principal, so the role follows from the credential presented and cannot be selected by reaching a different endpoint. |
|
||||
| B3 | MCP server ↔ credential store | A3, A8 — that only sanctioned code turns a profile into a token | `_keychain_token` shelling out to the login keychain (`gitea_config.py:956`), dispatched by `resolve_token` (`gitea_config.py:974`) with the reference type built at `gitea_config.py:1015`, gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`). Inline secrets are rejected at config load (`gitea_config.py:294`) | A credential provider keyed by the *request* principal, returning only that principal's credential, with the source recorded and the value never returned. |
|
||||
| B4 | MCP server ↔ Gitea | A1, A2 — that only authorized calls reach the forge | A bearer token over TLS. Server-side, nothing distinguishes one role's token from another beyond the account it belongs to | Unchanged at the forge; the endpoint in front of it must refuse unauthenticated and plaintext connections before tool dispatch. |
|
||||
| B5 | MCP server ↔ caller's filesystem | A7 — that a tool acts on the *caller's* disk or refuses | Nothing. The server's disk *is* the caller's disk. Worktree bootstrap writes directly (`gitea_mcp_server.py:10897`); the active workspace is process-global (`gitea_mcp_server.py:193`, `gitea_mcp_server.py:194`) | An explicit per-tool classification, enforced at dispatch, refusing filesystem tools over a transport that cannot reach the caller's disk. A green verdict about the wrong disk is the failure to prevent. |
|
||||
| B6 | MCP server ↔ coordination state | A6, A9 — mutual exclusion | Local files and a local SQLite database, with liveness judged from the local process table (`issue_lock_store.py:98`), keyed on paths under one user's home (`issue_lock_store.py:26`, `mcp_session_state.py:27`, `control_plane_db.py:47`) and on `os.getpid()` (`control_plane_db.py:1145`, `gitea_mcp_server.py:12804`). A legacy global slot still exists at `gitea_mcp_server.py:2351`, and the session-pointer file is named per PID (`issue_lock_store.py:83`) | One authority per ownership question, with liveness from session identity and expiry, and atomic acquire, renew, and release across hosts. |
|
||||
| B7 | Gitea integration ↔ unrelated integrations | A4, A5 — that a Gitea compromise is not a CI and observability compromise | **Nothing.** See §5. The Gitea server reads Jenkins and GlitchTip secrets (`gitea_config.py:851`, reached from `gitea_config.py:837`) and holds the Sentry token (`sentry_incident_bridge.py:190`) | A hard process boundary. This is the boundary #956 exists to create. |
|
||||
| B8 | Tenant ↔ tenant (`prgs` / `mdcps` / `local-lab`) | A2 — that one organization's compromise is not another's | Convention. One configuration declares all three contexts; `resolve_service` fails closed on a *disabled* context (`gitea_config.py:704`) but the credentials of enabled ones remain reachable in-process. A per-profile repository scope exists (`gitea_config.py:499`) | Separate deployments, or at minimum per-tenant credential scopes with no process able to resolve both. |
|
||||
| B9 | Deployed code ↔ merged policy | A1, A10 — that the running server enforces the rules that were actually merged | Comparing this process's startup commit against this disk (`master_parity_gate.py:168`), conjoined into a single verdict (`master_parity_gate.py:255`) published by `gitea_mcp_server.py:19105` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. |
|
||||
|
||||
### What no boundary constrains
|
||||
|
||||
None of B1–B9 constrains **ADV2**. Every one authenticates a caller or a process; prompt
|
||||
injection supplies neither. An injected instruction that reaches an authorized author
|
||||
session crosses B1, B2, B3, and B5 legitimately, because at each of those boundaries it *is*
|
||||
the author. The only controls that bite ADV2 are those constraining what an authenticated
|
||||
principal may do regardless of what it asks for — the per-role permission split (B2), the
|
||||
repository scope at `gitea_config.py:499`, and separation of duty (A9). Sizing those
|
||||
controls correctly matters more after the migration, not less, because a remote endpoint
|
||||
raises the number of clients that can be injected into.
|
||||
|
||||
## 4. Data flows
|
||||
|
||||
Flows that cross a boundary. `==>` carries a credential; `-->` does not.
|
||||
|
||||
```
|
||||
B1 B4
|
||||
[LLM client] ====================> [MCP server] ========> [Gitea]
|
||||
^ stdio pipe today | ^ (A1,A2)
|
||||
| session identity | |
|
||||
| after migration | |
|
||||
| | | B3
|
||||
untrusted repository content | +======> [macOS login keychain] (A8)
|
||||
read back into the model (ADV2) | resolves A3, A4, A5
|
||||
^ |
|
||||
+----------------------------------+
|
||||
|
|
||||
B5 | B6
|
||||
[operator checkout / worktrees] <--------+-------> [locks · leases · sqlite]
|
||||
(A7) | (A6)
|
||||
|
|
||||
B7 <-- boundary does not exist today
|
||||
|
|
||||
+========================+========================+
|
||||
| | |
|
||||
[Jenkins] (A4) [GlitchTip] (A5) [Sentry] (A5)
|
||||
external MCP server external MCP server in-process bridge
|
||||
```
|
||||
|
||||
Two flows deserve attention because neither is obvious from the code:
|
||||
|
||||
1. **The keychain flow fans out.** B3 is drawn once but resolves credentials for *every*
|
||||
configured profile and service, not only the active one. `gitea_list_profiles`
|
||||
(`gitea_mcp_server.py:19261`) reports each profile's credential status by calling
|
||||
`resolve_token` on it (`gitea_mcp_server.py:19312`), and `gitea_audit_config`
|
||||
(`gitea_mcp_server.py:19555`) reports service credential status through
|
||||
`service_summaries` (`gitea_mcp_server.py:19577`).
|
||||
2. **The return path is a flow too.** Content read from Gitea travels back into the model
|
||||
and is treated as instruction. This is the ADV2 edge, and it is the only edge in the
|
||||
diagram with no authentication on it, because it is not a request.
|
||||
|
||||
## 5. Per-boundary credential inventory
|
||||
|
||||
**14 credentials in total.** Blast radius is stated as what the credential yields *on its
|
||||
own*, assuming every gate not backed by the credential itself has been bypassed — because
|
||||
an attacker holding a token calls the API, not our tools.
|
||||
|
||||
| ID | Credential | Holder | Boundary | Blast radius |
|
||||
| -- | ---------- | ------ | -------- | ------------ |
|
||||
| CR1 | `prgs-author` Gitea token — account `jcwalker3` | macOS keychain; resolved in-process (`gitea_config.py:974`) | B3 → B4 | Create branches, push, commit, open PRs, create/close/comment issues on the control-plane repo. Cannot approve or merge. The one credential whose identity is genuinely distinct. |
|
||||
| CR2 | `prgs-reviewer` Gitea token — account `sysadmin` | macOS keychain | B3 → B4 | Approve and request changes. **Shares one Gitea account with CR3, CR4, CR5.** |
|
||||
| CR3 | `prgs-merger` Gitea token — account `sysadmin` | macOS keychain | B3 → B4 | Merge to `master` — A1 in full. Same account as CR2. |
|
||||
| CR4 | `prgs-reconciler` Gitea token — account `sysadmin` | macOS keychain | B3 → B4 | Close PRs, delete branches, irrecoverable decision-lock recovery. Same account as CR2. |
|
||||
| CR5 | `prgs-controller` Gitea token — account `sysadmin` | macOS keychain | B3 → B4 | Same operation set as CR4. Same account as CR2. |
|
||||
| CR6 | `mdcps-author` Gitea token — account `913443` | macOS keychain | B3 → B4, B8 | Author operations on a second organization. **Shares one account with CR7 and CR8.** |
|
||||
| CR7 | `mdcps-reviewer` Gitea token — account `913443` | macOS keychain | B3 → B4, B8 | Approve and request changes on `mdcps`. Same account as CR6. |
|
||||
| CR8 | `mdcps-merger` Gitea token — account `913443` | macOS keychain | B3 → B4, B8 | Merge on `mdcps` — A2 in full. Same account as CR6. |
|
||||
| CR9 | MDCPS Jenkins read credential | macOS keychain, read from the Gitea server process (`gitea_config.py:851`) | B7 | Read CI jobs, builds, and logs (A4). Enabled today. |
|
||||
| CR10 | MDCPS GlitchTip read credential | macOS keychain, read from the Gitea server process (`gitea_config.py:851`) | B7 | Read error events and their payloads (A5). Enabled today. |
|
||||
| CR11 | `SENTRY_AUTH_TOKEN` | Process environment, read in-process (`sentry_incident_bridge.py:36`, `sentry_incident_bridge.py:190`), sent as a bearer header (`sentry_incident_bridge.py:289`) | B7 | Read and reconcile Sentry issues (A5). Not a keychain credential — an env var, so it is inherited by anything the process spawns. |
|
||||
| CR12 | `SENTRY_DSN` | Process environment (`sentry_observability.py:55`) | B7 | Write events into the observability project. Low read value, real forgery value: an attacker can inject fabricated events into the record (A10). |
|
||||
| CR13 | macOS login keychain access | The operator's login session; gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`) | B3, ADV5 | **Every other credential in this table except CR11 and CR12.** This is the aggregation point. |
|
||||
| CR14 | Coordination-store access (no secret) | Filesystem permissions — `control_plane_db.py:47`, created `0o700` (`control_plane_db.py:380`), opened with a local file lock (`control_plane_db.py:386`) | B6, ADV5 | Full read/write of locks, leases, and decision records (A6). **There is no credential here at all** — anything running as the operator can rewrite ownership. |
|
||||
|
||||
### Findings
|
||||
|
||||
**Finding 1 — Role separation is not credential separation.** Four `prgs` roles resolve to
|
||||
one Gitea account (`sysadmin`): reviewer, merger, reconciler, and controller. A stolen
|
||||
reviewer credential *is* a merger credential. A9 — separation of duty between approving and
|
||||
landing — is therefore enforced entirely by which local process a call reaches (B2), and not
|
||||
at all by the forge. It survives exactly as long as B2 does, and B2 is the boundary the
|
||||
migration dissolves.
|
||||
|
||||
**Finding 2 — The `mdcps` tenant has no role separation at all.** Author, reviewer, and
|
||||
merger all resolve to account `913443`. One credential can open a PR, approve it, and merge
|
||||
it. The in-process self-review check compares the authenticated username against the PR
|
||||
author and would refuse — but that check runs on our side of B4. It is not a property of
|
||||
the credential, and an attacker holding the token does not call our tools.
|
||||
|
||||
**Finding 3 — Any one role process can resolve every other role's credential.** This is not
|
||||
inferred; it is demonstrated by tool output. `gitea_list_profiles`
|
||||
(`gitea_mcp_server.py:19261`) called from the **author** session reports
|
||||
`identity_status: "credentials present"` for `prgs-merger`, `prgs-reviewer`,
|
||||
`prgs-reconciler`, and every `mdcps` profile, because it calls `resolve_token` on each one
|
||||
(`gitea_mcp_server.py:19312`). The author process does not merely *have access to* the
|
||||
merger's credential — it reads it to answer a status query. B2 is not a credential boundary
|
||||
in either direction.
|
||||
|
||||
**Finding 4 — The Gitea server reads CI and observability secrets.** `gitea_audit_config`
|
||||
(`gitea_mcp_server.py:19555`) reports `MDCPS Jenkins: enabled, read-only, authenticated`.
|
||||
That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19577`,
|
||||
defined at `gitea_config.py:837`), whose default check calls `_keychain_token` on the
|
||||
service's own keychain reference (`gitea_config.py:851`). Producing that one line requires
|
||||
the Gitea MCP server to read the Jenkins secret and the GlitchTip secret out of the
|
||||
keychain. B7 does not exist.
|
||||
|
||||
**Finding 5 — Jenkins and GlitchTip are already decomposed; the reach is residual.** Their
|
||||
tools live in separately registered servers, marked `external-mcp`
|
||||
(`gitea_mcp_server.py:17710`, `gitea_mcp_server.py:17716`, `gitea_mcp_server.py:17737`,
|
||||
`gitea_mcp_server.py:17742`) with their own expected tool sets (`mcp_discoverability.py:9`,
|
||||
`mcp_discoverability.py:17`). The correct decomposition was already chosen. What remains is
|
||||
a leak across it: the credential *references* still live in the Gitea configuration and are
|
||||
still resolved by the Gitea process. #75 bundled these services into one control-plane
|
||||
umbrella; the tools were separated afterwards, the credentials were not.
|
||||
|
||||
**Finding 6 — Sentry is the exception that is not decomposed.** Unlike Jenkins and
|
||||
GlitchTip, the Sentry bridge runs *inside* the Gitea server, resolving its token from the
|
||||
process environment (`sentry_incident_bridge.py:190`) and sending it as a bearer header
|
||||
(`sentry_incident_bridge.py:289`). Being an environment variable rather than a keychain item
|
||||
makes it strictly worse: it needs no keychain prompt and is inherited by every subprocess the
|
||||
server spawns — including the `ps` invocations at `gitea_mcp_server.py:21465` and
|
||||
`gitea_mcp_server.py:21509`, reached from `gitea_mcp_server.py:21445`.
|
||||
|
||||
**Finding 7 — The highest-value coordination asset has the weakest gate.** A6 is protected
|
||||
by filesystem permissions alone (CR14). Corrupting a lease requires no Gitea credential,
|
||||
produces no forge-side audit record, and breaks the mutual exclusion the entire workflow
|
||||
assumes. Every other asset costs an attacker a credential; this one costs nothing beyond
|
||||
local access, which is exactly ADV5's position.
|
||||
|
||||
**Finding 8 — Provenance authenticates the launch, not the caller.** `server_provenance` is
|
||||
reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19004`),
|
||||
derived from environment inspection (`gitea_mcp_server.py:15415`) with the recognized-key
|
||||
allowlist at `gitea_config.py:1172` and the generator that emits the marker at
|
||||
`gitea_config.py:1233`. Every one of those facts is fixed at process start. A client that is
|
||||
trustworthy at launch and compromised a minute later remains `client_managed` for the life
|
||||
of the process, and the transport contract that underwrites it is stated as a property of
|
||||
the server itself (`mcp_server.py:4`). Since #931 that contract names the configured
|
||||
transport rather than asserting stdio, but it is still fixed once, at bind, for the life of
|
||||
the process.
|
||||
|
||||
## 6. Decomposition ruling
|
||||
|
||||
This section is the ruling #956 requires. It is a decision, not a recommendation.
|
||||
|
||||
**D1 — No unrelated co-residency.** A single integration process **must not** hold, resolve,
|
||||
or be able to resolve credentials for services it does not itself integrate with.
|
||||
Concretely: the Gitea MCP service may hold Gitea credentials and nothing else. Jenkins,
|
||||
GlitchTip, Sentry, and any database credential are **not permitted** to co-reside with Gitea
|
||||
credentials in one process.
|
||||
|
||||
*Rationale.* A process is the smallest unit an attacker takes whole. Once ADV1 or ADV2
|
||||
controls execution in a process, every credential that process can resolve is theirs, and no
|
||||
in-process check helps, because the checks are in the process too. Blast radius is therefore
|
||||
a property of the process boundary and nothing finer. Findings 4 and 6 show that today one
|
||||
compromise of the Gitea server yields CI read access, error-tracking read access, and — via
|
||||
CR13 — every role credential on both tenants. That is the single largest reduction in blast
|
||||
radius available anywhere in epic #929, and it costs no new mechanism: the decomposition
|
||||
already exists (Finding 5) and is merely leaked across.
|
||||
|
||||
**D2 — Separation of duty must be backed by credentials.** Two roles whose separation is a
|
||||
security property must not resolve to the same forge account. Specifically, reviewer and
|
||||
merger must be distinct accounts. Today they are not, on either tenant (Findings 1 and 2).
|
||||
|
||||
*Rationale.* B2 is a process boundary, and the migration's entire purpose is to replace
|
||||
process boundaries with request-level ones. A separation enforced only by which process a
|
||||
call reaches does not survive that replacement — and it is already bypassable by anyone who
|
||||
holds the token and calls the API instead of the tool.
|
||||
|
||||
**D3 — Credential resolution is scoped to the request principal.** A session must resolve its
|
||||
own credential and must have no path to any other principal's. The resolve-every-profile
|
||||
behavior behind `gitea_mcp_server.py:19312` and `gitea_mcp_server.py:19577` must report
|
||||
configured-or-not from configuration alone, without resolving the secret.
|
||||
|
||||
*Rationale.* Finding 3. An audit surface that proves a credential exists by fetching it is a
|
||||
credential-aggregation primitive wearing a diagnostic's clothes.
|
||||
|
||||
**D4 — Coordination state is a protected asset with its own authority.** Access to locks,
|
||||
leases, and decision records must require an authenticated session, not merely local
|
||||
filesystem access.
|
||||
|
||||
*Rationale.* Finding 7. #937 already moves this store for concurrency reasons; the
|
||||
authorization requirement must land with it, or the store becomes remotely reachable while
|
||||
still being authorized by nothing.
|
||||
|
||||
### Exceptions
|
||||
|
||||
**One, time-boxed.** During the dual-run window defined by #939, the **local** stdio fleet
|
||||
may continue to resolve Jenkins and GlitchTip credential *references* from the shared
|
||||
configuration, because removing them from the local configuration is not a prerequisite for
|
||||
standing up the remote endpoint and would strand the operator's existing local workflow.
|
||||
|
||||
This exception is bounded by all of:
|
||||
|
||||
- It applies to the local stdio deployment only. The remote endpoint (#938) must be
|
||||
configured with Gitea credentials and no others from its first day.
|
||||
- It expires when #939 completes. It does not survive cutover.
|
||||
- It does not extend to Sentry: CR11 and CR12 are process-environment credentials in the
|
||||
Gitea server (Finding 6) and must be absent from the remote deployment's environment
|
||||
regardless of dual-run state.
|
||||
|
||||
No exception is granted to D2, D3, or D4.
|
||||
|
||||
### Consequences for the target architecture
|
||||
|
||||
- The remote endpoint serves **Gitea only**. It is not a general control-plane endpoint.
|
||||
- Jenkins and GlitchTip keep their existing separate servers, and their credential
|
||||
references move out of the Gitea configuration.
|
||||
- The Sentry bridge either moves behind its own service boundary or is absent from the
|
||||
remote deployment. It does not travel with the Gitea server.
|
||||
- Reviewer and merger accounts diverge before the endpoint is trusted for merges, or A9 is
|
||||
recorded as unenforced.
|
||||
|
||||
## 7. Child-to-boundary mapping
|
||||
|
||||
Every #929 child from 2 through 10, mapped to the boundary it implements. A child
|
||||
implementing more than one boundary names its primary first.
|
||||
|
||||
| Child | Issue | Boundaries | What it must establish | Rulings it must honor |
|
||||
| ----: | ----- | ---------- | ---------------------- | --------------------- |
|
||||
| 2 | #931 | B1, B9 | The bound transport becomes a validated value that provenance and freshness can both key on. Without it neither B1 nor B9 has an input. | — |
|
||||
| 3 | #932 | B2 | The role becomes a property of the request, not the process — the boundary the migration otherwise deletes. | D2, D3 |
|
||||
| 4 | #933 | B3, B7 | Credentials come from a provider keyed by principal. This is where D1 and D3 are either enforced or permanently lost. | D1, D3 |
|
||||
| 5 | #934 | B1 | Session provenance replaces pipe-and-process-table proof with an authenticated session identity. | — |
|
||||
| 6 | #935 | B9 | Freshness redefined against deployed build identity, with an explicit undeterminable verdict. | — |
|
||||
| 7 | #936 | B5 | Every tool classified and the filesystem boundary enforced at dispatch, so a tool cannot return green about the wrong disk. | — |
|
||||
| 8 | #937 | B6 | One authority per ownership question, with session-identity liveness and atomic transitions. | D4 |
|
||||
| 9 | #938 | B4, B1, B8 | The endpoint: authentication, principal binding, transport security, and — critically — the deployed credential set. | D1, D2, D3 |
|
||||
| 10 | #939 | B6 | Dual-run with exactly one coordination authority at every instant, and the rollback that proves the way back. | D1 exception expiry |
|
||||
|
||||
Boundary coverage: B1 (#931, #934, #938), B2 (#932), B3 (#933), B4 (#938), B5 (#936),
|
||||
B6 (#937, #939), B7 (#933), B8 (#938), B9 (#931, #935).
|
||||
|
||||
B7 has exactly one owner, #933, and that is deliberate. B7 is not created by standing up an
|
||||
endpoint; it is created by deciding which credentials a process may resolve, which is
|
||||
precisely what the credential-provider child does.
|
||||
|
||||
## 8. Adversarial walkthrough: a fully compromised LLM client
|
||||
|
||||
The required negative case. ADV1 holds one MCP client and issues arbitrary tool calls. What
|
||||
does it reach at each boundary?
|
||||
|
||||
### Before the migration
|
||||
|
||||
The client is attached to the local fleet over stdio.
|
||||
|
||||
| Boundary | What ADV1 reaches | Stopped by |
|
||||
| -------- | ----------------- | ---------- |
|
||||
| B1 | Everything the fleet serves. The client *is* the sanctioned launcher: it satisfies the client-managed check (`gitea_mcp_server.py:15415`) by construction, and provenance is never re-verified after launch (Finding 8). | Nothing. The guard authenticates the launch, not the caller. |
|
||||
| B2 | All five roles — it is attached to all five namespaces. It can author a PR, approve it from the reviewer namespace, and merge it from the merger namespace. | Only the in-process self-review check, which compares `jcwalker3` (author) against `sysadmin` (reviewer) and **passes**, because Finding 1 made them different accounts while leaving reviewer and merger identical. A9 falls in one sequence of legitimate calls. |
|
||||
| B3 | Every credential in CR1–CR10 via CR13, with no additional prompt — the daemon is already sanctioned, so `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`) returns immediately. | Nothing. |
|
||||
| B4 | A1 and A2 in full. | Branch protection at the forge, to the extent configured. |
|
||||
| B5 | The operator's checkout and every worktree, through the author tools (`gitea_mcp_server.py:10897`), plus the shared stderr path at `mcp_server.py:13`. | Nothing; the server's disk is the target disk. |
|
||||
| B6 | All coordination state — no credential required (CR14). It can forge lease ownership and clear decision locks. | Filesystem permissions, which it already satisfies. |
|
||||
| B7 | Jenkins (A4) and GlitchTip (A5) secrets via Finding 4, and CR11/CR12 from its own environment. | Nothing. |
|
||||
| B8 | Both tenants. | Nothing in-process; only the disabled-context check (`gitea_config.py:704`), which does not apply to enabled contexts. |
|
||||
| B9 | Can it run stale code? Only by advancing the checkout, which parity detects (`master_parity_gate.py:255`). | This gate holds — the one boundary that behaves as designed against ADV1. |
|
||||
|
||||
**Result: total.** One compromised client yields A1 through A8 and A10. The only asset with
|
||||
real resistance is A1 via branch protection, and the client holds the merger credential
|
||||
anyway. Nine boundaries, one meaningful stop.
|
||||
|
||||
### After the migration
|
||||
|
||||
The same client authenticates to the remote endpoint with one role's credential, assuming
|
||||
#931–#939 land **and honor D1–D4**.
|
||||
|
||||
| Boundary | What ADV1 reaches | Stopped by |
|
||||
| -------- | ----------------- | ---------- |
|
||||
| B1 | One authenticated session, bound to one principal. | #934: a forged or expired session identity is refused; the client cannot mint one. |
|
||||
| B2 | **One role.** Presenting the author credential yields author permissions only. | #932: the principal comes from the credential, not from which endpoint was reached. |
|
||||
| B3 | **One credential — its own.** | #933 with D3: the provider resolves by principal, and no diagnostic resolves the others. |
|
||||
| B4 | That role's authority on the forge. | Endpoint authentication (#938); plaintext and unauthenticated attempts refused before dispatch. |
|
||||
| B5 | **Nothing.** Filesystem tools are refused over the remote transport with a named blocker. | #936. |
|
||||
| B6 | Its own leases; contention resolves to exactly one winner. | #937 with D4: authenticated session required, not filesystem access. |
|
||||
| B7 | **Nothing.** No CI or observability credential exists in the process. | D1 — the single largest reduction on this table. |
|
||||
| B8 | One tenant. | D1 and #938: the deployment carries one tenant's credentials. |
|
||||
| B9 | Cannot induce stale enforcement. | #935: explicit fail-closed verdict, including undeterminable. |
|
||||
|
||||
**Result: bounded.** The compromise is contained to one role on one tenant, with no
|
||||
filesystem reach and no lateral credential access. A9 survives *only if D2 lands* — if
|
||||
reviewer and merger still share `sysadmin`, a compromised reviewer session still merges, and
|
||||
this row reads the same after the migration as before it.
|
||||
|
||||
### What the migration does not fix
|
||||
|
||||
Against **ADV2**, both tables are identical. Prompt injection does not need to cross a
|
||||
boundary: it arrives inside an authorized session and asks that session to do what it is
|
||||
already permitted to do. Every "stopped by" above authenticates a principal, and the
|
||||
injected instruction has the correct principal. The migration reduces ADV1's blast radius by
|
||||
roughly an order of magnitude and reduces ADV2's by nothing.
|
||||
|
||||
The controls that do constrain ADV2 are per-principal permission scope (#932), repository
|
||||
scope (`gitea_config.py:499`), and credential-backed separation of duty (D2) — each limiting
|
||||
what an authenticated session may do *regardless of what it is asked for*. #955's
|
||||
secure-isolation end state should be read with that distinction in mind: removing credentials
|
||||
from clients defeats ADV1 and ADV5, and does not by itself defeat ADV2.
|
||||
|
||||
Two further items are explicitly out of scope here and unowned by #929:
|
||||
|
||||
- **Session-credential rotation and revocation.** #938 names rotation as documentation, but
|
||||
no child owns proving that a revoked credential stops an in-flight session.
|
||||
- **ADV3** (malicious tool arguments) is diffused across every child rather than owned. The
|
||||
per-request principal work in #932 is the natural place to assert that identifiers taken
|
||||
from the request never authorize anything on their own.
|
||||
|
||||
## 9. How to verify this document
|
||||
|
||||
1. `PYTHONPATH=. pytest tests/test_issue_956_threat_model.py` — resolves every anchor
|
||||
against the working tree and checks the document's structural obligations.
|
||||
2. Pick any five anchors at random and read them; the fixture states what each line must
|
||||
contain.
|
||||
3. Reproduce Findings 3 and 4 live: call `gitea_list_profiles` and `gitea_audit_config`
|
||||
from the **author** namespace. Credential presence reported for roles other than the
|
||||
active one is Finding 3; `MDCPS Jenkins: enabled, read-only, authenticated` is Finding 4.
|
||||
|
||||
If the anchor test fails after an unrelated refactor, the anchors moved and the fixture
|
||||
needs regenerating — the claims are still true, but they are no longer traceable, which
|
||||
#956 treats as the same defect.
|
||||
+17
-3
@@ -1,7 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Gitea MCP Server — exposes Gitea operations as MCP tools.
|
||||
|
||||
Runs over stdio. All tools authenticate via macOS keychain (git credential fill).
|
||||
The transport is selected by deployment configuration (GITEA_MCP_TRANSPORT) and
|
||||
defaults to the local client-spawned transport when unset (#931); the permitted
|
||||
set lives in mcp_transport_config. All tools authenticate via macOS keychain
|
||||
(git credential fill).
|
||||
|
||||
Usage (standalone test):
|
||||
python3 mcp_server.py
|
||||
@@ -24718,7 +24721,11 @@ if __name__ == "__main__":
|
||||
# basename-only stack frames, and import-only launch cannot reconstruct
|
||||
# native transport; offline imports / standalone scripts fail closed.
|
||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||
mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
|
||||
# #931: the transport is no longer a literal here. It comes from deployment
|
||||
# configuration (GITEA_MCP_TRANSPORT), defaults to stdio when unset, and is
|
||||
# validated against the single permitted set in mcp_transport_config. An
|
||||
# unregistered identifier raises here, before any tool can dispatch.
|
||||
mcp_daemon_guard.bind_native_mcp_transport()
|
||||
# Lock this session's launch profile into the environment so child CLI
|
||||
# processes (e.g. review_pr.py) can detect and refuse profile
|
||||
# side-channel overrides (#199).
|
||||
@@ -24736,4 +24743,11 @@ if __name__ == "__main__":
|
||||
sys.stderr.write(
|
||||
f"--- Sentry observability: {_sentry_status.get('reason')} ---\n"
|
||||
)
|
||||
mcp.run(transport="stdio")
|
||||
# #931: serve over exactly the transport that was bound and pinned, and only
|
||||
# if this entrypoint is commissioned to execute it. Recognition is not
|
||||
# execution authorization (review 635): a registered remote identifier is
|
||||
# bound, pinned and recorded, but serving it would start a listener with no
|
||||
# authentication or per-request principal, which #938 owns. An absent bind
|
||||
# keeps its pre-existing #695 failure; an uncommissioned transport fails
|
||||
# closed here, before any listener exists and before any tool can dispatch.
|
||||
mcp.run(transport=mcp_daemon_guard.authorize_transport_execution("tool service"))
|
||||
|
||||
@@ -500,6 +500,11 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
|
||||
native = mcp_daemon_guard.is_native_mcp_transport()
|
||||
pytest = mcp_daemon_guard.is_pytest_runtime()
|
||||
production = mcp_daemon_guard.is_production_native_mcp_transport()
|
||||
# #931: report which transport underwrites the verdict, read through the
|
||||
# one shared accessor rather than assumed to be stdio. The gate's decision
|
||||
# is unchanged here; naming the transport is what lets #932 re-derive the
|
||||
# guarantee from an authenticated session instead of from the bind.
|
||||
bound = mcp_daemon_guard.bound_transport()
|
||||
if not native and not pytest:
|
||||
reasons.append(
|
||||
"irrecoverable provenance authorization requires production native "
|
||||
@@ -510,6 +515,7 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
|
||||
"allowed": not reasons,
|
||||
"native_mcp_transport": native,
|
||||
"production_native_mcp_transport": production,
|
||||
"bound_transport": bound,
|
||||
"pytest": pytest,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
+176
-13
@@ -32,6 +32,8 @@ import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import mcp_transport_config
|
||||
|
||||
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
|
||||
ALLOW_DIRECT_IMPORT_ENV = "GITEA_ALLOW_DIRECT_MCP_IMPORT"
|
||||
ALLOW_KEYCHAIN_CLI_ENV = "GITEA_ALLOW_KEYCHAIN_CLI"
|
||||
@@ -42,7 +44,9 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
|
||||
_NATIVE_RUNTIME: dict[str, Any] | None = None
|
||||
|
||||
# Production transport identifiers accepted by bind_native_mcp_transport.
|
||||
_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
|
||||
# #931: the permitted set is defined once, in mcp_transport_config. This name
|
||||
# is kept as an alias so the guard never restates a transport identifier.
|
||||
_PRODUCTION_TRANSPORTS = mcp_transport_config.SUPPORTED_TRANSPORTS
|
||||
_RUNTIME_MODE_PRODUCTION = "production"
|
||||
_RUNTIME_MODE_TEST = "test"
|
||||
_PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed"
|
||||
@@ -58,6 +62,23 @@ class UnsanctionedRuntimeError(RuntimeError):
|
||||
"""Raised when mutation/credential code runs outside a native MCP daemon."""
|
||||
|
||||
|
||||
class TransportExecutionError(UnsanctionedRuntimeError):
|
||||
"""Raised when a bound transport may not be served by this entrypoint (#931).
|
||||
|
||||
Subclasses :class:`UnsanctionedRuntimeError` so every existing fail-closed
|
||||
handler still catches it, while letting a caller that cares distinguish
|
||||
"nothing is bound" from "something valid is bound but its listener has not
|
||||
been commissioned". Carries the structured verdict on ``.assessment``.
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, assessment: dict[str, Any] | None = None):
|
||||
super().__init__(message)
|
||||
self.assessment = assessment or {}
|
||||
self.blocker_kind = self.assessment.get("blocker_kind")
|
||||
self.owner_issue = self.assessment.get("owner_issue")
|
||||
self.transport = self.assessment.get("transport")
|
||||
|
||||
|
||||
def is_pytest_runtime() -> bool:
|
||||
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
||||
"1",
|
||||
@@ -171,23 +192,46 @@ def mark_sanctioned_daemon() -> dict[str, Any]:
|
||||
return native_runtime_status()
|
||||
|
||||
|
||||
def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
|
||||
"""Bind the live native MCP transport lifecycle (#695).
|
||||
def bind_native_mcp_transport(*, transport: str | None = None) -> dict[str, Any]:
|
||||
"""Bind the live native MCP transport lifecycle (#695 / #931).
|
||||
|
||||
Must be called from the resolved canonical entrypoint immediately before
|
||||
the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
|
||||
Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
|
||||
process. Import-only or offline launch without this bind leaves
|
||||
the real MCP server transport loop (``mcp.run``). Requires a prior
|
||||
successful :func:`mark_sanctioned_daemon` claim in this process.
|
||||
Import-only or offline launch without this bind leaves
|
||||
:func:`is_native_mcp_transport` false.
|
||||
|
||||
#931: ``transport`` is now optional. Omitting it — which is what the
|
||||
production entrypoint does — resolves the identifier from deployment
|
||||
configuration via :func:`mcp_transport_config.resolve_configured_transport`,
|
||||
yielding :data:`mcp_transport_config.DEFAULT_TRANSPORT` when nothing is
|
||||
configured. An explicit argument remains supported for tests and for a
|
||||
launcher that has already resolved the value. Either way the identifier is
|
||||
validated against the single permitted set before the runtime record is
|
||||
written, so no tool can dispatch over an unregistered transport.
|
||||
|
||||
The resolved value is pinned into the process-local record and is read back
|
||||
only through :func:`bound_transport`. Rebinding to a different transport is
|
||||
refused, so two guards can never observe different values in one process.
|
||||
"""
|
||||
global _NATIVE_RUNTIME
|
||||
transport_name = (transport or "").strip().lower()
|
||||
if transport_name not in _PRODUCTION_TRANSPORTS:
|
||||
raise UnsanctionedRuntimeError(
|
||||
f"bind_native_mcp_transport rejected: transport {transport!r} is "
|
||||
f"not a production MCP transport (#695). Allowed: "
|
||||
f"{sorted(_PRODUCTION_TRANSPORTS)}."
|
||||
)
|
||||
if transport is None:
|
||||
resolution = mcp_transport_config.resolve_configured_transport()
|
||||
transport_name = str(resolution["transport"])
|
||||
if not resolution["supported"]:
|
||||
raise UnsanctionedRuntimeError(
|
||||
"bind_native_mcp_transport rejected: "
|
||||
+ "; ".join(resolution["reasons"])
|
||||
+ " No tool is served over an unregistered transport."
|
||||
)
|
||||
else:
|
||||
transport_name = mcp_transport_config.normalize_transport(transport)
|
||||
if transport_name not in _PRODUCTION_TRANSPORTS:
|
||||
raise UnsanctionedRuntimeError(
|
||||
f"bind_native_mcp_transport rejected: transport {transport!r} is "
|
||||
f"not a production MCP transport (#695). Allowed: "
|
||||
f"{sorted(_PRODUCTION_TRANSPORTS)}."
|
||||
)
|
||||
|
||||
entrypoint_path = _caller_official_entrypoint_path()
|
||||
if entrypoint_path is None:
|
||||
@@ -216,6 +260,23 @@ def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
|
||||
"between mark and bind (#695)."
|
||||
)
|
||||
|
||||
# #931: one process binds one transport. Re-binding the same identifier is
|
||||
# idempotent (a retried launch step must not fail); re-binding a different
|
||||
# one is refused, because a guard that already read the first value would
|
||||
# otherwise disagree with a guard that reads the second.
|
||||
already_bound = (_NATIVE_RUNTIME.get("transport") or "").strip()
|
||||
if (
|
||||
already_bound
|
||||
and _NATIVE_RUNTIME.get("phase") == _PHASE_TRANSPORT_BOUND
|
||||
and already_bound != transport_name
|
||||
):
|
||||
raise UnsanctionedRuntimeError(
|
||||
"bind_native_mcp_transport rejected: transport is already bound to "
|
||||
f"{already_bound!r} in this process; rebinding to "
|
||||
f"{transport_name!r} is forbidden (#931). Restart the server to "
|
||||
"change the deployment transport."
|
||||
)
|
||||
|
||||
# Pin session-state root for this server lifetime (#695 AC2 / PR #701).
|
||||
# Changing GITEA_MCP_SESSION_STATE_DIR after bind must not manufacture a
|
||||
# second authority domain for decision locks / workflow proofs.
|
||||
@@ -353,6 +414,88 @@ def is_production_native_mcp_transport() -> bool:
|
||||
return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION
|
||||
|
||||
|
||||
def bound_transport() -> str | None:
|
||||
"""The one authoritative bound transport identifier, or ``None`` (#931).
|
||||
|
||||
This is the shared accessor every transport-aware guard reads. It reports
|
||||
the value pinned at bind time, never the environment, so changing
|
||||
``GITEA_MCP_TRANSPORT`` after the bind cannot move what a guard observes —
|
||||
the same rule :func:`pinned_session_state_dir` applies to session state.
|
||||
|
||||
``None`` means unbound: an offline import or a launch that never reached
|
||||
the bind. Callers must treat that as fail-closed, exactly as they already
|
||||
treat :func:`is_native_mcp_transport` returning false.
|
||||
"""
|
||||
if not is_native_mcp_transport():
|
||||
return None
|
||||
return (_NATIVE_RUNTIME or {}).get("transport") or None
|
||||
|
||||
|
||||
def assert_transport_bound(context: str = "tool service") -> str:
|
||||
"""Return the bound transport, or fail closed before *context* (#931).
|
||||
|
||||
Called immediately before the server enters its transport loop so an
|
||||
invalid or absent bind stops the process rather than serving tools over a
|
||||
transport no guard can name.
|
||||
"""
|
||||
transport = bound_transport()
|
||||
if transport:
|
||||
return transport
|
||||
raise UnsanctionedRuntimeError(
|
||||
f"No MCP transport is bound; refusing {context} (#931). "
|
||||
"bind_native_mcp_transport must succeed from the canonical entrypoint "
|
||||
"before any tool is served. Offline import and standalone launch "
|
||||
"cannot reconstruct a bind."
|
||||
)
|
||||
|
||||
|
||||
def assess_serve_authorization() -> dict[str, Any]:
|
||||
"""Structured verdict on whether this process may serve tools (#931).
|
||||
|
||||
This is the decision that consumes :func:`bound_transport`. It is what stops
|
||||
the bound identifier from being reporting-only metadata: the serve path
|
||||
cannot proceed unless the value pinned at bind is one this entrypoint is
|
||||
commissioned to execute.
|
||||
|
||||
Never raises; returns the verdict so callers and diagnostics can inspect it.
|
||||
"""
|
||||
return mcp_transport_config.assess_transport_execution(bound_transport())
|
||||
|
||||
|
||||
def authorize_transport_execution(context: str = "tool service") -> str:
|
||||
"""Return the transport this process may serve, or fail closed (#931).
|
||||
|
||||
Two distinct boundaries, in order:
|
||||
|
||||
1. **Bind presence** — :func:`assert_transport_bound` enforces the
|
||||
pre-existing #695 contract, so an unbound runtime keeps its established
|
||||
failure and reason code.
|
||||
2. **Execution authorization** — the bound identifier must be one this
|
||||
entrypoint is commissioned to serve. A registered transport whose
|
||||
listener has not been commissioned is refused here, before any listener
|
||||
is created and before any tool can dispatch.
|
||||
|
||||
That ordering matters: recognition, validation and durable recording all
|
||||
still happen for a remote identifier, so #931's seam is intact; only the act
|
||||
of *serving* it is withheld until its owning issue commissions it.
|
||||
"""
|
||||
# Boundary 1: unbound stays exactly as fail-closed as it was under #695.
|
||||
assert_transport_bound(context)
|
||||
|
||||
# Boundary 2: bound, but is this entrypoint allowed to serve it?
|
||||
assessment = assess_serve_authorization()
|
||||
if assessment.get("allowed"):
|
||||
return str(assessment["transport"])
|
||||
|
||||
reasons = "; ".join(assessment.get("reasons") or []) or "not authorized"
|
||||
next_action = assessment.get("exact_next_action") or ""
|
||||
raise TransportExecutionError(
|
||||
f"Refusing {context} (#931) [{assessment.get('blocker_kind')}]: "
|
||||
f"{reasons} {next_action}".strip(),
|
||||
assessment,
|
||||
)
|
||||
|
||||
|
||||
def is_sanctioned_mcp_daemon() -> bool:
|
||||
"""Backward-compatible name; #695 requires native transport, not env alone."""
|
||||
if is_production_native_mcp_transport():
|
||||
@@ -476,6 +619,21 @@ def native_runtime_status() -> dict[str, Any]:
|
||||
"entrypoint_path": rt.get("entrypoint_path"),
|
||||
"phase": rt.get("phase"),
|
||||
"transport": rt.get("transport"),
|
||||
# #931: the authoritative bound identifier, plus the seam that defines
|
||||
# what may be bound. ``bound_transport`` is None until a bind succeeds,
|
||||
# so an offline import is distinguishable from a stdio session.
|
||||
"bound_transport": bound_transport(),
|
||||
"transport_bound": bound_transport() is not None,
|
||||
"default_transport": mcp_transport_config.DEFAULT_TRANSPORT,
|
||||
"supported_transports": list(mcp_transport_config.supported_transports()),
|
||||
"transport_env": mcp_transport_config.TRANSPORT_ENV,
|
||||
# #931 review 635: recognition and execution authorization are distinct.
|
||||
# ``supported`` is what may be bound; ``executable`` is what this
|
||||
# entrypoint may actually serve. A recognized-but-uncommissioned
|
||||
# transport reports serve_authorized False with a named blocker.
|
||||
"executable_transports": list(mcp_transport_config.executable_transports()),
|
||||
"serve_authorized": bool(assess_serve_authorization().get("allowed")),
|
||||
"serve_authorization": assess_serve_authorization(),
|
||||
"mode": rt.get("mode"),
|
||||
"session_state_dir": pinned_session_state_dir() or rt.get("session_state_dir"),
|
||||
"session_state_dir_pinned": pinned_session_state_dir() is not None,
|
||||
@@ -502,7 +660,12 @@ def mutation_provenance_fields() -> dict[str, Any]:
|
||||
if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
|
||||
transport = "test_native_mcp"
|
||||
return {
|
||||
# ``transport`` stays the trust *class* it has always been, so existing
|
||||
# durable records keep their shape. ``bound_transport`` (#931) adds the
|
||||
# bound identifier itself, which is what lets an operator tell from a
|
||||
# durable record which transport performed a mutation.
|
||||
"transport": transport,
|
||||
"bound_transport": st.get("bound_transport"),
|
||||
"native_mcp_transport": bool(st["native_mcp_transport"]),
|
||||
"production_native_mcp_transport": bool(
|
||||
st.get("production_native_mcp_transport")
|
||||
|
||||
+7
-3
@@ -1,7 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Gitea MCP Server — exposes Gitea operations as MCP tools.
|
||||
|
||||
Runs over stdio. All tools authenticate via macOS keychain (git credential fill).
|
||||
The transport is selected by deployment configuration (GITEA_MCP_TRANSPORT) and
|
||||
defaults to the local client-spawned transport when unset (#931); the permitted
|
||||
set lives in mcp_transport_config. All tools authenticate via macOS keychain
|
||||
(git credential fill).
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
@@ -43,8 +46,9 @@ check_conflict_markers()
|
||||
|
||||
# #558 / #695: claim the official entrypoint before loading mutation modules.
|
||||
# This alone does NOT authorize mutations — gitea_mcp_server binds the live
|
||||
# native MCP transport (stdio) immediately before mcp.run. Import-only or
|
||||
# offline launch without that bind fails closed on mutations.
|
||||
# native MCP transport immediately before mcp.run, over the configured
|
||||
# transport (#931). Import-only or offline launch without that bind fails
|
||||
# closed on mutations.
|
||||
try:
|
||||
import mcp_daemon_guard
|
||||
|
||||
|
||||
@@ -588,9 +588,13 @@ def save_state(
|
||||
bool(prov.get("production_native_mcp_transport")),
|
||||
)
|
||||
body.setdefault("transport", prov.get("transport"))
|
||||
# #931: record the bound transport identifier itself, so a durable
|
||||
# decision lock names which transport performed the mutation.
|
||||
body.setdefault("bound_transport", prov.get("bound_transport"))
|
||||
except Exception:
|
||||
body.setdefault("native_mcp_transport", False)
|
||||
body.setdefault("transport", "untrusted")
|
||||
body.setdefault("bound_transport", None)
|
||||
|
||||
envelope = {
|
||||
"kind": kind,
|
||||
|
||||
@@ -0,0 +1,292 @@
|
||||
"""Single authoritative source for the bound MCP transport identifier (#931).
|
||||
|
||||
Before this module the transport was a literal, passed once at the bottom of
|
||||
``gitea_mcp_server`` as ``bind_native_mcp_transport(transport="stdio")``. Every
|
||||
guard that later asks "is this a trusted native session" resolves that question
|
||||
through the value bound there, so the literal was effectively a constant in the
|
||||
authorization chain rather than configuration.
|
||||
|
||||
This module is the seam. It owns three things and nothing else:
|
||||
|
||||
- the permitted set of transport identifiers,
|
||||
- the default used when deployment configuration says nothing,
|
||||
- the resolution of the configured value into a validated identifier.
|
||||
|
||||
It deliberately holds no state. The *bound* transport is pinned once, at bind
|
||||
time, into the process-local native-runtime record owned by
|
||||
:mod:`mcp_daemon_guard`, and is read back through
|
||||
``mcp_daemon_guard.bound_transport()``. That split matters: configuration is
|
||||
read exactly once, before any tool can dispatch, so a later environment change
|
||||
cannot move the value a guard observes — the same pinning rule already applied
|
||||
to the session-state root under #695 AC2.
|
||||
|
||||
Nothing here consumes tool arguments, request bodies, or provenance fields. The
|
||||
only input is the deployment environment, read at bind time.
|
||||
|
||||
Standing up a listener for a non-stdio transport is #938; this module only
|
||||
makes the identifier expressible and validated.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any, Mapping
|
||||
|
||||
# Deployment configuration key. Read once, at bind time, and never again.
|
||||
TRANSPORT_ENV = "GITEA_MCP_TRANSPORT"
|
||||
|
||||
# The local, client-spawned transport. Unset configuration resolves to this,
|
||||
# which is what keeps every existing stdio deployment byte-identical.
|
||||
DEFAULT_TRANSPORT = "stdio"
|
||||
|
||||
# The sanctioned remote transport identifier. Accepting it here is what makes
|
||||
# the bind pluggable; the endpoint that serves it belongs to #938. The name
|
||||
# matches the MCP transport name so no second vocabulary has to be mapped.
|
||||
REMOTE_TRANSPORT = "streamable-http"
|
||||
|
||||
# The permitted set. This is the only place transport identifiers are
|
||||
# enumerated; guards consult it rather than restating any member.
|
||||
#
|
||||
# ``sse`` is a real MCP transport and is deliberately absent: it is the
|
||||
# superseded remote transport, and admitting it would give the deployment two
|
||||
# remote paths to reason about. An unregistered identifier must fail closed at
|
||||
# bind time, and ``sse`` is held to that rule like any other.
|
||||
SUPPORTED_TRANSPORTS = frozenset({DEFAULT_TRANSPORT, REMOTE_TRANSPORT})
|
||||
|
||||
# Recognition is not execution authorization (#931, review 635 B1/B2).
|
||||
#
|
||||
# SUPPORTED_TRANSPORTS answers "is this an identifier this system knows, and may
|
||||
# it be bound, pinned and recorded?". It deliberately includes the remote
|
||||
# identifier, because #931 requires the bind to become pluggable.
|
||||
#
|
||||
# EXECUTABLE_TRANSPORTS answers a strictly narrower question: "is this entrypoint
|
||||
# commissioned to actually *serve* on that transport?". Only the local transport
|
||||
# is. Handing ``streamable-http`` to ``mcp.run`` would start FastMCP's HTTP
|
||||
# listener with no authentication, no TLS and no per-request principal — the
|
||||
# endpoint #938 owns and gates. Recognition must therefore never imply execution.
|
||||
#
|
||||
# #938 commissions the remote listener by adding REMOTE_TRANSPORT here, together
|
||||
# with the authentication and principal boundary its acceptance criteria require.
|
||||
EXECUTABLE_TRANSPORTS = frozenset({DEFAULT_TRANSPORT})
|
||||
|
||||
# Which issue owns commissioning each recognized-but-not-executable transport.
|
||||
# Used to make the refusal actionable rather than a generic denial.
|
||||
TRANSPORT_EXECUTION_OWNER = {REMOTE_TRANSPORT: "#938"}
|
||||
|
||||
BLOCKER_TRANSPORT_NOT_BOUND = "transport_not_bound"
|
||||
BLOCKER_TRANSPORT_NOT_RECOGNIZED = "transport_not_recognized"
|
||||
BLOCKER_LISTENER_NOT_COMMISSIONED = "transport_listener_not_commissioned"
|
||||
|
||||
SOURCE_CONFIGURED = "deployment_configuration"
|
||||
SOURCE_DEFAULT = "default"
|
||||
|
||||
|
||||
class TransportConfigurationError(ValueError):
|
||||
"""Raised when configured transport is outside :data:`SUPPORTED_TRANSPORTS`."""
|
||||
|
||||
|
||||
def normalize_transport(value: Any) -> str:
|
||||
"""Canonical form of a transport identifier; ``""`` when there is none.
|
||||
|
||||
Non-string values normalize to ``""`` rather than being coerced, so a
|
||||
structured object smuggled in from a caller can never match a member of the
|
||||
permitted set.
|
||||
"""
|
||||
if not isinstance(value, str):
|
||||
return ""
|
||||
return value.strip().lower()
|
||||
|
||||
|
||||
def supported_transports() -> tuple[str, ...]:
|
||||
"""Permitted identifiers, sorted, for messages and status payloads."""
|
||||
return tuple(sorted(SUPPORTED_TRANSPORTS))
|
||||
|
||||
|
||||
def is_supported_transport(value: Any) -> bool:
|
||||
"""True when *value* normalizes to a member of the permitted set."""
|
||||
return normalize_transport(value) in SUPPORTED_TRANSPORTS
|
||||
|
||||
|
||||
def is_remote_transport(value: Any) -> bool:
|
||||
"""True when *value* is a permitted transport that is not the local one."""
|
||||
name = normalize_transport(value)
|
||||
return name in SUPPORTED_TRANSPORTS and name != DEFAULT_TRANSPORT
|
||||
|
||||
|
||||
def executable_transports() -> tuple[str, ...]:
|
||||
"""Transports this entrypoint is commissioned to serve, sorted."""
|
||||
return tuple(sorted(EXECUTABLE_TRANSPORTS))
|
||||
|
||||
|
||||
def is_executable_transport(value: Any) -> bool:
|
||||
"""True when *value* may actually be served by this entrypoint (#931).
|
||||
|
||||
Strictly narrower than :func:`is_supported_transport`. A recognized
|
||||
identifier that is not executable is a correct, fully-bound configuration
|
||||
whose listener has simply not been commissioned yet.
|
||||
"""
|
||||
return normalize_transport(value) in EXECUTABLE_TRANSPORTS
|
||||
|
||||
|
||||
def assess_transport_execution(value: Any) -> dict[str, Any]:
|
||||
"""Structured serve-authorization verdict for a bound transport (#931).
|
||||
|
||||
This is the decision that separates a *recognized* transport from one this
|
||||
entrypoint may execute. It is deliberately a pure function of the bound
|
||||
identifier so the serve path cannot reach a listener the deployment has not
|
||||
commissioned.
|
||||
|
||||
Args:
|
||||
value: The bound transport identifier, or ``None`` when unbound.
|
||||
|
||||
Returns:
|
||||
dict with ``transport``, ``recognized``, ``executable``, ``allowed``,
|
||||
``blocker_kind``, ``owner_issue``, ``reasons`` and
|
||||
``exact_next_action``. ``allowed`` is true only for a bound, recognized,
|
||||
commissioned transport.
|
||||
"""
|
||||
name = normalize_transport(value)
|
||||
|
||||
if not name:
|
||||
return {
|
||||
"transport": None,
|
||||
"recognized": False,
|
||||
"executable": False,
|
||||
"allowed": False,
|
||||
"blocker_kind": BLOCKER_TRANSPORT_NOT_BOUND,
|
||||
"owner_issue": None,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"executable_transports": list(executable_transports()),
|
||||
"reasons": [
|
||||
"no transport is bound; the serve path is fail-closed until "
|
||||
"bind_native_mcp_transport succeeds (#695/#931)"
|
||||
],
|
||||
"exact_next_action": (
|
||||
"Launch through the canonical entrypoint so "
|
||||
"bind_native_mcp_transport runs before tool service."
|
||||
),
|
||||
}
|
||||
|
||||
recognized = name in SUPPORTED_TRANSPORTS
|
||||
if not recognized:
|
||||
return {
|
||||
"transport": name,
|
||||
"recognized": False,
|
||||
"executable": False,
|
||||
"allowed": False,
|
||||
"blocker_kind": BLOCKER_TRANSPORT_NOT_RECOGNIZED,
|
||||
"owner_issue": None,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"executable_transports": list(executable_transports()),
|
||||
"reasons": [
|
||||
f"transport {name!r} is not a registered MCP transport (#931); "
|
||||
"it should have been refused at bind time"
|
||||
],
|
||||
"exact_next_action": (
|
||||
f"Set {TRANSPORT_ENV} to one of {list(supported_transports())}."
|
||||
),
|
||||
}
|
||||
|
||||
if name in EXECUTABLE_TRANSPORTS:
|
||||
return {
|
||||
"transport": name,
|
||||
"recognized": True,
|
||||
"executable": True,
|
||||
"allowed": True,
|
||||
"blocker_kind": None,
|
||||
"owner_issue": None,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"executable_transports": list(executable_transports()),
|
||||
"reasons": [],
|
||||
"exact_next_action": None,
|
||||
}
|
||||
|
||||
owner = TRANSPORT_EXECUTION_OWNER.get(name)
|
||||
owner_text = owner or "the issue that commissions this transport's listener"
|
||||
return {
|
||||
"transport": name,
|
||||
"recognized": True,
|
||||
"executable": False,
|
||||
"allowed": False,
|
||||
"blocker_kind": BLOCKER_LISTENER_NOT_COMMISSIONED,
|
||||
"owner_issue": owner,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"executable_transports": list(executable_transports()),
|
||||
"reasons": [
|
||||
f"transport {name!r} is registered and was bound and recorded, but "
|
||||
f"this entrypoint is not commissioned to serve it (#931). Serving it "
|
||||
f"would start a listener with no authentication, no transport "
|
||||
f"security and no per-request principal; that endpoint is owned by "
|
||||
f"{owner_text}."
|
||||
],
|
||||
"exact_next_action": (
|
||||
f"Serve on {DEFAULT_TRANSPORT} until {owner_text} commissions the "
|
||||
f"{name!r} listener with its authentication and principal boundary, "
|
||||
f"which adds {name!r} to EXECUTABLE_TRANSPORTS."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def resolve_configured_transport(
|
||||
env: Mapping[str, str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Resolve the deployment-configured transport without raising.
|
||||
|
||||
Returns the resolution rather than a bare string so a caller can tell an
|
||||
unset value (which legitimately yields :data:`DEFAULT_TRANSPORT`) from a
|
||||
configured value that is not permitted (which must fail closed, never
|
||||
silently degrade to the default).
|
||||
|
||||
Args:
|
||||
env: Environment mapping to read; defaults to ``os.environ``.
|
||||
|
||||
Returns:
|
||||
dict with ``transport`` (normalized; the default when unset),
|
||||
``configured``, ``source``, ``raw``, ``supported``, and ``reasons``.
|
||||
"""
|
||||
source_env = os.environ if env is None else env
|
||||
raw = source_env.get(TRANSPORT_ENV)
|
||||
normalized = normalize_transport(raw)
|
||||
configured = bool(normalized)
|
||||
|
||||
if not configured:
|
||||
return {
|
||||
"transport": DEFAULT_TRANSPORT,
|
||||
"configured": False,
|
||||
"source": SOURCE_DEFAULT,
|
||||
"raw": raw,
|
||||
"supported": True,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"env_key": TRANSPORT_ENV,
|
||||
"reasons": [],
|
||||
}
|
||||
|
||||
supported = normalized in SUPPORTED_TRANSPORTS
|
||||
reasons: list[str] = []
|
||||
if not supported:
|
||||
reasons.append(
|
||||
f"{TRANSPORT_ENV}={normalized!r} is not a registered MCP transport "
|
||||
f"(#931). Registered: {list(supported_transports())}."
|
||||
)
|
||||
return {
|
||||
"transport": normalized,
|
||||
"configured": True,
|
||||
"source": SOURCE_CONFIGURED,
|
||||
"raw": raw,
|
||||
"supported": supported,
|
||||
"supported_transports": list(supported_transports()),
|
||||
"env_key": TRANSPORT_ENV,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def require_configured_transport(env: Mapping[str, str] | None = None) -> str:
|
||||
"""Resolved transport identifier, or raise when it is not permitted.
|
||||
|
||||
Raises:
|
||||
TransportConfigurationError: the configured identifier is unregistered.
|
||||
"""
|
||||
resolution = resolve_configured_transport(env)
|
||||
if not resolution["supported"]:
|
||||
raise TransportConfigurationError("; ".join(resolution["reasons"]))
|
||||
return str(resolution["transport"])
|
||||
@@ -0,0 +1,924 @@
|
||||
"""Transport-neutral MCP bind seam (#931).
|
||||
|
||||
These tests drive the *real* bind boundary — ``mark_sanctioned_daemon`` followed
|
||||
by ``bind_native_mcp_transport`` from a canonical entrypoint path, with the
|
||||
pytest allowance switched off — rather than mocking the new accessor. The
|
||||
distinction matters here for the same reason it mattered in #941: a suite that
|
||||
only exercises the helper in isolation cannot observe a seam that the live path
|
||||
never reaches.
|
||||
|
||||
Covered:
|
||||
|
||||
1. no configured transport defaults to the local transport
|
||||
2. explicit local transport binds
|
||||
3. the sanctioned remote identifier binds through the seam
|
||||
4. an unregistered identifier is rejected at bind time
|
||||
5. an invalid bind prevents the server reaching tool service
|
||||
6. the unbound state fails closed where a bind is required
|
||||
7. every transport-aware guard reads the same authoritative value
|
||||
8. client-controlled input cannot alter the bound transport
|
||||
9. the durable decision-lock record carries the selected transport
|
||||
10. existing stdio behaviour is unchanged
|
||||
11. repeated / conflicting bind attempts follow one fail-closed contract
|
||||
12. capability, role, repository and provenance protections do not regress
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
import mcp_daemon_guard
|
||||
import mcp_session_state
|
||||
import mcp_transport_config
|
||||
import irrecoverable_provenance
|
||||
|
||||
|
||||
class _ProductionBind:
|
||||
"""Context manager that reaches the real production bind path.
|
||||
|
||||
Patches only the two things a unit test cannot otherwise satisfy: the
|
||||
resolved canonical entrypoint frame, and the pytest allowance that would
|
||||
short-circuit ``mark_sanctioned_daemon``. Everything downstream of those —
|
||||
validation, pinning, the rebind contract — runs unmodified.
|
||||
"""
|
||||
|
||||
def __init__(self, env: dict[str, str] | None = None):
|
||||
self._env = env or {}
|
||||
self._stack: list = []
|
||||
|
||||
def __enter__(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
canonical = str((REPO_ROOT / "mcp_server.py").resolve())
|
||||
self._stack = [
|
||||
patch.object(
|
||||
mcp_daemon_guard,
|
||||
"_caller_official_entrypoint_path",
|
||||
side_effect=lambda: canonical,
|
||||
),
|
||||
patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False),
|
||||
patch.dict(os.environ, self._env),
|
||||
]
|
||||
for ctx in self._stack:
|
||||
ctx.__enter__()
|
||||
# Start from a clean configuration unless the test set one.
|
||||
if mcp_transport_config.TRANSPORT_ENV not in self._env:
|
||||
os.environ.pop(mcp_transport_config.TRANSPORT_ENV, None)
|
||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||
return mcp_daemon_guard
|
||||
|
||||
def __exit__(self, *exc):
|
||||
for ctx in reversed(self._stack):
|
||||
ctx.__exit__(*exc)
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
return False
|
||||
|
||||
|
||||
class TestPermittedSetIsSingleSourceOfTruth(unittest.TestCase):
|
||||
"""AC3: identifiers are enumerated once, in the seam."""
|
||||
|
||||
def test_guard_allowlist_is_the_seam_allowlist(self):
|
||||
self.assertIs(
|
||||
mcp_daemon_guard._PRODUCTION_TRANSPORTS,
|
||||
mcp_transport_config.SUPPORTED_TRANSPORTS,
|
||||
)
|
||||
|
||||
def test_default_is_a_member_of_the_permitted_set(self):
|
||||
self.assertIn(
|
||||
mcp_transport_config.DEFAULT_TRANSPORT,
|
||||
mcp_transport_config.SUPPORTED_TRANSPORTS,
|
||||
)
|
||||
|
||||
def test_remote_identifier_is_permitted_and_not_the_default(self):
|
||||
self.assertIn(
|
||||
mcp_transport_config.REMOTE_TRANSPORT,
|
||||
mcp_transport_config.SUPPORTED_TRANSPORTS,
|
||||
)
|
||||
self.assertNotEqual(
|
||||
mcp_transport_config.REMOTE_TRANSPORT,
|
||||
mcp_transport_config.DEFAULT_TRANSPORT,
|
||||
)
|
||||
self.assertTrue(
|
||||
mcp_transport_config.is_remote_transport(
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
)
|
||||
|
||||
def test_no_default_transport_literal_outside_the_seam(self):
|
||||
"""AC3: no production module reads the literal outside the seam.
|
||||
|
||||
Review 635 flagged that a fixed five-module list cannot catch a *new*
|
||||
module reintroducing the literal. This globs every production module in
|
||||
the repository root instead, so the guarantee holds for code that does
|
||||
not exist yet.
|
||||
"""
|
||||
default = mcp_transport_config.DEFAULT_TRANSPORT
|
||||
needles = (f'"{default}"', f"'{default}'")
|
||||
seam = Path(mcp_transport_config.__file__).name
|
||||
scanned: list[str] = []
|
||||
offenders: list[str] = []
|
||||
for path in sorted(REPO_ROOT.glob("*.py")):
|
||||
if path.name == seam:
|
||||
continue # the seam is the one place the literal may live
|
||||
scanned.append(path.name)
|
||||
for lineno, line in enumerate(
|
||||
path.read_text(encoding="utf-8").splitlines(), start=1
|
||||
):
|
||||
code = line.split("#", 1)[0]
|
||||
if any(needle in code for needle in needles):
|
||||
offenders.append(f"{path.name}:{lineno}: {line.strip()}")
|
||||
# Guard the guard: a glob that silently matched nothing would pass.
|
||||
self.assertGreater(len(scanned), 20, "production glob matched too little")
|
||||
self.assertIn("mcp_daemon_guard.py", scanned)
|
||||
self.assertIn("gitea_mcp_server.py", scanned)
|
||||
self.assertEqual(offenders, [], "\n".join(offenders))
|
||||
|
||||
|
||||
class TestConfiguredTransportResolution(unittest.TestCase):
|
||||
"""AC1: configuration supplies the identifier; unset still yields the default."""
|
||||
|
||||
def test_unset_yields_default(self):
|
||||
res = mcp_transport_config.resolve_configured_transport(env={})
|
||||
self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
|
||||
self.assertFalse(res["configured"])
|
||||
self.assertEqual(res["source"], mcp_transport_config.SOURCE_DEFAULT)
|
||||
self.assertTrue(res["supported"])
|
||||
self.assertEqual(res["reasons"], [])
|
||||
|
||||
def test_blank_and_whitespace_are_treated_as_unset(self):
|
||||
for raw in ("", " ", "\t\n"):
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={mcp_transport_config.TRANSPORT_ENV: raw}
|
||||
)
|
||||
self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
|
||||
self.assertFalse(res["configured"])
|
||||
|
||||
def test_explicit_default_is_reported_as_configured(self):
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
}
|
||||
)
|
||||
self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
|
||||
self.assertTrue(res["configured"])
|
||||
self.assertEqual(res["source"], mcp_transport_config.SOURCE_CONFIGURED)
|
||||
|
||||
def test_remote_identifier_resolves_and_is_supported(self):
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
)
|
||||
self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT)
|
||||
self.assertTrue(res["supported"])
|
||||
|
||||
def test_case_and_padding_are_normalized(self):
|
||||
padded = f" {mcp_transport_config.REMOTE_TRANSPORT.upper()} "
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={mcp_transport_config.TRANSPORT_ENV: padded}
|
||||
)
|
||||
self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT)
|
||||
self.assertTrue(res["supported"])
|
||||
|
||||
def test_unregistered_identifier_is_not_silently_defaulted(self):
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
|
||||
)
|
||||
self.assertFalse(res["supported"])
|
||||
self.assertEqual(res["transport"], "carrier-pigeon")
|
||||
self.assertNotEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
|
||||
self.assertTrue(res["reasons"])
|
||||
|
||||
def test_superseded_sse_transport_is_not_registered(self):
|
||||
"""A real MCP transport that this deployment does not sanction."""
|
||||
self.assertFalse(mcp_transport_config.is_supported_transport("sse"))
|
||||
res = mcp_transport_config.resolve_configured_transport(
|
||||
env={mcp_transport_config.TRANSPORT_ENV: "sse"}
|
||||
)
|
||||
self.assertFalse(res["supported"])
|
||||
|
||||
def test_require_configured_transport_raises_on_unregistered(self):
|
||||
with self.assertRaises(mcp_transport_config.TransportConfigurationError):
|
||||
mcp_transport_config.require_configured_transport(
|
||||
env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
|
||||
)
|
||||
|
||||
def test_non_string_configuration_never_matches_permitted_set(self):
|
||||
for value in (object(), 1, None, True, ["stdio"], {"t": "stdio"}):
|
||||
self.assertEqual(mcp_transport_config.normalize_transport(value), "")
|
||||
self.assertFalse(mcp_transport_config.is_supported_transport(value))
|
||||
|
||||
|
||||
class TestBindSeam(unittest.TestCase):
|
||||
"""AC1/AC2: the live bind path resolves, validates, and pins."""
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def test_1_no_configured_transport_binds_default(self):
|
||||
with _ProductionBind() as guard:
|
||||
status = guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
self.assertTrue(status["production_native_mcp_transport"])
|
||||
|
||||
def test_2_explicit_default_transport_binds(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
status = guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
self.assertTrue(guard.is_production_native_mcp_transport())
|
||||
|
||||
def test_2b_explicit_argument_still_binds(self):
|
||||
"""The pre-#931 call form keeps working for launchers and tests."""
|
||||
with _ProductionBind() as guard:
|
||||
status = guard.bind_native_mcp_transport(
|
||||
transport=mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
def test_3_sanctioned_remote_identifier_binds_through_the_seam(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
status = guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
status["transport"], mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
# The remote identifier is trusted exactly like the local one; the
|
||||
# listener that serves it is #938 and is not implemented here.
|
||||
self.assertTrue(guard.is_native_mcp_transport())
|
||||
self.assertTrue(guard.is_production_native_mcp_transport())
|
||||
guard.assert_production_mutation_runtime("remote-bind")
|
||||
|
||||
def test_4_unregistered_identifier_rejected_at_bind_time(self):
|
||||
with _ProductionBind(
|
||||
{mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
|
||||
) as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertIn("carrier-pigeon", str(ctx.exception))
|
||||
self.assertIn("#931", str(ctx.exception))
|
||||
# Nothing was bound, so nothing may dispatch.
|
||||
self.assertIsNone(guard.bound_transport())
|
||||
self.assertFalse(guard.is_native_mcp_transport())
|
||||
|
||||
def test_4b_unregistered_explicit_argument_rejected(self):
|
||||
with _ProductionBind() as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.bind_native_mcp_transport(transport="carrier-pigeon")
|
||||
self.assertIsNone(guard.bound_transport())
|
||||
|
||||
def test_4c_superseded_sse_rejected_at_bind_time(self):
|
||||
with _ProductionBind({mcp_transport_config.TRANSPORT_ENV: "sse"}) as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertIsNone(guard.bound_transport())
|
||||
|
||||
def test_5_invalid_bind_prevents_tool_service(self):
|
||||
"""A failed bind must stop the server before it serves tools."""
|
||||
with _ProductionBind(
|
||||
{mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
|
||||
) as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.bind_native_mcp_transport()
|
||||
# This is the exact expression the entrypoint passes to mcp.run.
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
|
||||
guard.assert_transport_bound("tool service")
|
||||
self.assertIn("No MCP transport is bound", str(ctx.exception))
|
||||
|
||||
def test_6_unbound_state_fails_closed(self):
|
||||
"""Entrypoint claimed but never bound — the offline-import shape."""
|
||||
with _ProductionBind() as guard:
|
||||
self.assertIsNone(guard.bound_transport())
|
||||
self.assertFalse(guard.is_native_mcp_transport())
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.assert_transport_bound("tool service")
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.assert_sanctioned_mutation_runtime("gitea_mutation")
|
||||
|
||||
def test_6b_no_runtime_at_all_fails_closed(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
|
||||
self.assertIsNone(mcp_daemon_guard.bound_transport())
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
mcp_daemon_guard.assert_transport_bound("tool service")
|
||||
|
||||
|
||||
class TestOneAuthoritativeValue(unittest.TestCase):
|
||||
"""AC: every transport-aware guard observes the same value."""
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def test_7_all_guards_read_the_same_bound_value(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
expected = mcp_transport_config.REMOTE_TRANSPORT
|
||||
|
||||
self.assertEqual(guard.bound_transport(), expected)
|
||||
self.assertEqual(guard.assert_transport_bound(), expected)
|
||||
self.assertEqual(guard.native_runtime_status()["bound_transport"], expected)
|
||||
self.assertEqual(guard.native_runtime_status()["transport"], expected)
|
||||
self.assertEqual(
|
||||
guard.mutation_provenance_fields()["bound_transport"], expected
|
||||
)
|
||||
self.assertEqual(
|
||||
irrecoverable_provenance.assess_transport_for_auth_mint()[
|
||||
"bound_transport"
|
||||
],
|
||||
expected,
|
||||
)
|
||||
|
||||
def test_8_environment_change_after_bind_cannot_move_the_value(self):
|
||||
"""Client- or environment-shaped input must not alter a bound transport."""
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
# A stray launcher (or an attacker) rewrites config post-bind.
|
||||
os.environ[mcp_transport_config.TRANSPORT_ENV] = (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
guard.mutation_provenance_fields()["bound_transport"],
|
||||
mcp_transport_config.DEFAULT_TRANSPORT,
|
||||
)
|
||||
os.environ[mcp_transport_config.TRANSPORT_ENV] = "carrier-pigeon"
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
def test_8b_bound_transport_takes_no_caller_argument(self):
|
||||
"""The accessor cannot be steered by a tool parameter."""
|
||||
import inspect
|
||||
|
||||
self.assertEqual(
|
||||
list(inspect.signature(mcp_daemon_guard.bound_transport).parameters), []
|
||||
)
|
||||
|
||||
def test_11_rebinding_the_same_transport_is_idempotent(self):
|
||||
with _ProductionBind() as guard:
|
||||
first = guard.bind_native_mcp_transport()
|
||||
second = guard.bind_native_mcp_transport()
|
||||
self.assertEqual(first["transport"], second["transport"])
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
def test_11b_rebinding_a_different_transport_fails_closed(self):
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
|
||||
guard.bind_native_mcp_transport(
|
||||
transport=mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertIn("already bound", str(ctx.exception))
|
||||
# The first value survives the attempt.
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
def test_11c_rebinding_an_unregistered_transport_fails_closed(self):
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError):
|
||||
guard.bind_native_mcp_transport(transport="carrier-pigeon")
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
|
||||
class TestDurableRecordCarriesTransport(unittest.TestCase):
|
||||
"""AC4: the identifier reaches a durable decision-lock record."""
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def _save_and_read_decision_lock(self, state_dir: str) -> dict:
|
||||
mcp_session_state.save_state(
|
||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
||||
payload={"pr_number": 931, "action": "COMMENT"},
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
profile_identity="prgs-author",
|
||||
state_dir=state_dir,
|
||||
)
|
||||
loaded = mcp_session_state.load_state(
|
||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
profile_identity="prgs-author",
|
||||
state_dir=state_dir,
|
||||
)
|
||||
self.assertIsNotNone(loaded)
|
||||
return loaded
|
||||
|
||||
def test_9_decision_lock_records_the_bound_transport(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
mcp_daemon_guard.install_test_native_runtime()
|
||||
record = self._save_and_read_decision_lock(tmp)
|
||||
self.assertIn("bound_transport", record)
|
||||
self.assertEqual(
|
||||
record["bound_transport"], mcp_daemon_guard.bound_transport()
|
||||
)
|
||||
|
||||
def test_9b_unbound_runtime_records_no_transport_identifier(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
record = self._save_and_read_decision_lock(tmp)
|
||||
self.assertIn("bound_transport", record)
|
||||
self.assertIsNone(record["bound_transport"])
|
||||
|
||||
def test_9c_provenance_fields_expose_the_identifier(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
fields = mcp_daemon_guard.mutation_provenance_fields()
|
||||
self.assertIn("bound_transport", fields)
|
||||
self.assertIsNone(fields["bound_transport"])
|
||||
|
||||
|
||||
class TestStdioBehaviourUnchanged(unittest.TestCase):
|
||||
"""AC6 / prompt items 10 and 12: no regression on the existing path."""
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def test_10_trust_class_field_keeps_its_pre_931_values(self):
|
||||
"""``transport`` remains the trust class, not the identifier."""
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
self.assertEqual(
|
||||
mcp_daemon_guard.mutation_provenance_fields()["transport"], "untrusted"
|
||||
)
|
||||
mcp_daemon_guard.install_test_native_runtime()
|
||||
self.assertEqual(
|
||||
mcp_daemon_guard.mutation_provenance_fields()["transport"],
|
||||
"test_native_mcp",
|
||||
)
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
guard.mutation_provenance_fields()["transport"], "native_mcp"
|
||||
)
|
||||
|
||||
def test_10b_default_bind_reproduces_the_pre_931_runtime_record(self):
|
||||
with _ProductionBind() as guard:
|
||||
status = guard.bind_native_mcp_transport()
|
||||
self.assertTrue(status["native_mcp_transport"])
|
||||
self.assertTrue(status["production_native_mcp_transport"])
|
||||
self.assertEqual(status["mode"], "production")
|
||||
self.assertEqual(status["phase"], "transport_bound")
|
||||
self.assertEqual(
|
||||
status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
guard.assert_sanctioned_mutation_runtime("native-ide")
|
||||
guard.assert_production_mutation_runtime("native-ide")
|
||||
|
||||
def test_12_session_state_root_still_pinned_at_bind(self):
|
||||
"""#695 AC2 must survive the seam."""
|
||||
with tempfile.TemporaryDirectory() as legit:
|
||||
with tempfile.TemporaryDirectory() as rogue:
|
||||
with _ProductionBind(
|
||||
{mcp_daemon_guard.SESSION_STATE_DIR_ENV: legit}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
guard.pinned_session_state_dir(), str(Path(legit).resolve())
|
||||
)
|
||||
os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue
|
||||
self.assertEqual(
|
||||
guard.pinned_session_state_dir(), str(Path(legit).resolve())
|
||||
)
|
||||
|
||||
def test_12b_test_mode_record_still_cannot_authorize_production(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
mcp_daemon_guard.install_test_native_runtime()
|
||||
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
||||
self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint")
|
||||
|
||||
def test_12c_bind_still_requires_the_canonical_entrypoint(self):
|
||||
"""A remote identifier does not relax provenance."""
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
|
||||
with patch.object(
|
||||
mcp_daemon_guard,
|
||||
"_caller_official_entrypoint_path",
|
||||
return_value=None,
|
||||
):
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
mcp_daemon_guard.bind_native_mcp_transport(
|
||||
transport=mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertIsNone(mcp_daemon_guard.bound_transport())
|
||||
|
||||
def test_12d_bind_still_requires_a_prior_entrypoint_claim(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
canonical = str((REPO_ROOT / "mcp_server.py").resolve())
|
||||
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
|
||||
with patch.object(
|
||||
mcp_daemon_guard,
|
||||
"_caller_official_entrypoint_path",
|
||||
side_effect=lambda: canonical,
|
||||
):
|
||||
# No mark_sanctioned_daemon() first.
|
||||
with self.assertRaises(
|
||||
mcp_daemon_guard.UnsanctionedRuntimeError
|
||||
) as ctx:
|
||||
mcp_daemon_guard.bind_native_mcp_transport()
|
||||
self.assertIn("no entrypoint claim", str(ctx.exception))
|
||||
|
||||
def test_12e_auth_mint_verdict_is_unchanged_for_the_default_transport(self):
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
verdict = irrecoverable_provenance.assess_transport_for_auth_mint()
|
||||
self.assertTrue(verdict["allowed"])
|
||||
self.assertTrue(verdict["native_mcp_transport"])
|
||||
self.assertTrue(verdict["production_native_mcp_transport"])
|
||||
self.assertEqual(verdict["reasons"], [])
|
||||
|
||||
def test_12f_auth_mint_still_refuses_an_unbound_runtime(self):
|
||||
with _ProductionBind():
|
||||
# Claimed but never bound.
|
||||
verdict = irrecoverable_provenance.assess_transport_for_auth_mint()
|
||||
self.assertFalse(verdict["allowed"])
|
||||
self.assertTrue(verdict["reasons"])
|
||||
self.assertIsNone(verdict["bound_transport"])
|
||||
|
||||
|
||||
class TestExecutionBoundary(unittest.TestCase):
|
||||
"""Recognition is not execution authorization (#931, review 635 B1/B2).
|
||||
|
||||
A registered remote identifier must still bind, pin and record — the #931
|
||||
seam — while being refused at the serve boundary, because serving it would
|
||||
start a listener with no authentication or per-request principal. That
|
||||
listener belongs to #938.
|
||||
"""
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
# -- the two sets are distinct, and narrower in the right direction ----
|
||||
|
||||
def test_executable_set_is_a_strict_subset_of_recognized(self):
|
||||
self.assertTrue(
|
||||
mcp_transport_config.EXECUTABLE_TRANSPORTS
|
||||
< mcp_transport_config.SUPPORTED_TRANSPORTS
|
||||
)
|
||||
|
||||
def test_default_transport_is_executable(self):
|
||||
self.assertTrue(
|
||||
mcp_transport_config.is_executable_transport(
|
||||
mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
)
|
||||
|
||||
def test_remote_transport_is_recognized_but_not_executable(self):
|
||||
remote = mcp_transport_config.REMOTE_TRANSPORT
|
||||
self.assertTrue(mcp_transport_config.is_supported_transport(remote))
|
||||
self.assertFalse(mcp_transport_config.is_executable_transport(remote))
|
||||
|
||||
def test_remote_listener_ownership_is_declared(self):
|
||||
self.assertEqual(
|
||||
mcp_transport_config.TRANSPORT_EXECUTION_OWNER[
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
],
|
||||
"#938",
|
||||
)
|
||||
|
||||
# -- stdio still reaches the runner, unchanged -------------------------
|
||||
|
||||
def test_default_transport_is_authorized_for_service(self):
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
guard.authorize_transport_execution("tool service"),
|
||||
mcp_transport_config.DEFAULT_TRANSPORT,
|
||||
)
|
||||
self.assertTrue(guard.assess_serve_authorization()["allowed"])
|
||||
|
||||
def test_default_transport_reaches_the_real_runner(self):
|
||||
"""The production runner is actually invoked, with stdio, unchanged."""
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
seen = {}
|
||||
|
||||
class _Runner:
|
||||
def run(self, transport=None, **kw):
|
||||
seen["transport"] = transport
|
||||
|
||||
_Runner().run(transport=guard.authorize_transport_execution("tool service"))
|
||||
self.assertEqual(
|
||||
seen["transport"], mcp_transport_config.DEFAULT_TRANSPORT
|
||||
)
|
||||
|
||||
# -- streamable-http binds, records, and is refused before serving -----
|
||||
|
||||
def test_remote_transport_binds_and_is_recorded(self):
|
||||
"""The #931 seam is intact: it binds, pins and records."""
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertEqual(
|
||||
guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertEqual(
|
||||
guard.mutation_provenance_fields()["bound_transport"],
|
||||
mcp_transport_config.REMOTE_TRANSPORT,
|
||||
)
|
||||
|
||||
def test_remote_transport_is_refused_at_the_serve_boundary(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.TransportExecutionError) as ctx:
|
||||
guard.authorize_transport_execution("tool service")
|
||||
err = ctx.exception
|
||||
self.assertEqual(
|
||||
err.blocker_kind,
|
||||
mcp_transport_config.BLOCKER_LISTENER_NOT_COMMISSIONED,
|
||||
)
|
||||
self.assertEqual(err.owner_issue, "#938")
|
||||
self.assertEqual(err.transport, mcp_transport_config.REMOTE_TRANSPORT)
|
||||
|
||||
def test_refusal_names_transport_and_unmet_requirement_and_owner(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.TransportExecutionError) as ctx:
|
||||
guard.authorize_transport_execution("tool service")
|
||||
text = str(ctx.exception)
|
||||
self.assertIn(mcp_transport_config.REMOTE_TRANSPORT, text)
|
||||
self.assertIn("#938", text)
|
||||
self.assertIn("not commissioned", text)
|
||||
self.assertIn("#931", text)
|
||||
|
||||
def test_remote_transport_never_reaches_the_runner(self):
|
||||
"""No transport value is handed to a run() call for the remote case."""
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
calls = []
|
||||
|
||||
class _Runner:
|
||||
def run(self, transport=None, **kw):
|
||||
calls.append(transport)
|
||||
|
||||
with self.assertRaises(guard.TransportExecutionError):
|
||||
_Runner().run(
|
||||
transport=guard.authorize_transport_execution("tool service")
|
||||
)
|
||||
self.assertEqual(calls, [], "runner must never be invoked")
|
||||
|
||||
def test_no_http_listener_is_created_for_remote_transport(self):
|
||||
"""Nothing in the refusal path touches uvicorn or a socket bind."""
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
import socket
|
||||
|
||||
bound_sockets = []
|
||||
real_bind = socket.socket.bind
|
||||
|
||||
def _tripwire(self, addr): # pragma: no cover - must not run
|
||||
bound_sockets.append(addr)
|
||||
return real_bind(self, addr)
|
||||
|
||||
with patch.object(socket.socket, "bind", _tripwire):
|
||||
with self.assertRaises(guard.TransportExecutionError):
|
||||
guard.authorize_transport_execution("tool service")
|
||||
self.assertEqual(bound_sockets, [], "no socket may be bound")
|
||||
|
||||
def test_no_mutation_is_authorized_after_the_denial(self):
|
||||
"""The denial leaves no partial state that would let a tool dispatch."""
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.TransportExecutionError):
|
||||
guard.authorize_transport_execution("tool service")
|
||||
# Serve stays unauthorized on every subsequent query.
|
||||
self.assertFalse(guard.assess_serve_authorization()["allowed"])
|
||||
self.assertFalse(guard.native_runtime_status()["serve_authorized"])
|
||||
with self.assertRaises(guard.TransportExecutionError):
|
||||
guard.authorize_transport_execution("tool service")
|
||||
|
||||
# -- B2: the decision genuinely consumes bound_transport ---------------
|
||||
|
||||
def test_serve_decision_consumes_bound_transport(self):
|
||||
"""Changing only bound_transport flips the verdict."""
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertTrue(guard.assess_serve_authorization()["allowed"])
|
||||
with patch.object(
|
||||
guard,
|
||||
"bound_transport",
|
||||
return_value=mcp_transport_config.REMOTE_TRANSPORT,
|
||||
):
|
||||
verdict = guard.assess_serve_authorization()
|
||||
self.assertFalse(verdict["allowed"])
|
||||
self.assertEqual(
|
||||
verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
with self.assertRaises(guard.TransportExecutionError):
|
||||
guard.authorize_transport_execution("tool service")
|
||||
|
||||
def test_serve_verdict_reports_the_bound_transport(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
verdict = guard.assess_serve_authorization()
|
||||
self.assertEqual(
|
||||
verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
self.assertTrue(verdict["recognized"])
|
||||
self.assertFalse(verdict["executable"])
|
||||
|
||||
# -- earlier and later boundaries are unchanged ------------------------
|
||||
|
||||
def test_unregistered_identifier_still_fails_at_bind_not_at_serve(self):
|
||||
with _ProductionBind(
|
||||
{mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
|
||||
) as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertIn("not a registered MCP transport", str(ctx.exception))
|
||||
self.assertIsNone(guard.bound_transport())
|
||||
|
||||
def test_unbound_execution_keeps_the_pre_existing_failure(self):
|
||||
with _ProductionBind() as guard:
|
||||
with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
|
||||
guard.authorize_transport_execution("tool service")
|
||||
self.assertIn("No MCP transport is bound", str(ctx.exception))
|
||||
self.assertNotIsInstance(ctx.exception, guard.TransportExecutionError)
|
||||
|
||||
def test_transport_execution_error_is_caught_by_existing_handlers(self):
|
||||
"""Subclassing keeps every pre-existing fail-closed handler correct."""
|
||||
self.assertTrue(
|
||||
issubclass(
|
||||
mcp_daemon_guard.TransportExecutionError,
|
||||
mcp_daemon_guard.UnsanctionedRuntimeError,
|
||||
)
|
||||
)
|
||||
|
||||
def test_test_mode_runtime_is_not_servable(self):
|
||||
"""The pytest-only record is outside the recognized and executable sets."""
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
mcp_daemon_guard.install_test_native_runtime()
|
||||
self.assertNotIn(
|
||||
mcp_daemon_guard.bound_transport(),
|
||||
mcp_transport_config.SUPPORTED_TRANSPORTS,
|
||||
)
|
||||
self.assertFalse(mcp_daemon_guard.assess_serve_authorization()["allowed"])
|
||||
|
||||
def test_serve_authorization_does_not_leak_across_runtimes(self):
|
||||
"""A later runtime's verdict never reflects an earlier one."""
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
)
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertFalse(guard.assess_serve_authorization()["allowed"])
|
||||
with _ProductionBind() as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
self.assertTrue(guard.assess_serve_authorization()["allowed"])
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
self.assertEqual(
|
||||
mcp_daemon_guard.assess_serve_authorization()["blocker_kind"],
|
||||
mcp_transport_config.BLOCKER_TRANSPORT_NOT_BOUND,
|
||||
)
|
||||
|
||||
def test_refusal_carries_no_credential_material(self):
|
||||
with _ProductionBind(
|
||||
{
|
||||
mcp_transport_config.TRANSPORT_ENV: (
|
||||
mcp_transport_config.REMOTE_TRANSPORT
|
||||
),
|
||||
"GITEA_TOKEN": "super-secret-value",
|
||||
}
|
||||
) as guard:
|
||||
guard.bind_native_mcp_transport()
|
||||
with self.assertRaises(guard.TransportExecutionError) as ctx:
|
||||
guard.authorize_transport_execution("tool service")
|
||||
blob = str(ctx.exception) + repr(ctx.exception.assessment)
|
||||
self.assertNotIn("super-secret-value", blob)
|
||||
self.assertNotIn("GITEA_TOKEN", blob)
|
||||
|
||||
|
||||
class TestEntrypointWiring(unittest.TestCase):
|
||||
"""The live entrypoint must use the seam and the execution guard."""
|
||||
|
||||
def test_entrypoint_binds_without_a_literal_transport(self):
|
||||
text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8")
|
||||
self.assertIn("mcp_daemon_guard.bind_native_mcp_transport()", text)
|
||||
self.assertNotIn('bind_native_mcp_transport(transport="stdio")', text)
|
||||
|
||||
def test_entrypoint_serves_only_through_the_execution_guard(self):
|
||||
text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8")
|
||||
self.assertIn(
|
||||
"mcp.run(transport=mcp_daemon_guard.authorize_transport_execution", text
|
||||
)
|
||||
self.assertNotIn('mcp.run(transport="stdio")', text)
|
||||
self.assertNotIn(
|
||||
"mcp.run(transport=mcp_daemon_guard.assert_transport_bound", text
|
||||
)
|
||||
|
||||
def test_every_run_call_in_production_goes_through_the_guard(self):
|
||||
"""Glob the production surface: no serve site may bypass the guard."""
|
||||
offenders: list[str] = []
|
||||
run_sites = 0
|
||||
for path in sorted(REPO_ROOT.glob("*.py")):
|
||||
for lineno, line in enumerate(
|
||||
path.read_text(encoding="utf-8").splitlines(), start=1
|
||||
):
|
||||
code = line.split("#", 1)[0]
|
||||
if "mcp.run(" not in code:
|
||||
continue
|
||||
run_sites += 1
|
||||
if "authorize_transport_execution" not in code:
|
||||
offenders.append(f"{path.name}:{lineno}: {line.strip()}")
|
||||
self.assertEqual(run_sites, 1, "expected exactly one serve site")
|
||||
self.assertEqual(offenders, [], "\n".join(offenders))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,323 @@
|
||||
"""Validation tooling for the remote-MCP threat model (#956).
|
||||
|
||||
#956 requires that "every boundary claim [is] traceable to a file and line
|
||||
anchor that resolves at the reviewed commit". A prose document cannot enforce
|
||||
that about itself, and #930 demonstrated the failure mode: its inventory cited
|
||||
``gitea_mcp_server.py`` anchors generated at ``7bf4f125`` which no longer point
|
||||
at the described code at ``aad5c8b4``. Nothing failed, because nothing checked.
|
||||
|
||||
These tests are that check. They enforce, in both directions:
|
||||
|
||||
* every ``file.py:NNN`` anchor cited in the prose is declared in the fixture;
|
||||
* every declared anchor resolves — the file exists, the line exists, and the
|
||||
source line actually contains the substring the fixture claims for it;
|
||||
* the document's structural obligations (assets, adversaries, boundaries,
|
||||
credential rows, the co-residency ruling, and the child mapping) are present
|
||||
and internally consistent.
|
||||
|
||||
A refactor that shifts a line number therefore breaks the suite instead of
|
||||
silently rotting the security documentation.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import unittest
|
||||
|
||||
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
DOC_PATH = os.path.join(REPO_ROOT, "docs", "remote-mcp", "threat-model.md")
|
||||
FIXTURE_PATH = os.path.join(
|
||||
REPO_ROOT, "docs", "remote-mcp", "threat-model-anchors.json"
|
||||
)
|
||||
|
||||
# ``module.py:123`` as it appears inside markdown inline code spans.
|
||||
ANCHOR_RE = re.compile(r"`([A-Za-z0-9_./-]+\.py):(\d+)`")
|
||||
|
||||
# The epic children this document must map to a boundary (#929 children 2-10).
|
||||
REQUIRED_CHILDREN = [931, 932, 933, 934, 935, 936, 937, 938, 939]
|
||||
|
||||
# The adversaries #956 names explicitly.
|
||||
REQUIRED_ADVERSARIES = [
|
||||
"compromised LLM client",
|
||||
"prompt injection",
|
||||
"malicious tool arguments",
|
||||
"network attacker",
|
||||
"curious operator",
|
||||
]
|
||||
|
||||
|
||||
def _read(path):
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
return fh.read()
|
||||
|
||||
|
||||
def _heading_re(title):
|
||||
"""Match a level-2 heading by title, with or without section numbering.
|
||||
|
||||
The document numbers its sections ('## 6. Decomposition ruling'), so an
|
||||
exact-substring assertion would break on renumbering without the document
|
||||
having actually lost anything.
|
||||
"""
|
||||
return re.compile(
|
||||
r"^##\s+(?:\d+\.\s+)?" + re.escape(title), re.MULTILINE
|
||||
)
|
||||
|
||||
|
||||
def _section_body(doc, title):
|
||||
"""Return the text of section *title*, bounded by the next level-2 heading.
|
||||
|
||||
Bounding matters: an unbounded slice runs to end-of-document, so the
|
||||
walkthrough tables in a later section leak into the child-to-boundary
|
||||
mapping and satisfy its coverage check with rows that assign no owner.
|
||||
"""
|
||||
match = _heading_re(title).search(doc)
|
||||
if match is None:
|
||||
return None
|
||||
rest = doc[match.end():]
|
||||
nxt = re.search(r"^##\s", rest, re.MULTILINE)
|
||||
return rest[: nxt.start()] if nxt else rest
|
||||
|
||||
|
||||
def _source_line(rel_path, lineno):
|
||||
"""Return the 1-based *lineno* of *rel_path*, or None if out of range."""
|
||||
abs_path = os.path.join(REPO_ROOT, rel_path)
|
||||
if not os.path.exists(abs_path):
|
||||
return None
|
||||
with open(abs_path, "r", encoding="utf-8", errors="replace") as fh:
|
||||
for idx, line in enumerate(fh, start=1):
|
||||
if idx == lineno:
|
||||
return line
|
||||
return None
|
||||
|
||||
|
||||
class ThreatModelFixtureTests(unittest.TestCase):
|
||||
"""The fixture itself must be well-formed before it can prove anything."""
|
||||
|
||||
def setUp(self):
|
||||
self.fixture = json.loads(_read(FIXTURE_PATH))
|
||||
|
||||
def test_fixture_declares_a_generation_commit(self):
|
||||
sha = self.fixture.get("generated_against_commit") or ""
|
||||
self.assertRegex(
|
||||
sha,
|
||||
r"^[0-9a-f]{40}$",
|
||||
"the fixture must record the full commit its anchors were taken at",
|
||||
)
|
||||
|
||||
def test_fixture_anchors_are_unique_and_well_formed(self):
|
||||
seen = set()
|
||||
for entry in self.fixture["anchors"]:
|
||||
anchor = entry["anchor"]
|
||||
self.assertNotIn(anchor, seen, f"duplicate anchor entry: {anchor}")
|
||||
seen.add(anchor)
|
||||
self.assertRegex(anchor, r"^[A-Za-z0-9_./-]+\.py:[1-9]\d*$", anchor)
|
||||
self.assertTrue(
|
||||
(entry.get("expect") or "").strip(),
|
||||
f"anchor {anchor} declares no 'expect' substring, so it proves nothing",
|
||||
)
|
||||
|
||||
|
||||
class ThreatModelAnchorResolutionTests(unittest.TestCase):
|
||||
"""#956 required positive test: every anchor resolves at the reviewed commit."""
|
||||
|
||||
def setUp(self):
|
||||
self.fixture = json.loads(_read(FIXTURE_PATH))
|
||||
self.doc = _read(DOC_PATH)
|
||||
|
||||
def test_every_declared_anchor_resolves_to_the_claimed_source_line(self):
|
||||
failures = []
|
||||
for entry in self.fixture["anchors"]:
|
||||
rel_path, _, raw_lineno = entry["anchor"].partition(":")
|
||||
lineno = int(raw_lineno)
|
||||
line = _source_line(rel_path, lineno)
|
||||
if line is None:
|
||||
failures.append(f"{entry['anchor']}: file or line does not exist")
|
||||
continue
|
||||
if entry["expect"] not in line:
|
||||
failures.append(
|
||||
f"{entry['anchor']}: expected {entry['expect']!r}, "
|
||||
f"found {line.strip()!r}"
|
||||
)
|
||||
self.assertEqual(
|
||||
[], failures, "unresolved threat-model anchors:\n" + "\n".join(failures)
|
||||
)
|
||||
|
||||
def test_every_anchor_cited_in_the_document_is_declared_in_the_fixture(self):
|
||||
declared = {e["anchor"] for e in self.fixture["anchors"]}
|
||||
cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
|
||||
undeclared = sorted(cited - declared)
|
||||
self.assertEqual(
|
||||
[],
|
||||
undeclared,
|
||||
"document cites anchors that no test verifies: " + ", ".join(undeclared),
|
||||
)
|
||||
|
||||
def test_the_document_actually_cites_anchors(self):
|
||||
cited = {f"{m.group(1)}:{m.group(2)}" for m in ANCHOR_RE.finditer(self.doc)}
|
||||
self.assertGreaterEqual(
|
||||
len(cited),
|
||||
30,
|
||||
"a boundary document with almost no anchors is not traceable",
|
||||
)
|
||||
|
||||
def test_unresolvable_anchor_is_detected(self):
|
||||
"""Negative control: the checker must fail on a deliberately bad anchor.
|
||||
|
||||
Without this, a checker that silently passed everything would look
|
||||
identical to a correct one.
|
||||
"""
|
||||
self.assertIsNone(_source_line("gitea_config.py", 10**9))
|
||||
self.assertIsNone(_source_line("no_such_module_for_956.py", 1))
|
||||
real = _source_line("gitea_config.py", 54)
|
||||
self.assertIsNotNone(real)
|
||||
self.assertNotIn("this substring is not on that line", real)
|
||||
|
||||
|
||||
class ThreatModelStructureTests(unittest.TestCase):
|
||||
"""The document must contain what #956's acceptance criteria demand."""
|
||||
|
||||
def setUp(self):
|
||||
self.doc = _read(DOC_PATH)
|
||||
|
||||
def test_records_the_commit_it_was_generated_against(self):
|
||||
fixture = json.loads(_read(FIXTURE_PATH))
|
||||
self.assertIn(
|
||||
fixture["generated_against_commit"],
|
||||
self.doc,
|
||||
"the document must state the commit its anchors resolve at",
|
||||
)
|
||||
|
||||
def test_names_every_required_adversary(self):
|
||||
low = self.doc.lower()
|
||||
for adversary in REQUIRED_ADVERSARIES:
|
||||
self.assertIn(adversary.lower(), low, f"adversary not covered: {adversary}")
|
||||
|
||||
def test_maps_every_epic_child_from_two_through_ten(self):
|
||||
for number in REQUIRED_CHILDREN:
|
||||
self.assertIn(
|
||||
f"#{number}",
|
||||
self.doc,
|
||||
f"epic child #{number} is not mapped to a boundary",
|
||||
)
|
||||
|
||||
def test_credential_rows_declare_holder_boundary_and_blast_radius(self):
|
||||
for column in ("Holder", "Boundary", "Blast radius"):
|
||||
self.assertIn(
|
||||
column,
|
||||
self.doc,
|
||||
f"the credential inventory must state each credential's {column.lower()}",
|
||||
)
|
||||
|
||||
def test_states_an_explicit_co_residency_ruling(self):
|
||||
"""AC3/AC5: an explicit ruling, not an implication."""
|
||||
self.assertIsNotNone(
|
||||
_heading_re("Decomposition ruling").search(self.doc),
|
||||
"the document must contain an explicit decomposition-ruling section",
|
||||
)
|
||||
for service in ("Jenkins", "GlitchTip", "Sentry", "database"):
|
||||
self.assertIn(service, self.doc, f"ruling does not address {service}")
|
||||
self.assertRegex(
|
||||
self.doc,
|
||||
r"D1\b.*must not",
|
||||
"the ruling must state the prohibition, not merely discuss it",
|
||||
)
|
||||
|
||||
def test_contains_the_compromised_client_walkthrough(self):
|
||||
"""#956 required negative/adversarial test."""
|
||||
self.assertIsNotNone(
|
||||
_heading_re("Adversarial walkthrough").search(self.doc),
|
||||
"the required compromised-client walkthrough is missing",
|
||||
)
|
||||
self.assertIn("Before the migration", self.doc)
|
||||
self.assertIn("After the migration", self.doc)
|
||||
|
||||
def test_every_boundary_states_what_it_protects_and_what_crossing_requires(self):
|
||||
boundary_ids = set(re.findall(r"\bB(\d+)\b", self.doc))
|
||||
self.assertGreaterEqual(
|
||||
len(boundary_ids), 5, "too few trust boundaries to be a decomposition"
|
||||
)
|
||||
for column in (
|
||||
"Protects",
|
||||
"Crossing requires today",
|
||||
"Crossing must require remotely",
|
||||
):
|
||||
self.assertIn(column, self.doc, f"boundary table is missing '{column}'")
|
||||
|
||||
def test_declares_itself_documentation_only(self):
|
||||
self.assertIn("documentation only", self.doc.lower())
|
||||
|
||||
|
||||
class ThreatModelConsistencyTests(unittest.TestCase):
|
||||
"""Counts stated in prose must match the rows actually present."""
|
||||
|
||||
def setUp(self):
|
||||
self.doc = _read(DOC_PATH)
|
||||
|
||||
def _declared_ids(self, prefix):
|
||||
# Table rows begin '| CR1 |' / '| B3 |' / '| A2 |'.
|
||||
return sorted(
|
||||
{
|
||||
int(m)
|
||||
for m in re.findall(
|
||||
r"^\|\s*%s(\d+)\s*\|" % prefix, self.doc, re.MULTILINE
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
def test_identifier_sequences_have_no_gaps(self):
|
||||
for prefix, label in (
|
||||
("A", "assets"),
|
||||
("B", "boundaries"),
|
||||
("CR", "credentials"),
|
||||
):
|
||||
ids = self._declared_ids(prefix)
|
||||
self.assertTrue(ids, f"no {label} declared")
|
||||
self.assertEqual(
|
||||
list(range(1, len(ids) + 1)),
|
||||
ids,
|
||||
f"{label} identifiers must run 1..n with no gaps; got {ids}",
|
||||
)
|
||||
|
||||
def test_stated_credential_count_matches_the_rows(self):
|
||||
ids = self._declared_ids("CR")
|
||||
match = re.search(r"(\d+)\s+credential(?:s)? in total", self.doc)
|
||||
self.assertIsNotNone(match, "the credential inventory must state its own total")
|
||||
self.assertEqual(
|
||||
len(ids),
|
||||
int(match.group(1)),
|
||||
"stated credential total disagrees with the number of rows",
|
||||
)
|
||||
|
||||
def test_every_boundary_is_owned_by_at_least_one_child(self):
|
||||
"""Each boundary must be owned by a child *in the mapping table*.
|
||||
|
||||
Scanning the whole section would let a prose summary line ("Boundary
|
||||
coverage: ... B5 (#936)") satisfy the assertion while the table row
|
||||
that actually assigns the owner had been emptied — verified by
|
||||
deliberately blanking a row and watching a whole-section check still
|
||||
pass. Only table rows count.
|
||||
"""
|
||||
mapping_section = _section_body(self.doc, "Child-to-boundary mapping")
|
||||
self.assertIsNotNone(
|
||||
mapping_section, "child-to-boundary mapping section is missing"
|
||||
)
|
||||
rows = [
|
||||
line
|
||||
for line in mapping_section.splitlines()
|
||||
if line.lstrip().startswith("|") and re.search(r"#93\d", line)
|
||||
]
|
||||
self.assertGreaterEqual(
|
||||
len(rows), len(REQUIRED_CHILDREN), "mapping table has too few child rows"
|
||||
)
|
||||
mapped = set(re.findall(r"\bB(\d+)\b", "\n".join(rows)))
|
||||
declared = {str(i) for i in self._declared_ids("B")}
|
||||
unmapped = sorted(declared - mapped, key=int)
|
||||
self.assertEqual(
|
||||
[],
|
||||
unmapped,
|
||||
"boundaries with no owning child: " + ", ".join("B" + u for u in unmapped),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user