Compare commits
15 Commits
472e6850fe
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 9f75e28094 | |||
| c063842b2e | |||
| e0861bcb03 | |||
| 9d6a2e0a5f | |||
| 205f089c44 | |||
| ff920a6496 | |||
| fbf1bc5f5c | |||
| 255cfc87dd | |||
| 8d2eb23237 | |||
| 7fa1bb9cfb | |||
| ed3cc106aa | |||
| e63cf5b5eb | |||
| 6dbd51b2a4 | |||
| e9c67e7292 | |||
| 65ea7514d2 |
@@ -6,6 +6,7 @@ __pycache__/
|
||||
# Real JSON runtime-profile configs may reference private hosts; keep only the example.
|
||||
gitea-mcp*.json
|
||||
!gitea-mcp.example.json
|
||||
!gitea-mcp.v2-contexts.example.json
|
||||
.vscode/
|
||||
graphify-out/
|
||||
branches/
|
||||
|
||||
@@ -134,8 +134,50 @@ Rules:
|
||||
appears in both, it is forbidden.
|
||||
- An operation not present in `allowed_operations` is treated as **not
|
||||
allowed** (deny by default).
|
||||
- These categories are descriptive for this issue. Their runtime enforcement is
|
||||
out of scope here (see roadmap links).
|
||||
|
||||
## Operation-name normalization (#106)
|
||||
|
||||
Canonical operation names are namespaced: `{service}.{area}.{verb}` (e.g.
|
||||
`gitea.pr.merge`, `jenkins.build.read`). Legacy unqualified spellings are
|
||||
accepted **only** through the explicit alias table below (the code of record
|
||||
is `GITEA_OPERATION_ALIASES` in `gitea_config.py`; the enforcement matrix is
|
||||
`tests/test_op_normalization.py`).
|
||||
|
||||
| Legacy spelling | Canonical operation |
|
||||
|-------------------|----------------------------|
|
||||
| `read` | `gitea.read` |
|
||||
| `review` | `gitea.pr.review` |
|
||||
| `comment` | `gitea.pr.comment` |
|
||||
| `approve` | `gitea.pr.approve` |
|
||||
| `request_changes` | `gitea.pr.request_changes` |
|
||||
| `merge` | `gitea.pr.merge` |
|
||||
| `pr.create` | `gitea.pr.create` |
|
||||
| `branch.push` | `gitea.branch.push` |
|
||||
| `branch` | `gitea.branch.create` |
|
||||
| `commit` | `gitea.repo.commit` |
|
||||
| `push` | `gitea.branch.push` |
|
||||
| `open_pr` | `gitea.pr.create` |
|
||||
|
||||
For non-Gitea services, a single unqualified word namespaces to the checked
|
||||
service (`read` → `jenkins.read` when checking Jenkins); names already
|
||||
prefixed with that service pass through unchanged.
|
||||
|
||||
Enforcement rules (`gitea_config.check_operation`, run **before** any
|
||||
allowed/forbidden membership check):
|
||||
|
||||
- Unknown operation names fail closed (denied).
|
||||
- Ambiguous names — dotted names that are neither service-prefixed nor in the
|
||||
alias table — fail closed.
|
||||
- Cross-service names are never accepted by the wrong service
|
||||
(`jenkins.read` never matches a Gitea check, and a Gitea alias is never
|
||||
applied to another service).
|
||||
- `forbidden_operations` overrides `allowed_operations` after both sides are
|
||||
normalized, so a legacy spelling can never bypass a canonical forbidden
|
||||
entry (or vice versa).
|
||||
- An allowed entry that cannot be normalized grants nothing; a forbidden
|
||||
entry that cannot be normalized denies the request. Normalization can
|
||||
therefore never silently widen permissions.
|
||||
- An empty or missing `allowed_operations` list denies everything.
|
||||
|
||||
## Identity and fail-closed rules
|
||||
|
||||
|
||||
@@ -339,22 +339,25 @@ touching anything.
|
||||
- **Steps:** confirm eligibility; require explicit confirmation
|
||||
(`MERGE PR <n>`); optionally pin head SHA / changed-file set; merge only when
|
||||
Gitea reports the PR mergeable (branch-protection checks satisfied). No force,
|
||||
no ignore-checks.
|
||||
no ignore-checks. Verify that remote master contains the merge commit or the expected squashed changes (do not assume a "closed" PR succeeded without verifying the actual landed changes).
|
||||
- **Prompt:** `Use any eligible merger profile to merge PR #N if checks pass and
|
||||
it is mergeable. Confirm with "MERGE PR N". Do not force-merge.`
|
||||
|
||||
### Close the issue after merge / Reconciliation
|
||||
|
||||
- **Profile:** issue-manager or merger.
|
||||
- **Steps:** verify remote `master` actually contains the merge; close the
|
||||
issue; release `status:in-progress` (if it cannot be removed, report why).
|
||||
- **Steps:** Verify remote `master` actually contains the merge (post-merge file-presence verification):
|
||||
- Run: `git fetch <remote> --prune; git checkout master; git pull <remote> master --ff-only`
|
||||
- Verify that expected files added/modified in the PR are present on `master` (or absent if deleted).
|
||||
- Alternatively, verify with: `git log --oneline -- <expected-file>` or `git merge-base --is-ancestor <pr-head-sha> master`
|
||||
- Close the issue; release `status:in-progress` (if it cannot be removed, report why).
|
||||
- **If closed but not merged (`merged=false`):** Stop normal flow. Do not delete worktrees. Compare PR content to remote `master`.
|
||||
- **fully landed:** comment it landed, remove `status:in-progress`, clean up.
|
||||
- **partially landed:** reopen issue, create corrective PR for missing pieces.
|
||||
- **not landed:** reopen issue/PR, do not clean up.
|
||||
- **Direct push to master:** is forbidden except as a documented recovery exception. Final reports must include why, commits, PR metadata, and repaired labels.
|
||||
- **Final reports:** must include both PR metadata (state, merged flag, merge commit) and Git content (remote master hash, expected content present).
|
||||
- **Prompt (normal):** `After confirming master contains the merge of PR #N, close issue #M and delete the merged branch.`
|
||||
- **Final reports:** must include both PR metadata (state, merged flag, merge commit) and Git content (remote master hash, expected content present, verification method used & results).
|
||||
- **Prompt (normal):** `After verifying master contains the merge of PR #N using post-merge file-presence verification, close issue #M and delete the merged branch. Include verification details in the report.`
|
||||
- **Prompt (reconcile):** `Reconcile closed-not-merged PR #N by verifying if its content landed on master.`
|
||||
|
||||
### Stop on blocker
|
||||
@@ -364,28 +367,37 @@ touching anything.
|
||||
files, detected secret, or any production/deploy behavior — **stop, report the
|
||||
blocker, and take no mutating action.** Fail closed; never work around a gate.
|
||||
|
||||
## Controller Handoff Summary (required, every task)
|
||||
## Controller Handoff (required, every task)
|
||||
|
||||
Every task — implementation, review, merge, triage, documentation,
|
||||
discussion-only, or blocked planning — **must end with a
|
||||
`Controller Handoff Summary`** so a controller LLM can pick up the state
|
||||
without rereading the conversation. The canonical format and rules live in the
|
||||
portable skill:
|
||||
`Controller Handoff`** so a controller LLM can pick up the state
|
||||
without rereading the conversation. The canonical formats and rules live in
|
||||
the portable skill:
|
||||
[`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) §K.
|
||||
|
||||
Sections (in order): Work performed · Current state (repo, branch/master
|
||||
commit, issue #s, PR #s, complete/blocked/ready-for-review/discussion-only) ·
|
||||
Files changed · Validation · Issues encountered · Review needed? (one of the
|
||||
five fixed answers) · Next recommended action · Safety confirmations
|
||||
(no self-review; no self-merge; no release/tag changes unless requested; no
|
||||
secrets; no production access unless authorized).
|
||||
**Compact format is the default** — nine lines (`Task / Repo/state /
|
||||
Issues/PRs / Changed / Validation / Blockers / Review / Next / Safety`),
|
||||
written for controller-LLM readability, not a full human status report. The
|
||||
`Safety:` line is never omitted (usually
|
||||
`no self-review; no self-merge; no tags; no secrets; no prod`). PR bodies
|
||||
still carry the full review detail — the handoff never replaces PR
|
||||
documentation.
|
||||
|
||||
**The long form** (Work performed · Current state · Files changed ·
|
||||
Validation · Issues encountered · Review needed? · Next recommended action ·
|
||||
Safety confirmations) **is reserved for high-risk or complex tasks**: a
|
||||
merge/tag/release happened, validation failed, permissions/profile gates
|
||||
blocked work, secrets or production access were involved, an owner decision
|
||||
is complicated, the task spanned multiple repos or cross-issue state, or the
|
||||
owner explicitly asks for it.
|
||||
|
||||
Hard rules: never omit it; never bury blockers earlier only; an opened PR
|
||||
means "Review needed — PR is open"; a blocked merge names the exact gate;
|
||||
discussion-only comments need owner/design feedback, not code review; any
|
||||
touched release state names the exact tag/commit and why. Design debates
|
||||
belong in **discussion/RFC issues** (e.g. #100 `profiles.json v2`) — comment
|
||||
on the issue, create no branches/PRs, and end the comment with this summary.
|
||||
on the issue, create no branches/PRs, and end the comment with this handoff.
|
||||
|
||||
## Fail-closed behavior
|
||||
|
||||
@@ -411,6 +423,9 @@ with the profile and authenticated user when `GITEA_AUDIT_LOG` is set (see
|
||||
|
||||
## Releases and version tags
|
||||
|
||||
All release tagging, version bumps, and validation must comply with the [Release / Version Process SOP](release-version-sop.md).
|
||||
|
||||
|
||||
Versions follow SemVer — **`vMAJOR.MINOR.PATCH`**, using **`v0.x.y`** while
|
||||
unstable. Pick the bump by the largest change since the last tag:
|
||||
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
# Release / Version Process SOP
|
||||
|
||||
Operator standard operating procedure for cutting a versioned release of
|
||||
Gitea-Tools: version bump, checks, merge, tag, and cleanup.
|
||||
|
||||
> **Scope.** This is the **human/operator** SOP. It is deliberately distinct
|
||||
> from [`release-workflows.md`](release-workflows.md), which describes the
|
||||
> **future `release-mcp` orchestrator** boundary (a coordination concept), not
|
||||
> the day-to-day tagging process. When they disagree, this document governs how
|
||||
> a release is actually cut today.
|
||||
|
||||
---
|
||||
|
||||
## 1. Branch flow
|
||||
|
||||
The repo is **`master`-based**. Releases are cut from `master`; there is no
|
||||
separate `dev`/`release` branch unless and until that is explicitly introduced
|
||||
and this SOP is updated to match. All work lands on `master` via reviewed PRs
|
||||
from short-lived, issue-linked branches (e.g. `docs/issue-68-...`).
|
||||
|
||||
## 2. Where "the version" lives
|
||||
|
||||
There is **no `VERSION` file and no `CHANGELOG` file** in the repo today. The
|
||||
released version is expressed **only as an annotated git tag** of the form
|
||||
`vMAJOR.MINOR.PATCH` (existing tags: `v1.0.0`, `v1.0.1`). Release notes are
|
||||
carried as the **annotated tag's message** (via `--notes-file`), not a tracked
|
||||
changelog.
|
||||
|
||||
> Do **not** confuse this with `SUPPORTED_VERSION` in `gitea_config.py` — that is
|
||||
> the **config-schema** version, unrelated to the application release version.
|
||||
|
||||
If a `VERSION`/`CHANGELOG` file is added later, update this SOP to list it under
|
||||
"files to update".
|
||||
|
||||
## 3. Deciding the version bump (SemVer)
|
||||
|
||||
Pick the bump against the last tag using semantic-versioning intent:
|
||||
|
||||
* **PATCH** (`v1.0.1 → v1.0.2`): bug fixes, docs, tests, internal cleanups — no
|
||||
change to tool names, parameters, return payloads, or behavior.
|
||||
* **MINOR** (`v1.0.1 → v1.1.0`): backward-compatible additions — new MCP tool,
|
||||
new optional parameter, new script, additive behavior.
|
||||
* **MAJOR** (`v1.1.0 → v2.0.0`): backward-**incompatible** changes — renamed or
|
||||
removed tools, changed return-payload shape, changed default behavior, or a
|
||||
tightened safety gate that rejects previously-accepted input.
|
||||
|
||||
When unsure between two levels, choose the higher one.
|
||||
|
||||
## 4. Preparing a version-bump / release PR
|
||||
|
||||
Releases are still gated by the normal issue-first, PR-reviewed flow.
|
||||
|
||||
1. Open (or use) a tracking issue for the release and **claim it** with
|
||||
`status:in-progress` (see §9).
|
||||
2. Create an isolated, issue-linked branch + worktree from latest `master`
|
||||
(e.g. `chore/issue-63-v1.1.0`). Never commit directly to `master`.
|
||||
3. Include in the PR:
|
||||
* Any code/docs changes that belong to the release.
|
||||
* The **release notes** for the annotated tag (draft them in the PR body or a
|
||||
notes file you will pass to `scripts/release-tag --notes-file`).
|
||||
* If a `VERSION`/`CHANGELOG` file exists at that time, its update.
|
||||
4. Open the PR **targeting `master`**.
|
||||
|
||||
The tag is **not** created in the PR. Tagging happens only after merge (§6).
|
||||
|
||||
## 5. Required checks before release
|
||||
|
||||
Run all of these green before merging the release PR and before tagging:
|
||||
|
||||
```bash
|
||||
python3 -m py_compile mcp_server.py
|
||||
python3 -m py_compile manage_labels.py
|
||||
bash -n scripts/clear-provenance
|
||||
./venv/bin/python -m pytest tests/ -q
|
||||
git diff --check
|
||||
```
|
||||
|
||||
Plus a secret sweep (there is no third-party scanner wired in; do a staged-diff
|
||||
sweep — see [`developer-testing-guidelines.md`](developer-testing-guidelines.md)
|
||||
§7):
|
||||
|
||||
```bash
|
||||
git diff --cached | grep -nEi "authorization: (basic|bearer)|password[:=]|token=[A-Za-z0-9]" || echo "clean"
|
||||
```
|
||||
|
||||
`scripts/release-tag` **also** runs the test suite itself before tagging (unless
|
||||
`--skip-tests` is passed), so tests are enforced twice by default.
|
||||
|
||||
## 6. Running `scripts/release-tag`
|
||||
|
||||
Tag **only after** the release PR is merged to `master`. `scripts/release-tag`
|
||||
enforces the tagging policy and is **safe by default** (creates nothing on a
|
||||
dry-run; never pushes without `--push`).
|
||||
|
||||
Before it tags, it requires **all** of:
|
||||
|
||||
* version matches `vMAJOR.MINOR.PATCH` (SemVer);
|
||||
* `fetch --prune` has run;
|
||||
* you are **on `master`**;
|
||||
* the worktree is **clean** (no uncommitted changes);
|
||||
* local `master` **equals** `<remote>/master`;
|
||||
* `HEAD` is that same commit (the commit is present on remote master);
|
||||
* the tag does **not** already exist locally or on the remote;
|
||||
* the test suite passes (unless `--skip-tests`, which warns).
|
||||
|
||||
Typical sequence:
|
||||
|
||||
```bash
|
||||
# 1. Dry-run to confirm the plan (changes nothing)
|
||||
scripts/release-tag --dry-run v1.1.0
|
||||
|
||||
# 2. Create the annotated tag locally, with release notes
|
||||
scripts/release-tag v1.1.0 --notes-file /path/to/release-notes.md
|
||||
|
||||
# 3. Push the tag only when ready
|
||||
scripts/release-tag v1.1.0 --notes-file /path/to/release-notes.md --push
|
||||
```
|
||||
|
||||
Env injection points (mainly for CI/tests):
|
||||
`RELEASE_TAG_REMOTE` (default `prgs`), `RELEASE_TAG_TEST_CMD`
|
||||
(default `./venv/bin/python -m pytest tests/ -q`).
|
||||
|
||||
## 7. Who may merge / tag
|
||||
|
||||
* The release PR must be **merged by someone other than its author** — the
|
||||
author-cannot-merge safety gate applies to releases exactly as to any other PR.
|
||||
* Merge uses the gated `gitea_merge_pr` workflow; CLI/legacy merge is disabled.
|
||||
* Whoever tags must operate on clean master synced to the remote (enforced by
|
||||
`scripts/release-tag`). Tagging is an operator action performed after merge.
|
||||
|
||||
## 8. Self-review / self-merge restrictions
|
||||
|
||||
Release PRs are **not** exempt from the safety model:
|
||||
|
||||
* No self-review — the author may not approve their own release PR.
|
||||
* No self-merge — a different eligible identity merges.
|
||||
* These gates are enforced by the MCP tooling and must not be bypassed.
|
||||
|
||||
## 9. Handling `status:in-progress` during release work
|
||||
|
||||
* **Claim** the release tracking issue with `status:in-progress` before starting.
|
||||
* Keep it claimed while the release PR is open and under review.
|
||||
* On merge/close, the tracker-hygiene automation releases `status:in-progress`
|
||||
for issues the PR closes; if it remains after the release lands, release it
|
||||
explicitly. Do not leave a shipped release issue marked in-progress.
|
||||
|
||||
## 10. Branch / worktree cleanup after merge
|
||||
|
||||
After the release PR merges and the tag is pushed:
|
||||
|
||||
* Delete the remote release branch (if repo policy allows).
|
||||
* Remove the local worktree and delete the local branch:
|
||||
|
||||
```bash
|
||||
git worktree remove branches/<release-worktree>
|
||||
git branch -d <release-branch>
|
||||
git worktree prune
|
||||
```
|
||||
* Confirm the root repo is clean and on `master` synced to the remote.
|
||||
|
||||
## 11. What NOT to do
|
||||
|
||||
* **No direct commits to `master`.** All changes land via reviewed PRs.
|
||||
* **No force-push** (to `master` or to tags).
|
||||
* **No self-merge** of a release PR.
|
||||
* **No tagging before merge** — tag only commits already on remote `master`.
|
||||
* **No release from a dirty worktree** — `scripts/release-tag` refuses, and so
|
||||
should you.
|
||||
* **No `--skip-tests`** for a real release unless there is an explicit,
|
||||
documented reason.
|
||||
* **No re-tagging / moving an existing tag** — pick the next version instead.
|
||||
|
||||
## 12. Post-Merge Verification & Audit Lessons (v1.1.0)
|
||||
|
||||
During the v1.1.0 release audit, we identified a critical reconciliation issue (captured in historical PRs/issues #68 and #82):
|
||||
* **The "Closed" State Trap:** Gitea PRs marked as `closed` are not guaranteed to be `merged` (they can be closed without merging, leading to silent omissions of code/documentation changes).
|
||||
* **Mandatory Post-Merge File/Commit Presence Probe:** Reviewers/mergers must perform explicit post-merge validation. Do not assume a merge succeeded.
|
||||
- Check that the merged branch head is an ancestor of the target branch (`master`):
|
||||
```bash
|
||||
git fetch <remote> --prune
|
||||
git merge-base --is-ancestor <pr-head-sha> <remote>/master
|
||||
```
|
||||
- Probe file presence for expected modifications/additions:
|
||||
```bash
|
||||
git log --oneline -- <expected-file>
|
||||
# and confirm file presence:
|
||||
ls -la docs/release-version-sop.md
|
||||
```
|
||||
* **Verify in Handoff:** Final report blocks must explicitly document the verification method and probe results.
|
||||
@@ -0,0 +1,80 @@
|
||||
{
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"example-context": {
|
||||
"enabled": true,
|
||||
"label": "Example environment",
|
||||
"description": "One deployment environment: its Gitea plus non-Gitea services.",
|
||||
"default_owner": "Example-Org",
|
||||
"gitea": {
|
||||
"enabled": true,
|
||||
"kind": "gitea",
|
||||
"base_url": "https://gitea.example.invalid"
|
||||
},
|
||||
"services": {
|
||||
"jenkins": {
|
||||
"enabled": true,
|
||||
"kind": "jenkins",
|
||||
"label": "Example Jenkins",
|
||||
"base_url": "https://jenkins.example.invalid",
|
||||
"auth": { "type": "keychain", "id": "example-jenkins-token" },
|
||||
"capabilities": ["read"]
|
||||
},
|
||||
"glitchtip": {
|
||||
"enabled": false,
|
||||
"kind": "glitchtip",
|
||||
"label": "Example GlitchTip (disabled: defined but unavailable)",
|
||||
"base_url": "",
|
||||
"auth": { "type": "keychain", "id": "example-glitchtip-token" },
|
||||
"capabilities": ["read"],
|
||||
"allow_raw_events": false
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"example-author": {
|
||||
"enabled": true,
|
||||
"context": "example-context",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"execution_profile": "example-author",
|
||||
"audit_label": "example-author",
|
||||
"auth": { "type": "keychain", "id": "example-gitea-author-token" },
|
||||
"allowed_operations": ["read", "branch", "commit", "push", "open_pr", "comment"],
|
||||
"forbidden_operations": ["approve", "request_changes", "merge"]
|
||||
},
|
||||
"example-reviewer": {
|
||||
"enabled": true,
|
||||
"context": "example-context",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"execution_profile": "example-reviewer",
|
||||
"audit_label": "example-reviewer",
|
||||
"auth": { "type": "keychain", "id": "example-gitea-reviewer-token" },
|
||||
"allowed_operations": ["read", "review", "comment", "approve", "request_changes", "merge"],
|
||||
"forbidden_operations": ["branch", "commit", "push", "open_pr"]
|
||||
}
|
||||
},
|
||||
"projects": {
|
||||
"/absolute/path/to/local/repo": {
|
||||
"enabled": true,
|
||||
"context": "example-context",
|
||||
"default_owner": "Example-Org",
|
||||
"default_repo": "Example-Repo",
|
||||
"default_author_profile": "example-author",
|
||||
"default_reviewer_profile": "example-reviewer"
|
||||
}
|
||||
},
|
||||
"rules": {
|
||||
"disabled_behavior": "Defined but unavailable for action. MCP tools may report disabled entries during audits, but must not use them automatically.",
|
||||
"no_silent_fallback": true,
|
||||
"tokens_in_json": false,
|
||||
"token_storage": "keychain",
|
||||
"identity_must_match_task": true,
|
||||
"same_username_cannot_review_own_pr": true,
|
||||
"hide_service_urls_from_llm": true,
|
||||
"hide_keychain_ids_from_llm": true,
|
||||
"mcp_resolves_endpoints": true
|
||||
}
|
||||
}
|
||||
+7
-3
@@ -123,13 +123,17 @@ def get_auth_header(host):
|
||||
token = os.environ.get("GITEA_TOKEN")
|
||||
|
||||
# 3. Fall back to a JSON runtime-profile token reference (token_env).
|
||||
# Explicit env tokens above take precedence. A broken config never breaks
|
||||
# auth here — it fails closed to "no token"; the clear error surfaces via
|
||||
# get_profile() / startup instead.
|
||||
# Explicit env tokens above take precedence. When GITEA_MCP_CONFIG is
|
||||
# configured, a broken config or unresolvable profile/credential fails
|
||||
# closed here (no silent fallback to Basic auth or another source,
|
||||
# #120). Without a configured JSON layer, env-only behaviour is
|
||||
# unchanged.
|
||||
if not token:
|
||||
try:
|
||||
token = gitea_config.resolve_token(gitea_config.resolve_profile())
|
||||
except gitea_config.ConfigError:
|
||||
if gitea_config.config_path():
|
||||
raise
|
||||
token = None
|
||||
|
||||
if token:
|
||||
|
||||
+692
-10
@@ -54,11 +54,125 @@ ENV_CONFIG_PATH = "GITEA_MCP_CONFIG"
|
||||
ENV_PROFILE = "GITEA_MCP_PROFILE"
|
||||
|
||||
SUPPORTED_VERSION = 1
|
||||
SUPPORTED_VERSIONS = (1, 2)
|
||||
_AUTH_TYPES = ("keychain", "env")
|
||||
|
||||
# Profile names go into env vars, keychain ids, and JSON keys — keep them tame.
|
||||
_PROFILE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
|
||||
|
||||
# v2 address segments (environment / service / identity) must be dot-free so
|
||||
# the dotted profile address {env}.{service}.{identity} stays unambiguous.
|
||||
_SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$")
|
||||
|
||||
# Placeholder usernames must never activate (fail closed until provisioned).
|
||||
_TBD_RE = re.compile(r"(?i)^tbd(-|$)")
|
||||
|
||||
# Keys that would mean an inline secret wherever they appear.
|
||||
_INLINE_SECRET_KEYS = ("token", "password", "secret")
|
||||
|
||||
# ── Operation-name normalization table (#106; minimal subset landed in #103) ───
|
||||
# Canonical operations are namespaced ({service}.{area}.{verb}). Legacy
|
||||
# unqualified spellings are accepted ONLY through this explicit table — never
|
||||
# by guessing. The same table is the documentation of record (see
|
||||
# docs/gitea-execution-profiles.md) and is exercised by
|
||||
# tests/test_op_normalization.py.
|
||||
GITEA_OPERATION_ALIASES = {
|
||||
"read": "gitea.read",
|
||||
"review": "gitea.pr.review",
|
||||
"comment": "gitea.pr.comment",
|
||||
"approve": "gitea.pr.approve",
|
||||
"request_changes": "gitea.pr.request_changes",
|
||||
"merge": "gitea.pr.merge",
|
||||
"pr.create": "gitea.pr.create",
|
||||
"branch.push": "gitea.branch.push",
|
||||
# Contexts-shape author verbs (#120) — the invariant checks below depend on
|
||||
# "push"/"open_pr" normalizing to the two author-only ops.
|
||||
"branch": "gitea.branch.create",
|
||||
"commit": "gitea.repo.commit",
|
||||
"push": "gitea.branch.push",
|
||||
"open_pr": "gitea.pr.create",
|
||||
}
|
||||
_REVIEW_MERGE_OPS = frozenset({"gitea.pr.approve", "gitea.pr.merge"})
|
||||
_AUTHOR_ONLY_OPS = frozenset({"gitea.pr.create", "gitea.branch.push"})
|
||||
|
||||
|
||||
def normalize_operation(op, service="gitea"):
|
||||
"""Return the canonical namespaced name for *op*, or fail closed (#106).
|
||||
|
||||
- already namespaced for this service (``{service}.*``) → unchanged
|
||||
- known unqualified Gitea ops → mapped via ``GITEA_OPERATION_ALIASES``
|
||||
- unqualified single-word ops on non-Gitea services → ``{service}.{op}``
|
||||
- anything else — foreign service prefixes, dotted names outside the
|
||||
table, unknown unqualified names — is unknown or ambiguous → ConfigError
|
||||
|
||||
Normalization never crosses services (a Gitea alias is never applied to
|
||||
another service) and never widens permissions: an operation that cannot
|
||||
be normalized grants and matches nothing.
|
||||
"""
|
||||
if not isinstance(op, str) or not op:
|
||||
raise ConfigError("operation must be a non-empty string (fail closed)")
|
||||
if op.startswith(service + "."):
|
||||
return op
|
||||
if service == "gitea" and op in GITEA_OPERATION_ALIASES:
|
||||
return GITEA_OPERATION_ALIASES[op]
|
||||
if service != "gitea" and "." not in op:
|
||||
return f"{service}.{op}"
|
||||
raise ConfigError(
|
||||
f"operation {op!r} cannot be normalized safely for service "
|
||||
f"'{service}' (unknown, ambiguous, or cross-service; fail closed)"
|
||||
)
|
||||
|
||||
|
||||
def check_operation(op, allowed, forbidden=(), service="gitea"):
|
||||
"""Decide whether *op* is permitted. Returns ``(bool, reason)`` (#106).
|
||||
|
||||
Everything is normalized via :func:`normalize_operation` BEFORE any
|
||||
membership check, so legacy and canonical spellings always compare equal.
|
||||
Reasons: ``allowed``, ``invalid-operation``, ``invalid-forbidden-entry``,
|
||||
``forbidden``, ``no-allowed-operations``, ``not-allowed``.
|
||||
|
||||
Fail-closed rules:
|
||||
- an *op* that cannot be normalized is denied (``invalid-operation``)
|
||||
- a forbidden entry that cannot be normalized denies the request
|
||||
(``invalid-forbidden-entry``) — dropping it would silently narrow the
|
||||
forbidden set, i.e. widen permissions
|
||||
- an allowed entry that cannot be normalized is ignored — it grants
|
||||
nothing, so permissions never widen
|
||||
- ``forbidden`` always overrides ``allowed``
|
||||
- an empty or missing allowed list denies everything
|
||||
"""
|
||||
try:
|
||||
op_n = normalize_operation(op, service)
|
||||
except ConfigError:
|
||||
return (False, "invalid-operation")
|
||||
forbidden_n = set()
|
||||
for entry in (forbidden or ()):
|
||||
try:
|
||||
forbidden_n.add(normalize_operation(entry, service))
|
||||
except ConfigError:
|
||||
return (False, "invalid-forbidden-entry")
|
||||
if op_n in forbidden_n:
|
||||
return (False, "forbidden")
|
||||
if not allowed:
|
||||
return (False, "no-allowed-operations")
|
||||
allowed_n = set()
|
||||
for entry in allowed:
|
||||
try:
|
||||
allowed_n.add(normalize_operation(entry, service))
|
||||
except ConfigError:
|
||||
continue
|
||||
if op_n in allowed_n:
|
||||
return (True, "allowed")
|
||||
return (False, "not-allowed")
|
||||
|
||||
|
||||
def _normalize_op(service, op, addr):
|
||||
"""Normalize *op* for identity *addr*, or fail closed with context."""
|
||||
try:
|
||||
return normalize_operation(op, service)
|
||||
except ConfigError as exc:
|
||||
raise ConfigError(f"identity '{addr}': {exc}") from None
|
||||
|
||||
# Default canonical config location (one file shared by all LLM launchers).
|
||||
DEFAULT_CONFIG_PATH = os.path.join(
|
||||
os.path.expanduser("~"), ".config", "gitea-tools", "profiles.json"
|
||||
@@ -108,16 +222,550 @@ def load_config(path=None):
|
||||
) from None
|
||||
except OSError as exc:
|
||||
raise ConfigError(f"could not read {path}: {exc.strerror}") from None
|
||||
if not isinstance(data, dict) or not isinstance(data.get("profiles"), dict):
|
||||
raise ConfigError(f"{path} must be a JSON object with a 'profiles' object")
|
||||
version = data.get("version", SUPPORTED_VERSION)
|
||||
if not isinstance(data, dict):
|
||||
raise ConfigError(f"{path} must be a JSON object")
|
||||
version = data.get("version")
|
||||
if version is None:
|
||||
# Fail closed (#103): an unversioned config is ambiguous between v1 and
|
||||
# v2 shapes, so it is refused rather than guessed.
|
||||
raise ConfigError(
|
||||
f"{path} is missing the required 'version' field; "
|
||||
f"expected one of {list(SUPPORTED_VERSIONS)}"
|
||||
)
|
||||
if version == 2:
|
||||
return _load_v2_any(data, path)
|
||||
if version != SUPPORTED_VERSION:
|
||||
raise ConfigError(
|
||||
f"{path} has unsupported version {version!r}; expected {SUPPORTED_VERSION}"
|
||||
f"{path} has unsupported version {version!r}; "
|
||||
f"expected one of {list(SUPPORTED_VERSIONS)}"
|
||||
)
|
||||
if not isinstance(data.get("profiles"), dict):
|
||||
raise ConfigError(f"{path} must be a JSON object with a 'profiles' object")
|
||||
return data
|
||||
|
||||
|
||||
# ── profiles.json version 2 (#103): environment → service → identity ──────────
|
||||
# v2 files are validated and *flattened* at load time into the same
|
||||
# {"profiles": {...}} shape v1 consumers already understand, keyed by the
|
||||
# canonical dotted address {environment}.{service}.{identity}. Two extra
|
||||
# top-level keys are carried: "aliases" (exact-name compatibility selectors)
|
||||
# and "unavailable" (addresses that fail closed at selection, e.g. TBD users).
|
||||
|
||||
def _validate_identity_auth(addr, auth):
|
||||
"""Require and validate an identity 'auth' reference. Rejects inline secrets."""
|
||||
if auth is None:
|
||||
raise ConfigError(f"identity '{addr}' is missing an 'auth' reference")
|
||||
if not isinstance(auth, dict):
|
||||
raise ConfigError(f"identity '{addr}' has a non-object 'auth'")
|
||||
for key in _INLINE_SECRET_KEYS:
|
||||
if key in auth:
|
||||
raise ConfigError(
|
||||
f"identity '{addr}' auth must not contain an inline '{key}'; "
|
||||
"store secrets in the keychain and reference them by id"
|
||||
)
|
||||
_validate_auth(addr, auth)
|
||||
|
||||
|
||||
def _flatten_identity(env_name, svc_name, svc, ident_name, ident):
|
||||
"""Validate one v2 identity and return (addr, flattened_profile).
|
||||
|
||||
The flattened profile is v1-shaped (base_url/auth/username/defaults) plus
|
||||
v2 metadata (profile_path, environment, service, identity, role) and
|
||||
normalized operation lists. Raises ConfigError on any invariant violation.
|
||||
"""
|
||||
addr = f"{env_name}.{svc_name}.{ident_name}"
|
||||
if not isinstance(ident, dict):
|
||||
raise ConfigError(f"identity '{addr}' must be a JSON object")
|
||||
for key in _INLINE_SECRET_KEYS:
|
||||
if key in ident:
|
||||
raise ConfigError(
|
||||
f"identity '{addr}' must not contain an inline '{key}'; "
|
||||
"use an 'auth' reference instead"
|
||||
)
|
||||
_validate_identity_auth(addr, ident.get("auth"))
|
||||
|
||||
base_url = ident.get("base_url") or svc.get("base_url")
|
||||
if not base_url:
|
||||
raise ConfigError(
|
||||
f"identity '{addr}' has no 'base_url' at identity or service level"
|
||||
)
|
||||
|
||||
allowed = ident.get("allowed_operations") or []
|
||||
forbidden = ident.get("forbidden_operations") or []
|
||||
if not isinstance(allowed, list) or not isinstance(forbidden, list):
|
||||
raise ConfigError(f"identity '{addr}' operation fields must be lists")
|
||||
allowed_n = {_normalize_op(svc_name, op, addr) for op in allowed}
|
||||
forbidden_n = {_normalize_op(svc_name, op, addr) for op in forbidden}
|
||||
|
||||
# Reviewer-identity deadlock rule (#100/#103): an identity that may approve
|
||||
# or merge PRs must explicitly forbid creating PRs and pushing branches,
|
||||
# so the reviewer identity can never author the PR it must review.
|
||||
if allowed_n & _REVIEW_MERGE_OPS:
|
||||
missing = sorted(_AUTHOR_ONLY_OPS - forbidden_n)
|
||||
if missing:
|
||||
raise ConfigError(
|
||||
f"identity '{addr}' allows PR approve/merge but does not forbid "
|
||||
f"{missing}; reviewer identities must forbid gitea.pr.create and "
|
||||
"gitea.branch.push (reviewer-identity deadlock rule)"
|
||||
)
|
||||
|
||||
profile = {
|
||||
"profile_path": addr,
|
||||
"environment": env_name,
|
||||
"service": svc_name,
|
||||
"identity": ident_name,
|
||||
"base_url": base_url,
|
||||
"auth": ident["auth"],
|
||||
"allowed_operations": sorted(allowed_n),
|
||||
"forbidden_operations": sorted(forbidden_n),
|
||||
}
|
||||
# Service-level defaults inherit unless the identity overrides them.
|
||||
for key in ("default_owner", "default_repo", "default_org"):
|
||||
value = ident.get(key, svc.get(key))
|
||||
if value:
|
||||
profile[key] = value
|
||||
for key in ("role", "username", "execution_profile", "audit_label"):
|
||||
if ident.get(key):
|
||||
profile[key] = ident[key]
|
||||
return addr, profile
|
||||
|
||||
|
||||
def _load_v2(data, path):
|
||||
"""Validate a v2 config and return the flattened, resolvable structure."""
|
||||
environments = data.get("environments")
|
||||
if not isinstance(environments, dict) or not environments:
|
||||
raise ConfigError(
|
||||
f"{path} version 2 config requires a non-empty 'environments' object"
|
||||
)
|
||||
profiles = {}
|
||||
unavailable = {}
|
||||
for env_name, env in environments.items():
|
||||
if not _SEGMENT_RE.match(env_name or ""):
|
||||
raise ConfigError(f"invalid environment name {env_name!r} (no dots)")
|
||||
if not isinstance(env, dict):
|
||||
raise ConfigError(f"environment '{env_name}' must be a JSON object")
|
||||
services = env.get("services")
|
||||
if not isinstance(services, dict) or not services:
|
||||
raise ConfigError(
|
||||
f"environment '{env_name}' requires a non-empty 'services' object"
|
||||
)
|
||||
for svc_name, svc in services.items():
|
||||
if not _SEGMENT_RE.match(svc_name or ""):
|
||||
raise ConfigError(
|
||||
f"invalid service name {svc_name!r} in '{env_name}' (no dots)"
|
||||
)
|
||||
if not isinstance(svc, dict):
|
||||
raise ConfigError(
|
||||
f"service '{env_name}.{svc_name}' must be a JSON object"
|
||||
)
|
||||
identities = svc.get("identities")
|
||||
if not isinstance(identities, dict) or not identities:
|
||||
raise ConfigError(
|
||||
f"service '{env_name}.{svc_name}' requires a non-empty "
|
||||
"'identities' object"
|
||||
)
|
||||
for ident_name, ident in identities.items():
|
||||
if not _SEGMENT_RE.match(ident_name or ""):
|
||||
raise ConfigError(
|
||||
f"invalid identity name {ident_name!r} in "
|
||||
f"'{env_name}.{svc_name}' (no dots)"
|
||||
)
|
||||
addr, profile = _flatten_identity(
|
||||
env_name, svc_name, svc, ident_name, ident
|
||||
)
|
||||
username = profile.get("username") or ""
|
||||
if _TBD_RE.match(username):
|
||||
# Fail closed at selection, without blocking every other
|
||||
# identity in the file (see #103 acceptance criteria).
|
||||
unavailable[addr] = (
|
||||
f"identity '{addr}' username {username!r} is a TBD "
|
||||
"placeholder; provision the account before use "
|
||||
"(fail closed)"
|
||||
)
|
||||
else:
|
||||
profiles[addr] = profile
|
||||
|
||||
aliases = data.get("aliases") or {}
|
||||
if not isinstance(aliases, dict):
|
||||
raise ConfigError(f"{path} 'aliases' must be a JSON object")
|
||||
known = set(profiles) | set(unavailable)
|
||||
for alias, target in aliases.items():
|
||||
if not isinstance(target, str) or not target:
|
||||
raise ConfigError(f"alias '{alias}' target must be a non-empty string")
|
||||
if alias in known and alias != target:
|
||||
raise ConfigError(
|
||||
f"selector '{alias}' is both an alias and a profile address "
|
||||
"with a different target (conflicting selector; fail closed)"
|
||||
)
|
||||
if target not in known:
|
||||
raise ConfigError(
|
||||
f"alias '{alias}' points to unknown profile '{target}'"
|
||||
)
|
||||
return {
|
||||
"version": 2,
|
||||
"profiles": profiles,
|
||||
"aliases": dict(aliases),
|
||||
"unavailable": unavailable,
|
||||
}
|
||||
|
||||
|
||||
# ── profiles.json version 2 *contexts* shape (#120) ───────────────────────────
|
||||
# The canonical machine config groups everything by context: top-level
|
||||
# "contexts" (each with a gitea block and non-Gitea "services"), flat
|
||||
# "profiles" (Gitea identities pointing at a context), "projects" (local repo
|
||||
# paths mapped to a context), and "rules". Every context/profile/service/
|
||||
# project carries a required boolean "enabled": disabled entries are surfaced
|
||||
# in audits but fail closed at selection — never a silent fallback. Loading
|
||||
# flattens profiles into the same {"profiles": {...}, "unavailable": {...}}
|
||||
# model v1 consumers and select_profile() already understand, and carries the
|
||||
# validated "contexts"/"projects"/"rules" through for service resolution.
|
||||
|
||||
def _load_v2_any(data, path):
|
||||
"""Dispatch a version-2 file to its shape loader; ambiguity fails closed."""
|
||||
has_contexts = "contexts" in data
|
||||
has_environments = "environments" in data
|
||||
if has_contexts and has_environments:
|
||||
raise ConfigError(
|
||||
f"{path} version 2 config must not mix 'contexts' and "
|
||||
"'environments' shapes (ambiguous; fail closed)"
|
||||
)
|
||||
if has_contexts:
|
||||
return _load_v2_contexts(data, path)
|
||||
return _load_v2(data, path)
|
||||
|
||||
|
||||
def _require_enabled(kind, name, obj):
|
||||
"""Return the required boolean ``enabled`` flag, failing closed."""
|
||||
enabled = obj.get("enabled")
|
||||
if not isinstance(enabled, bool):
|
||||
raise ConfigError(
|
||||
f"{kind} '{name}' requires a boolean 'enabled' flag (fail closed)"
|
||||
)
|
||||
return enabled
|
||||
|
||||
|
||||
def _reject_inline_secrets(kind, name, obj):
|
||||
for key in _INLINE_SECRET_KEYS:
|
||||
if key in obj:
|
||||
raise ConfigError(
|
||||
f"{kind} '{name}' must not contain an inline '{key}'; "
|
||||
"store secrets in the keychain and reference them by id"
|
||||
)
|
||||
|
||||
|
||||
def _validate_context_service(ctx_name, svc_name, svc):
|
||||
"""Validate one context service entry (auth reference only, no secrets)."""
|
||||
addr = f"{ctx_name}.{svc_name}"
|
||||
if not isinstance(svc, dict):
|
||||
raise ConfigError(f"service '{addr}' must be a JSON object")
|
||||
_require_enabled("service", addr, svc)
|
||||
_reject_inline_secrets("service", addr, svc)
|
||||
if "auth" in svc:
|
||||
_validate_auth(addr, svc["auth"])
|
||||
|
||||
|
||||
def _load_v2_contexts(data, path):
|
||||
"""Validate a v2 contexts-shape config and return the resolvable structure."""
|
||||
contexts = data.get("contexts")
|
||||
if not isinstance(contexts, dict) or not contexts:
|
||||
raise ConfigError(
|
||||
f"{path} version 2 contexts config requires a non-empty "
|
||||
"'contexts' object"
|
||||
)
|
||||
for ctx_name, ctx in contexts.items():
|
||||
if not _PROFILE_NAME_RE.match(ctx_name or ""):
|
||||
raise ConfigError(f"invalid context name {ctx_name!r}")
|
||||
if not isinstance(ctx, dict):
|
||||
raise ConfigError(f"context '{ctx_name}' must be a JSON object")
|
||||
_require_enabled("context", ctx_name, ctx)
|
||||
gitea = ctx.get("gitea")
|
||||
if gitea is not None:
|
||||
if not isinstance(gitea, dict):
|
||||
raise ConfigError(
|
||||
f"context '{ctx_name}' has a non-object 'gitea' block")
|
||||
_require_enabled("service", f"{ctx_name}.gitea", gitea)
|
||||
_reject_inline_secrets("service", f"{ctx_name}.gitea", gitea)
|
||||
services = ctx.get("services") or {}
|
||||
if not isinstance(services, dict):
|
||||
raise ConfigError(
|
||||
f"context '{ctx_name}' has a non-object 'services' block")
|
||||
for svc_name, svc in services.items():
|
||||
_validate_context_service(ctx_name, svc_name, svc)
|
||||
|
||||
raw_profiles = data.get("profiles")
|
||||
if not isinstance(raw_profiles, dict) or not raw_profiles:
|
||||
raise ConfigError(
|
||||
f"{path} version 2 contexts config requires a non-empty "
|
||||
"'profiles' object"
|
||||
)
|
||||
profiles = {}
|
||||
unavailable = {}
|
||||
for name, raw in raw_profiles.items():
|
||||
if not is_valid_profile_name(name):
|
||||
raise ConfigError(f"invalid profile name {name!r}")
|
||||
if not isinstance(raw, dict):
|
||||
raise ConfigError(f"profile '{name}' must be a JSON object")
|
||||
enabled = _require_enabled("profile", name, raw)
|
||||
_reject_inline_secrets("profile", name, raw)
|
||||
_validate_identity_auth(name, raw.get("auth"))
|
||||
ctx_name = raw.get("context")
|
||||
if ctx_name not in contexts:
|
||||
raise ConfigError(
|
||||
f"profile '{name}' references unknown context {ctx_name!r}")
|
||||
context = contexts[ctx_name]
|
||||
|
||||
allowed = raw.get("allowed_operations") or []
|
||||
forbidden = raw.get("forbidden_operations") or []
|
||||
if not isinstance(allowed, list) or not isinstance(forbidden, list):
|
||||
raise ConfigError(f"profile '{name}' operation fields must be lists")
|
||||
allowed_n = {_normalize_op("gitea", op, name) for op in allowed}
|
||||
forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden}
|
||||
# Reviewer-identity deadlock rule (#100/#103) applies here unchanged.
|
||||
if allowed_n & _REVIEW_MERGE_OPS:
|
||||
missing = sorted(_AUTHOR_ONLY_OPS - forbidden_n)
|
||||
if missing:
|
||||
raise ConfigError(
|
||||
f"profile '{name}' allows PR approve/merge but does not "
|
||||
f"forbid {missing}; reviewer identities must forbid "
|
||||
"gitea.pr.create and gitea.branch.push "
|
||||
"(reviewer-identity deadlock rule)"
|
||||
)
|
||||
|
||||
profile = dict(raw)
|
||||
profile["allowed_operations"] = sorted(allowed_n)
|
||||
profile["forbidden_operations"] = sorted(forbidden_n)
|
||||
gitea = context.get("gitea") or {}
|
||||
if not profile.get("base_url") and gitea.get("enabled"):
|
||||
profile["base_url"] = gitea.get("base_url")
|
||||
|
||||
username = profile.get("username") or ""
|
||||
if not enabled:
|
||||
unavailable[name] = (
|
||||
f"profile '{name}' is disabled (enabled: false); defined but "
|
||||
"unavailable for action — refusing, no fallback"
|
||||
)
|
||||
elif not context.get("enabled"):
|
||||
unavailable[name] = (
|
||||
f"profile '{name}' belongs to context '{ctx_name}' which is "
|
||||
"disabled (enabled: false); refusing, no fallback"
|
||||
)
|
||||
elif not profile.get("base_url"):
|
||||
unavailable[name] = (
|
||||
f"profile '{name}' has no usable base_url (none set and the "
|
||||
f"context '{ctx_name}' gitea service is disabled or has none); "
|
||||
"fail closed"
|
||||
)
|
||||
elif _TBD_RE.match(username):
|
||||
unavailable[name] = (
|
||||
f"profile '{name}' username {username!r} is a TBD placeholder; "
|
||||
"provision the account before use (fail closed)"
|
||||
)
|
||||
else:
|
||||
profiles[name] = profile
|
||||
continue
|
||||
# Unavailable profiles keep their (secret-free) body for audits only.
|
||||
profile["_unavailable_reason"] = unavailable[name]
|
||||
profiles.setdefault("_audit_only", {})
|
||||
profiles["_audit_only"][name] = profile
|
||||
|
||||
projects = data.get("projects") or {}
|
||||
if not isinstance(projects, dict):
|
||||
raise ConfigError(f"{path} 'projects' must be a JSON object")
|
||||
for proj_path, proj in projects.items():
|
||||
if not isinstance(proj, dict):
|
||||
raise ConfigError(f"project '{proj_path}' must be a JSON object")
|
||||
_require_enabled("project", proj_path, proj)
|
||||
if proj.get("context") not in contexts:
|
||||
raise ConfigError(
|
||||
f"project '{proj_path}' references unknown context "
|
||||
f"{proj.get('context')!r}"
|
||||
)
|
||||
|
||||
rules = data.get("rules") or {}
|
||||
if not isinstance(rules, dict):
|
||||
raise ConfigError(f"{path} 'rules' must be a JSON object")
|
||||
|
||||
audit_only = profiles.pop("_audit_only", {})
|
||||
return {
|
||||
"version": 2,
|
||||
"shape": "contexts",
|
||||
"profiles": profiles,
|
||||
"unavailable": unavailable,
|
||||
"audit_only_profiles": audit_only,
|
||||
"contexts": contexts,
|
||||
"projects": projects,
|
||||
"rules": rules,
|
||||
}
|
||||
|
||||
|
||||
def resolve_service(config, context_name, service_name):
|
||||
"""Return one context service's config for *internal* MCP use.
|
||||
|
||||
The returned dict includes the endpoint base_url and the keychain auth
|
||||
*reference* — both are for MCP-internal resolution only and must never be
|
||||
echoed into normal LLM-facing output (see audit_config/service_summaries).
|
||||
Fails closed on an unknown or disabled context/service; never falls back
|
||||
to another service.
|
||||
"""
|
||||
contexts = (config or {}).get("contexts")
|
||||
if not isinstance(contexts, dict):
|
||||
raise ConfigError(
|
||||
"service resolution requires a version 2 contexts config")
|
||||
ctx = contexts.get(context_name)
|
||||
if ctx is None:
|
||||
raise ConfigError(
|
||||
f"unknown context '{context_name}' (fail closed, no fallback)")
|
||||
if not ctx.get("enabled"):
|
||||
raise ConfigError(
|
||||
f"context '{context_name}' is disabled; its services are defined "
|
||||
"but unavailable for action (no fallback)"
|
||||
)
|
||||
if service_name == "gitea":
|
||||
service = ctx.get("gitea")
|
||||
else:
|
||||
service = (ctx.get("services") or {}).get(service_name)
|
||||
if service is None:
|
||||
raise ConfigError(
|
||||
f"unknown service '{service_name}' in context '{context_name}' "
|
||||
"(fail closed, no fallback)"
|
||||
)
|
||||
if not service.get("enabled"):
|
||||
raise ConfigError(
|
||||
f"service '{context_name}.{service_name}' is disabled; defined "
|
||||
"but unavailable for action — refusing, no fallback"
|
||||
)
|
||||
return dict(service)
|
||||
|
||||
|
||||
def project_for_path(config, path):
|
||||
"""Map a local project *path* to its context entry, failing closed.
|
||||
|
||||
Returns None when the path is not configured (feature off for that repo).
|
||||
Raises :class:`ConfigError` when the project or its context is disabled —
|
||||
a configured-but-disabled project must never be acted on.
|
||||
"""
|
||||
projects = (config or {}).get("projects") or {}
|
||||
project = projects.get(path)
|
||||
if project is None:
|
||||
return None
|
||||
if not project.get("enabled"):
|
||||
raise ConfigError(
|
||||
f"project '{path}' is disabled (enabled: false); refusing, "
|
||||
"no fallback"
|
||||
)
|
||||
contexts = (config or {}).get("contexts") or {}
|
||||
ctx = contexts.get(project.get("context")) or {}
|
||||
if not ctx.get("enabled"):
|
||||
raise ConfigError(
|
||||
f"project '{path}' maps to context '{project.get('context')}' "
|
||||
"which is disabled; refusing, no fallback"
|
||||
)
|
||||
return dict(project)
|
||||
|
||||
|
||||
def _audit_profile_entry(name, profile, enabled, reveal_endpoints):
|
||||
"""One LLM-safe audit row: no endpoint URLs, no keychain ids, no tokens."""
|
||||
auth = profile.get("auth") if isinstance(profile, dict) else None
|
||||
entry = {
|
||||
"name": name,
|
||||
"enabled": enabled,
|
||||
"context": profile.get("context") or profile.get("environment"),
|
||||
"role": profile.get("role"),
|
||||
"username": profile.get("username"),
|
||||
"auth": (auth or {}).get("type") if isinstance(auth, dict) else None,
|
||||
}
|
||||
reason = profile.get("_unavailable_reason")
|
||||
if reason:
|
||||
entry["reason"] = reason
|
||||
if reveal_endpoints:
|
||||
entry["base_url"] = profile.get("base_url")
|
||||
entry["auth_source"] = auth_source_name(profile)
|
||||
return entry
|
||||
|
||||
|
||||
def audit_config(config, reveal_endpoints=False):
|
||||
"""Report enabled/disabled profiles and services without secrets.
|
||||
|
||||
Default output is LLM-safe: names, contexts, enabled state, capability
|
||||
labels, and the auth *type* only — never endpoint URLs, keychain ids,
|
||||
token values, or auth source names. ``reveal_endpoints=True`` is the
|
||||
explicit admin/debug opt-in for local diagnostics: it adds base URLs and
|
||||
non-secret auth source names (``keychain:<id>`` / env var name). Token
|
||||
values are never included on any path.
|
||||
"""
|
||||
if config is None:
|
||||
return {"version": None, "profiles": [], "services": []}
|
||||
report = {
|
||||
"version": config.get("version"),
|
||||
"shape": config.get("shape") or ("environments"
|
||||
if config.get("aliases") is not None
|
||||
else "profiles"),
|
||||
"profiles": [],
|
||||
"services": [],
|
||||
}
|
||||
for name, profile in (config.get("profiles") or {}).items():
|
||||
if not isinstance(profile, dict):
|
||||
continue
|
||||
report["profiles"].append(_audit_profile_entry(
|
||||
name, profile, True, reveal_endpoints))
|
||||
for name, profile in (config.get("audit_only_profiles") or {}).items():
|
||||
report["profiles"].append(_audit_profile_entry(
|
||||
name, profile, False, reveal_endpoints))
|
||||
|
||||
for ctx_name, ctx in (config.get("contexts") or {}).items():
|
||||
ctx_enabled = bool(ctx.get("enabled"))
|
||||
for svc_name, svc in (ctx.get("services") or {}).items():
|
||||
entry = {
|
||||
"context": ctx_name,
|
||||
"name": svc_name,
|
||||
"kind": svc.get("kind"),
|
||||
"label": svc.get("label"),
|
||||
"enabled": ctx_enabled and bool(svc.get("enabled")),
|
||||
"capabilities": list(svc.get("capabilities") or []),
|
||||
"auth": (svc.get("auth") or {}).get("type"),
|
||||
}
|
||||
if reveal_endpoints:
|
||||
entry["base_url"] = svc.get("base_url")
|
||||
entry["auth_source"] = auth_source_name(svc)
|
||||
report["services"].append(entry)
|
||||
return report
|
||||
|
||||
|
||||
def service_summaries(config, auth_check=None):
|
||||
"""Safe one-line service summaries for LLM sessions.
|
||||
|
||||
Each line reports label + state only (e.g. ``PRGS Jenkins: enabled,
|
||||
read-only, authenticated`` / ``PRGS Sentry: disabled``) — never endpoint
|
||||
URLs, keychain ids, or token values. *auth_check* is a callable taking the
|
||||
service dict and returning True when its credential resolves; it defaults
|
||||
to a local keychain presence check and its result is reported only as
|
||||
``authenticated`` / ``no credential``.
|
||||
"""
|
||||
if auth_check is None:
|
||||
def auth_check(service):
|
||||
auth = service.get("auth") or {}
|
||||
if auth.get("type") == "keychain":
|
||||
return _keychain_token(auth.get("id")) is not None
|
||||
if auth.get("type") == "env":
|
||||
return bool(os.environ.get(auth.get("name") or ""))
|
||||
return False
|
||||
|
||||
lines = []
|
||||
for ctx_name, ctx in (config.get("contexts") or {}).items():
|
||||
ctx_enabled = bool(ctx.get("enabled"))
|
||||
for svc_name, svc in (ctx.get("services") or {}).items():
|
||||
label = svc.get("label") or f"{ctx_name} {svc_name}"
|
||||
if not (ctx_enabled and svc.get("enabled")):
|
||||
lines.append(f"{label}: disabled")
|
||||
continue
|
||||
caps = list(svc.get("capabilities") or [])
|
||||
cap_part = "read-only" if caps == ["read"] else ", ".join(caps)
|
||||
auth_part = "authenticated" if auth_check(svc) else "no credential"
|
||||
parts = ["enabled"] + ([cap_part] if cap_part else []) + [auth_part]
|
||||
lines.append(f"{label}: " + ", ".join(parts))
|
||||
return lines
|
||||
|
||||
|
||||
def _validate_auth(name, auth):
|
||||
"""Validate a profile's optional ``auth`` reference. Never echoes secrets."""
|
||||
if auth is None:
|
||||
@@ -147,18 +795,25 @@ def select_profile(config, name=None):
|
||||
if config is None:
|
||||
return None
|
||||
profiles = config.get("profiles", {})
|
||||
aliases = config.get("aliases") or {}
|
||||
unavailable = config.get("unavailable") or {}
|
||||
name = name or selected_profile_name()
|
||||
available = sorted(profiles)
|
||||
available = sorted(set(profiles) | set(aliases))
|
||||
if not name:
|
||||
raise ConfigError(
|
||||
f"{ENV_CONFIG_PATH} is set but {ENV_PROFILE} is not; "
|
||||
f"available profiles: {available}"
|
||||
)
|
||||
if name not in profiles:
|
||||
# Strict resolution order (#103): exact alias → exact profile address →
|
||||
# fail closed. No fuzzy matching, no partial matches, no defaults.
|
||||
resolved = aliases.get(name, name)
|
||||
if resolved in unavailable:
|
||||
raise ConfigError(unavailable[resolved])
|
||||
if resolved not in profiles:
|
||||
raise ConfigError(
|
||||
f"profile '{name}' not found in config; available profiles: {available}"
|
||||
)
|
||||
profile = profiles[name]
|
||||
profile = profiles[resolved]
|
||||
if not isinstance(profile, dict):
|
||||
raise ConfigError(f"profile '{name}' must be a JSON object")
|
||||
for secret_key in ("token", "password"):
|
||||
@@ -292,9 +947,21 @@ def validate_config(config):
|
||||
problems = []
|
||||
if not isinstance(config, dict):
|
||||
return ["config is not a JSON object"]
|
||||
if config.get("version", SUPPORTED_VERSION) != SUPPORTED_VERSION:
|
||||
version = config.get("version")
|
||||
if version is None:
|
||||
problems.append(
|
||||
f"unsupported version {config.get('version')!r} (expected {SUPPORTED_VERSION})"
|
||||
f"missing required 'version' (expected one of {list(SUPPORTED_VERSIONS)})"
|
||||
)
|
||||
elif version == 2:
|
||||
# v2 validation is all-or-nothing via the loader's invariants.
|
||||
try:
|
||||
_load_v2_any(config, "<config>")
|
||||
except ConfigError as exc:
|
||||
problems.append(str(exc))
|
||||
return problems
|
||||
elif version != SUPPORTED_VERSION:
|
||||
problems.append(
|
||||
f"unsupported version {version!r} (expected one of {list(SUPPORTED_VERSIONS)})"
|
||||
)
|
||||
profiles = config.get("profiles")
|
||||
if not isinstance(profiles, dict):
|
||||
@@ -445,5 +1112,20 @@ if __name__ == "__main__": # pragma: no cover - thin CLI dispatch
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "menu":
|
||||
import gitea_config_menu
|
||||
raise SystemExit(gitea_config_menu.main(sys.argv[2:]))
|
||||
print("usage: python gitea_config.py menu", file=sys.stderr)
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "audit":
|
||||
# Local admin/debug diagnostics (#120). --reveal-endpoints is the
|
||||
# explicit opt-in that adds base URLs and non-secret auth source
|
||||
# names; token values are never printed on any path.
|
||||
try:
|
||||
config = load_config(config_path() or DEFAULT_CONFIG_PATH)
|
||||
report = audit_config(
|
||||
config, reveal_endpoints="--reveal-endpoints" in sys.argv[2:])
|
||||
report["summaries"] = service_summaries(config)
|
||||
except ConfigError as exc:
|
||||
print(f"config error: {exc}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
print(json.dumps(report, indent=2))
|
||||
raise SystemExit(0)
|
||||
print("usage: python gitea_config.py menu | audit [--reveal-endpoints]",
|
||||
file=sys.stderr)
|
||||
raise SystemExit(2)
|
||||
|
||||
+78
-16
@@ -43,6 +43,16 @@ from gitea_auth import ( # noqa: E402
|
||||
get_profile,
|
||||
)
|
||||
import gitea_audit # noqa: E402
|
||||
import gitea_config # noqa: E402
|
||||
|
||||
|
||||
def _reveal_endpoints() -> bool:
|
||||
"""Admin/debug opt-in (#120): include endpoint URLs and token source
|
||||
names in tool output. Off by default so normal LLM-facing responses
|
||||
expose only logical names and status. Never affects token values, which
|
||||
are excluded on every path."""
|
||||
return (os.environ.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() \
|
||||
in ("1", "true", "yes")
|
||||
|
||||
mcp = FastMCP("gitea-tools", instructions=(
|
||||
"Gitea issue tracker and PR management for dadeschools and prgs instances. "
|
||||
@@ -511,14 +521,24 @@ def gitea_check_pr_eligibility(
|
||||
return result
|
||||
|
||||
# Profile capability check (metadata only; not enforcement of the action).
|
||||
# Both the action and the profile lists are normalized before comparison
|
||||
# (#106), so legacy spellings ("merge") and canonical namespaced ops
|
||||
# ("gitea.pr.merge") always match each other and never cross services.
|
||||
allowed = profile["allowed_operations"]
|
||||
forbidden = profile["forbidden_operations"]
|
||||
if not allowed:
|
||||
reasons.append("profile has no configured allowed operations (fail closed)")
|
||||
if action in forbidden:
|
||||
reasons.append(f"profile forbids '{action}'")
|
||||
elif action not in allowed:
|
||||
reasons.append(f"profile is not allowed to {action}")
|
||||
op_ok, op_reason = gitea_config.check_operation(action, allowed, forbidden)
|
||||
if not op_ok:
|
||||
if op_reason == "no-allowed-operations":
|
||||
reasons.append(
|
||||
"profile has no configured allowed operations (fail closed)")
|
||||
elif op_reason == "forbidden":
|
||||
reasons.append(f"profile forbids '{action}'")
|
||||
elif op_reason == "invalid-forbidden-entry":
|
||||
reasons.append(
|
||||
"profile has an unrecognized forbidden operation entry "
|
||||
"(fail closed)")
|
||||
else:
|
||||
reasons.append(f"profile is not allowed to {action}")
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
|
||||
@@ -1382,21 +1402,26 @@ def gitea_whoami(
|
||||
"Verify the configured token is valid for this instance."
|
||||
)
|
||||
# Runtime profile metadata is non-secret (name + allowed op categories).
|
||||
# The token is resolved separately and is never included here.
|
||||
# The token is resolved separately and is never included here. Endpoint
|
||||
# URLs stay out of normal LLM-facing output (#120): the logical remote
|
||||
# name is the addressing surface; 'server' appears only under the
|
||||
# GITEA_MCP_REVEAL_ENDPOINTS admin opt-in.
|
||||
profile = get_profile()
|
||||
return {
|
||||
result = {
|
||||
"authenticated": True,
|
||||
"username": data.get("login"),
|
||||
"display_name": data.get("full_name") or None,
|
||||
"user_id": data.get("id"),
|
||||
"email": data.get("email") or None,
|
||||
"server": f"https://{h}",
|
||||
"remote": remote,
|
||||
"profile": {
|
||||
"profile_name": profile["profile_name"],
|
||||
"allowed_operations": profile["allowed_operations"],
|
||||
},
|
||||
}
|
||||
if _reveal_endpoints():
|
||||
result["server"] = f"https://{h}"
|
||||
return result
|
||||
|
||||
|
||||
|
||||
@@ -1427,9 +1452,11 @@ def gitea_get_profile(
|
||||
|
||||
Read-only. Reports the non-secret configuration of the running MCP
|
||||
process (profile name, allowed/forbidden operation categories, audit
|
||||
label, token *source name*, base URL) plus the resolved server for the
|
||||
given remote. Optionally resolves the authenticated username via
|
||||
``gitea_whoami``'s endpoint so an LLM can see who this runtime acts as.
|
||||
label, auth *status*). Endpoint URLs and token source names are hidden
|
||||
from normal output (#120) and appear only under the
|
||||
GITEA_MCP_REVEAL_ENDPOINTS admin opt-in. Optionally resolves the
|
||||
authenticated username via ``gitea_whoami``'s endpoint so an LLM can see
|
||||
who this runtime acts as.
|
||||
|
||||
This tool never mutates Gitea and never approves, merges, comments, or
|
||||
creates anything. It never returns the token value, Authorization header,
|
||||
@@ -1447,18 +1474,25 @@ def gitea_get_profile(
|
||||
'verified', 'unknown', 'unavailable', or 'not_resolved'.
|
||||
"""
|
||||
profile = get_profile()
|
||||
reveal = _reveal_endpoints()
|
||||
result = {
|
||||
"profile_name": profile["profile_name"],
|
||||
"allowed_operations": profile["allowed_operations"],
|
||||
"forbidden_operations": profile["forbidden_operations"],
|
||||
"audit_label": profile["audit_label"],
|
||||
"token_source_name": profile["token_source_name"],
|
||||
"base_url": profile["base_url"],
|
||||
# Auth is reported as a status only (#120): the token source *name*
|
||||
# (env var name / keychain id) joins endpoint URLs behind the
|
||||
# GITEA_MCP_REVEAL_ENDPOINTS admin opt-in. Token values never appear.
|
||||
"auth_status": ("configured" if profile["token_source_name"]
|
||||
else "unconfigured"),
|
||||
"remote": remote if remote in REMOTES else None,
|
||||
"server": None,
|
||||
"authenticated_username": None,
|
||||
"identity_status": "not_resolved",
|
||||
}
|
||||
if reveal:
|
||||
result["token_source_name"] = profile["token_source_name"]
|
||||
result["base_url"] = profile["base_url"]
|
||||
result["server"] = None
|
||||
|
||||
if remote not in REMOTES:
|
||||
# Mark ambiguity rather than raising: the tool stays inspectable.
|
||||
@@ -1467,7 +1501,8 @@ def gitea_get_profile(
|
||||
return result
|
||||
|
||||
h = host or REMOTES[remote]["host"]
|
||||
result["server"] = f"https://{h}"
|
||||
if reveal:
|
||||
result["server"] = f"https://{h}"
|
||||
|
||||
if resolve_identity:
|
||||
try:
|
||||
@@ -1487,6 +1522,33 @@ def gitea_get_profile(
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_audit_config() -> dict:
|
||||
"""Audit the configured profiles/services: enabled state, no secrets.
|
||||
|
||||
Read-only and local-only: loads the canonical profiles.json named by
|
||||
GITEA_MCP_CONFIG and reports profile/service names, contexts, enabled
|
||||
state, capabilities, auth *status*, and one-line service summaries (e.g.
|
||||
``PRGS Jenkins: enabled, read-only, authenticated``). Disabled entries
|
||||
are listed so they can be audited, but the server refuses to act with
|
||||
them and never falls back to another profile or service.
|
||||
|
||||
Never includes endpoint URLs, keychain ids, token source names, or token
|
||||
values. Endpoint-revealing diagnostics exist only in the local admin CLI
|
||||
(``python3 gitea_config.py audit --reveal-endpoints``), never over MCP.
|
||||
"""
|
||||
config = gitea_config.load_config()
|
||||
if config is None:
|
||||
return {
|
||||
"configured": False,
|
||||
"message": "No GITEA_MCP_CONFIG configured; env-only mode.",
|
||||
}
|
||||
report = gitea_config.audit_config(config)
|
||||
report["configured"] = True
|
||||
report["summaries"] = gitea_config.service_summaries(config)
|
||||
return report
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_mark_issue(
|
||||
issue_number: int,
|
||||
|
||||
@@ -48,7 +48,7 @@ Additional issue-first rules:
|
||||
the existing one. Design debates belong on the issue, where other LLMs
|
||||
comment directly. Discussion-only tasks must **not** create branches or PRs;
|
||||
their comments should include recommendations, risks, open questions, and a
|
||||
Controller Handoff Summary (§K).
|
||||
Controller Handoff (§K; compact format unless high-risk).
|
||||
- **If the repo/tracker home for the work is unclear, stop and ask for an
|
||||
owner decision.** Do not create a new repository or a new tracker unless
|
||||
explicitly approved by the owner.
|
||||
@@ -182,7 +182,7 @@ results); and merge with a **pinned head SHA** and, where supported, the
|
||||
**expected changed-file set**, so a moved head or widened diff refuses the
|
||||
merge. After a real merge:
|
||||
|
||||
1. Confirm remote `master` actually contains the merge commit (A PR is not done just because `master` moved. A PR is done only when: Gitea reports the PR merged or reconciliation documents equivalent content on `master`; remote `master` contains the expected content; linked issues are closed; `status:in-progress` is removed).
|
||||
1. Confirm remote `master` actually contains the merge commit or expected squashed changes via post-merge file-presence verification (A PR is not done just because `master` moved or is marked "closed". Verify that expected files added/modified in the PR are actually present on `master` using `git pull`, `git log --oneline -- <file>`, or `git merge-base --is-ancestor`; linked issues are closed; `status:in-progress` is removed).
|
||||
2. Close/release the issue.
|
||||
3. Whenever an issue is closed, check for `status:in-progress`: remove it, or report why it could not be removed.
|
||||
4. Do not delete the remote source branch until: PR `merged=true`, or reconciliation confirms content is safely landed, or the issue owner explicitly abandons the work.
|
||||
@@ -190,7 +190,7 @@ merge. After a real merge:
|
||||
6. Remove the branch worktree folder (`scripts/worktree-clean --delete-branch <branch>`). Branches/worktrees are cleaned only after the above is verified.
|
||||
7. Fetch/prune.
|
||||
8. Confirm the main checkout is clean and current (`0 0` vs remote).
|
||||
9. Final merge/reconciliation reports must include both: PR metadata (state, merged flag, merge commit/hash) and Git content (remote master hash, expected content present or not).
|
||||
9. Final merge/reconciliation reports must include: PR metadata (state, merged flag, merge commit/hash), Git content (remote master hash, expected content present or not), and the exact post-merge verification method used & results.
|
||||
|
||||
Never run cleanup before the merge is confirmed on remote `master`.
|
||||
|
||||
@@ -248,16 +248,42 @@ Ready-to-copy templates live in [`templates/`](templates/):
|
||||
- [`worktree-cleanup.md`](templates/worktree-cleanup.md) — clean up after merge.
|
||||
- [`release-tag.md`](templates/release-tag.md) — create a release tag.
|
||||
|
||||
## K. Controller Handoff Summary (required, every task)
|
||||
## K. Controller Handoff (required, every task)
|
||||
|
||||
Every LLM task **must end with a `Controller Handoff Summary`** — whether the
|
||||
Every LLM task **must end with a `Controller Handoff`** — whether the
|
||||
task was implementation, review, merge, issue triage, documentation,
|
||||
discussion-only, or blocked planning. It lets a controller LLM understand the
|
||||
current state immediately, without rereading the conversation.
|
||||
|
||||
Rules:
|
||||
**The compact format is the default.** It is written for controller-LLM
|
||||
readability, not as a full human status report. PR bodies still carry the
|
||||
full review detail — the handoff never replaces PR documentation.
|
||||
|
||||
- Never omit the summary.
|
||||
Compact format (default):
|
||||
|
||||
```md
|
||||
## Controller Handoff
|
||||
|
||||
- Task:
|
||||
- Repo/state:
|
||||
- Issues/PRs:
|
||||
- Changed:
|
||||
- Validation:
|
||||
- Blockers:
|
||||
- Review:
|
||||
- Next:
|
||||
- Safety:
|
||||
```
|
||||
|
||||
The `Safety:` line is never omitted; it is usually:
|
||||
|
||||
```text
|
||||
no self-review; no self-merge; no tags; no secrets; no prod
|
||||
```
|
||||
|
||||
Rules (both formats):
|
||||
|
||||
- Never omit the handoff, and never omit the safety confirmations.
|
||||
- Never bury blockers in earlier text only — they must appear here.
|
||||
- If you opened a PR, state clearly that review is needed.
|
||||
- If you reviewed but could not merge, name the exact gate that blocked it.
|
||||
@@ -269,7 +295,18 @@ Rules:
|
||||
bypass classifiers, profile gates, missing permissions, or live-consent
|
||||
requirements**; give the owner concrete options.
|
||||
|
||||
Required format:
|
||||
**Use the long format below instead of the compact one only when the task was
|
||||
high-risk or complex** — i.e. when any of these happened:
|
||||
|
||||
- a merge, tag, or release
|
||||
- failed validation
|
||||
- permissions/profile gates blocked work
|
||||
- secrets or production access were involved
|
||||
- a complicated owner decision
|
||||
- multiple repos or cross-issue state
|
||||
- the owner explicitly asks for the full format
|
||||
|
||||
Long format (high-risk/complex tasks only):
|
||||
|
||||
```md
|
||||
## Controller Handoff Summary
|
||||
|
||||
@@ -23,12 +23,17 @@ Steps:
|
||||
4. If any gate fails → STOP and report.
|
||||
4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
|
||||
optionally pinning the reviewed head SHA / changed-file set.
|
||||
5. Confirm remote master now contains the merge commit.
|
||||
5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
|
||||
*Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
|
||||
|
||||
Then run the cleanup template (worktree-cleanup.md):
|
||||
- Verify expected file/commit presence on master (post-merge file-presence verification):
|
||||
- Run: git fetch <remote> --prune; git checkout master; git pull <remote> master --ff-only
|
||||
- Verify that the expected files added/modified in the PR are present on master (or absent if deleted).
|
||||
- Alternatively, verify with: git log --oneline -- <expected-file> or git merge-base --is-ancestor <pr-head-sha> master
|
||||
- close/release issue #<n>, remove status:in-progress (if it cannot be removed, report why)
|
||||
- delete remote branch, remove local branch + worktree folder
|
||||
- fetch/prune; confirm main checkout is clean and current (0 0).
|
||||
|
||||
Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash & Git content check.
|
||||
Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash, post-merge verification method used & verification results.
|
||||
```
|
||||
|
||||
@@ -38,6 +38,6 @@ Steps:
|
||||
- Eligibility: passed/failed
|
||||
|
||||
Handoff: reviewer identity, PR author, scope verdict, checks + results, decision —
|
||||
formatted as the Controller Handoff Summary (SKILL.md §K); if you could not
|
||||
merge, name the exact gate that blocked it.
|
||||
formatted per SKILL.md §K (compact by default; long form if a merge happened
|
||||
or a gate blocked you); if you could not merge, name the exact gate.
|
||||
```
|
||||
|
||||
@@ -43,6 +43,6 @@ Steps:
|
||||
9. Stop before review/merge — you are the author.
|
||||
|
||||
Handoff: issue #, branch, worktree path, files changed, checks + results, PR URL —
|
||||
formatted as the Controller Handoff Summary (SKILL.md §K); end with
|
||||
"Review needed — PR is open".
|
||||
formatted as the compact Controller Handoff (SKILL.md §K; long form only on
|
||||
the high-risk triggers); Review line: "Review needed — PR is open".
|
||||
```
|
||||
|
||||
+11
-6
@@ -127,11 +127,14 @@ class TestLoadSelect(_ConfigBase):
|
||||
gitea_config.resolve_profile()
|
||||
self.assertIn("version", str(ctx.exception))
|
||||
|
||||
def test_missing_version_defaults_ok(self):
|
||||
def test_missing_version_fails_closed(self):
|
||||
# Changed by #103: an unversioned config is ambiguous between the v1
|
||||
# and v2 shapes, so the loader now refuses to guess.
|
||||
self._write({"profiles": {"prgs": {"base_url": "https://x"}}})
|
||||
with patch.dict(os.environ, self._env("prgs"), clear=True):
|
||||
self.assertEqual(
|
||||
gitea_config.resolve_profile()["base_url"], "https://x")
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.resolve_profile()
|
||||
self.assertIn("version", str(ctx.exception))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -281,11 +284,13 @@ class TestAuthIntegration(_ConfigBase):
|
||||
self.assertEqual(header, "token process-token")
|
||||
|
||||
def test_auth_header_unresolvable_ref_fails_closed(self):
|
||||
# env token ref points at an unset var -> ConfigError inside resolve is
|
||||
# swallowed to "no token"; auth falls through to (mocked-empty) basic.
|
||||
# env token ref points at an unset var -> with GITEA_MCP_CONFIG set the
|
||||
# ConfigError propagates (fail closed, #120): no silent fallback to
|
||||
# Basic auth or another credential source.
|
||||
with patch.dict(os.environ, self._env("mdcps-env"), clear=True):
|
||||
with patch("gitea_auth.get_credentials", return_value=("", "")):
|
||||
self.assertIsNone(gitea_auth.get_auth_header("gitea.example.com"))
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_auth.get_auth_header("gitea.example.com")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,367 @@
|
||||
"""Tests for profiles.json version 2 (#103): environment → service → identity.
|
||||
|
||||
Covers: v2 loading + flattening, dotted-path and alias resolution with strict
|
||||
order (exact alias → exact address → fail closed), legacy v1 names via aliases,
|
||||
fail-closed validation (missing/unknown version, malformed hierarchy, ambiguous
|
||||
selectors, TBD-* usernames, reviewer-identity deadlock rule, inline secrets,
|
||||
missing auth, unnormalizable operations), service-default inheritance, and that
|
||||
flattened v2 profiles still work with resolve_token. No network, no secrets.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import copy
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_config # noqa: E402
|
||||
|
||||
FAKE_TOKEN = "fake-token-for-tests" # not a real credential
|
||||
|
||||
|
||||
def v2_config():
|
||||
"""A fresh, valid v2 config exercising both environments."""
|
||||
return {
|
||||
"version": 2,
|
||||
"environments": {
|
||||
"prgs": {
|
||||
"services": {
|
||||
"gitea": {
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"default_owner": "Scaled-Tech-Consulting",
|
||||
"identities": {
|
||||
"author": {
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "prgs.gitea.author.token"},
|
||||
"execution_profile": "prgs-author",
|
||||
"audit_label": "prgs-author",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create",
|
||||
"gitea.branch.push", "gitea.pr.create",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve", "gitea.pr.merge",
|
||||
],
|
||||
},
|
||||
"reviewer": {
|
||||
"role": "reviewer",
|
||||
"username": "sysadmin",
|
||||
"auth": {"type": "env",
|
||||
"name": "PRGS_REVIEWER_TOKEN"},
|
||||
"execution_profile": "prgs-reviewer",
|
||||
"audit_label": "prgs-reviewer",
|
||||
"default_repo": "Gitea-Tools",
|
||||
"allowed_operations": [
|
||||
"read", "review", "comment", "approve",
|
||||
"request_changes", "merge",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.create", "gitea.branch.push",
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"mdcps": {
|
||||
"services": {
|
||||
"gitea": {
|
||||
"base_url": "https://gitea.dadeschools.net",
|
||||
"identities": {
|
||||
"author": {
|
||||
"role": "author",
|
||||
"username": "913443",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "mdcps.gitea.author.token"},
|
||||
"allowed_operations": ["gitea.read"],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve", "gitea.pr.merge",
|
||||
],
|
||||
},
|
||||
"reviewer": {
|
||||
"role": "reviewer",
|
||||
"username": "TBD-second-mdcps-user",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "mdcps.gitea.reviewer.token"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.create", "gitea.branch.push",
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
"jenkins": {
|
||||
"base_url": "https://jenkins.dadeschools.net",
|
||||
"identities": {
|
||||
"reader": {
|
||||
"role": "reader",
|
||||
"username": "svc-jenkins-read",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "mdcps.jenkins.reader.token"},
|
||||
"allowed_operations": ["read", "jenkins.build.read"],
|
||||
"forbidden_operations": ["jenkins.build.trigger"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"aliases": {
|
||||
"mdcps": "mdcps.gitea.author",
|
||||
"prgs-author": "prgs.gitea.author",
|
||||
"prgs-reviewer": "prgs.gitea.reviewer",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class _V2Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.path = os.path.join(self._dir.name, "profiles.json")
|
||||
self._write(v2_config())
|
||||
|
||||
def tearDown(self):
|
||||
self._dir.cleanup()
|
||||
|
||||
def _write(self, obj):
|
||||
with open(self.path, "w", encoding="utf-8") as fh:
|
||||
fh.write(obj if isinstance(obj, str) else json.dumps(obj))
|
||||
|
||||
def _env(self, profile, **extra):
|
||||
env = {"GITEA_MCP_CONFIG": self.path, "GITEA_MCP_PROFILE": profile}
|
||||
env.update(extra)
|
||||
return env
|
||||
|
||||
def _resolve(self, profile):
|
||||
with patch.dict(os.environ, self._env(profile), clear=True):
|
||||
return gitea_config.resolve_profile()
|
||||
|
||||
def _load_raises(self, mutate, needle):
|
||||
cfg = v2_config()
|
||||
mutate(cfg)
|
||||
self._write(cfg)
|
||||
with patch.dict(os.environ, self._env("prgs.gitea.author"), clear=True):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.resolve_profile()
|
||||
self.assertIn(needle, str(ctx.exception))
|
||||
return str(ctx.exception)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Happy path: loading, dotted paths, aliases, inheritance
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestV2Loads(_V2Base):
|
||||
|
||||
def test_dotted_path_resolution(self):
|
||||
p = self._resolve("prgs.gitea.author")
|
||||
self.assertEqual(p["base_url"], "https://gitea.prgs.cc")
|
||||
self.assertEqual(p["username"], "jcwalker3")
|
||||
self.assertEqual(p["profile_path"], "prgs.gitea.author")
|
||||
self.assertEqual(p["environment"], "prgs")
|
||||
self.assertEqual(p["service"], "gitea")
|
||||
self.assertEqual(p["identity"], "author")
|
||||
self.assertEqual(p["role"], "author")
|
||||
|
||||
def test_alias_resolution_legacy_names(self):
|
||||
for legacy, addr in (
|
||||
("mdcps", "mdcps.gitea.author"),
|
||||
("prgs-author", "prgs.gitea.author"),
|
||||
("prgs-reviewer", "prgs.gitea.reviewer"),
|
||||
):
|
||||
p = self._resolve(legacy)
|
||||
self.assertEqual(p["profile_path"], addr, legacy)
|
||||
|
||||
def test_service_defaults_inherit_and_identity_overrides(self):
|
||||
author = self._resolve("prgs.gitea.author")
|
||||
self.assertEqual(author["default_owner"], "Scaled-Tech-Consulting")
|
||||
self.assertNotIn("default_repo", author)
|
||||
reviewer = self._resolve("prgs.gitea.reviewer")
|
||||
self.assertEqual(reviewer["default_owner"], "Scaled-Tech-Consulting")
|
||||
self.assertEqual(reviewer["default_repo"], "Gitea-Tools")
|
||||
|
||||
def test_unqualified_ops_normalized_minimally(self):
|
||||
reviewer = self._resolve("prgs.gitea.reviewer")
|
||||
self.assertIn("gitea.pr.merge", reviewer["allowed_operations"])
|
||||
self.assertIn("gitea.read", reviewer["allowed_operations"])
|
||||
self.assertNotIn("merge", reviewer["allowed_operations"])
|
||||
jenkins = self._resolve("mdcps.jenkins.reader")
|
||||
self.assertIn("jenkins.read", jenkins["allowed_operations"])
|
||||
self.assertIn("jenkins.build.read", jenkins["allowed_operations"])
|
||||
|
||||
def test_resolve_token_works_on_flattened_profile(self):
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
self._env("prgs.gitea.reviewer", PRGS_REVIEWER_TOKEN=FAKE_TOKEN),
|
||||
clear=True,
|
||||
):
|
||||
profile = gitea_config.resolve_profile()
|
||||
self.assertEqual(gitea_config.resolve_token(profile), FAKE_TOKEN)
|
||||
|
||||
def test_auth_source_name_on_flattened_profile(self):
|
||||
p = self._resolve("mdcps.gitea.author")
|
||||
self.assertEqual(
|
||||
gitea_config.auth_source_name(p), "keychain:mdcps.gitea.author.token"
|
||||
)
|
||||
|
||||
def test_v1_config_still_loads(self):
|
||||
self._write({
|
||||
"version": 1,
|
||||
"profiles": {"prgs": {
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "prgs-gitea-token"},
|
||||
}},
|
||||
})
|
||||
p = self._resolve("prgs")
|
||||
self.assertEqual(p["base_url"], "https://gitea.prgs.cc")
|
||||
|
||||
def test_validate_config_accepts_valid_v2(self):
|
||||
self.assertEqual(gitea_config.validate_config(v2_config()), [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fail-closed: selectors
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestV2Selectors(_V2Base):
|
||||
|
||||
def test_unknown_selector_fails_closed(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
self._resolve("prgs.gitea") # partial address — no fuzzy matching
|
||||
self.assertIn("not found", str(ctx.exception))
|
||||
|
||||
def test_no_fuzzy_matching_on_near_miss(self):
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
self._resolve("prgs-reviewers")
|
||||
|
||||
def test_conflicting_alias_and_address_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["aliases"]["prgs.gitea.author"] = "prgs.gitea.reviewer"
|
||||
self._load_raises(mutate, "conflicting selector")
|
||||
|
||||
def test_alias_to_unknown_target_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["aliases"]["ghost"] = "prgs.gitea.nope"
|
||||
self._load_raises(mutate, "unknown profile")
|
||||
|
||||
def test_tbd_username_fails_closed_on_selection(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
self._resolve("mdcps.gitea.reviewer")
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("TBD", msg)
|
||||
self.assertIn("provision", msg)
|
||||
|
||||
def test_tbd_identity_does_not_block_other_identities(self):
|
||||
# Same file contains the TBD reviewer; author still resolves.
|
||||
p = self._resolve("mdcps.gitea.author")
|
||||
self.assertEqual(p["username"], "913443")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fail-closed: structure and versions
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestV2Structure(_V2Base):
|
||||
|
||||
def test_missing_version_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
del cfg["version"]
|
||||
self._load_raises(mutate, "version")
|
||||
|
||||
def test_unknown_version_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["version"] = 3
|
||||
self._load_raises(mutate, "unsupported version")
|
||||
|
||||
def test_missing_environments_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
del cfg["environments"]
|
||||
self._load_raises(mutate, "environments")
|
||||
|
||||
def test_malformed_environment_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["environments"]["prgs"] = "not-an-object"
|
||||
self._load_raises(mutate, "must be a JSON object")
|
||||
|
||||
def test_missing_services_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["environments"]["prgs"]["services"] = {}
|
||||
self._load_raises(mutate, "services")
|
||||
|
||||
def test_missing_identities_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
cfg["environments"]["prgs"]["services"]["gitea"]["identities"] = {}
|
||||
self._load_raises(mutate, "identities")
|
||||
|
||||
def test_dotted_segment_name_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
envs = cfg["environments"]
|
||||
envs["bad.env"] = copy.deepcopy(envs["prgs"])
|
||||
self._load_raises(mutate, "invalid environment name")
|
||||
|
||||
def test_missing_base_url_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
svc = cfg["environments"]["prgs"]["services"]["gitea"]
|
||||
del svc["base_url"]
|
||||
self._load_raises(mutate, "base_url")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fail-closed: identity invariants
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestV2IdentityInvariants(_V2Base):
|
||||
|
||||
def _ident(self, cfg, addr="prgs.gitea.author"):
|
||||
env, svc, ident = addr.split(".")
|
||||
return cfg["environments"][env]["services"][svc]["identities"][ident]
|
||||
|
||||
def test_missing_auth_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
del self._ident(cfg)["auth"]
|
||||
self._load_raises(mutate, "missing an 'auth' reference")
|
||||
|
||||
def test_inline_secret_in_identity_rejected(self):
|
||||
def mutate(cfg):
|
||||
self._ident(cfg)["token"] = "oops-not-a-real-secret"
|
||||
msg = self._load_raises(mutate, "inline 'token'")
|
||||
self.assertNotIn("oops-not-a-real-secret", msg)
|
||||
|
||||
def test_inline_secret_in_auth_rejected(self):
|
||||
def mutate(cfg):
|
||||
self._ident(cfg)["auth"]["password"] = "oops-not-a-real-secret"
|
||||
msg = self._load_raises(mutate, "inline 'password'")
|
||||
self.assertNotIn("oops-not-a-real-secret", msg)
|
||||
|
||||
def test_reviewer_deadlock_invariant_enforced(self):
|
||||
def mutate(cfg):
|
||||
reviewer = self._ident(cfg, "prgs.gitea.reviewer")
|
||||
reviewer["forbidden_operations"] = [] # can approve/merge AND create
|
||||
msg = self._load_raises(mutate, "deadlock")
|
||||
self.assertIn("gitea.pr.create", msg)
|
||||
|
||||
def test_reviewer_deadlock_applies_to_unqualified_merge(self):
|
||||
def mutate(cfg):
|
||||
author = self._ident(cfg)
|
||||
author["allowed_operations"] = ["merge"] # normalized to gitea.pr.merge
|
||||
author["forbidden_operations"] = []
|
||||
self._load_raises(mutate, "deadlock")
|
||||
|
||||
def test_unnormalizable_operation_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
self._ident(cfg)["allowed_operations"] = ["frobnicate"]
|
||||
self._load_raises(mutate, "cannot be normalized")
|
||||
|
||||
def test_foreign_namespace_operation_fails_closed(self):
|
||||
def mutate(cfg):
|
||||
reader = self._ident(cfg, "mdcps.jenkins.reader")
|
||||
reader["allowed_operations"] = ["gitea.pr.merge"]
|
||||
self._load_raises(mutate, "cannot be normalized")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,446 @@
|
||||
"""Tests for profiles.json version 2 *contexts* shape (#120).
|
||||
|
||||
The canonical machine config uses ``contexts`` / ``profiles`` / ``projects`` /
|
||||
``rules`` with explicit ``enabled`` flags. Covers: loading + active-profile
|
||||
resolution via GITEA_MCP_PROFILE, fail-closed refusal of disabled profiles /
|
||||
contexts / services / projects, project-to-context mapping, base-URL fallback
|
||||
from the context's gitea block, keychain-only auth references, LLM-safe audit
|
||||
output (no endpoint URLs, no keychain ids, no tokens) with an explicit
|
||||
admin/debug opt-in, v1 compatibility, and the no-silent-fallback rule in
|
||||
gitea_auth.get_auth_header. No network, no real secrets.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_config # noqa: E402
|
||||
import gitea_auth # noqa: E402
|
||||
|
||||
FAKE_TOKEN = "fake-token-for-tests" # not a real credential
|
||||
|
||||
|
||||
def contexts_config():
|
||||
"""A fresh, valid v2 contexts-shape config with enabled/disabled entries."""
|
||||
return {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"prgs": {
|
||||
"enabled": True,
|
||||
"label": "Local / PRGS",
|
||||
"default_owner": "Scaled-Tech-Consulting",
|
||||
"gitea": {
|
||||
"enabled": True,
|
||||
"kind": "gitea",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
},
|
||||
"services": {
|
||||
"jenkins": {
|
||||
"enabled": True,
|
||||
"kind": "jenkins",
|
||||
"label": "PRGS Jenkins",
|
||||
"base_url": "https://jenkins.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "prgs-jenkins-token"},
|
||||
"capabilities": ["read"],
|
||||
},
|
||||
"sentry": {
|
||||
"enabled": False,
|
||||
"kind": "sentry",
|
||||
"label": "PRGS Sentry",
|
||||
"base_url": "",
|
||||
"auth": {"type": "keychain", "id": "prgs-sentry-token"},
|
||||
"capabilities": ["read"],
|
||||
},
|
||||
},
|
||||
},
|
||||
"lab": {
|
||||
"enabled": False,
|
||||
"gitea": {"enabled": False, "kind": "gitea", "base_url": ""},
|
||||
"services": {
|
||||
"jenkins": {
|
||||
"enabled": False,
|
||||
"kind": "jenkins",
|
||||
"label": "Lab Jenkins",
|
||||
"base_url": "http://localhost:8080",
|
||||
"auth": {"type": "keychain", "id": "lab-jenkins-token"},
|
||||
"capabilities": ["read"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"profiles": {
|
||||
"prgs-author": {
|
||||
"enabled": True,
|
||||
"context": "prgs",
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"execution_profile": "prgs-author",
|
||||
"audit_label": "prgs-author",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "prgs-gitea-author-token"},
|
||||
"allowed_operations": [
|
||||
"read", "branch", "commit", "push", "open_pr", "comment",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"approve", "request_changes", "merge",
|
||||
],
|
||||
},
|
||||
"prgs-reviewer": {
|
||||
"enabled": True,
|
||||
"context": "prgs",
|
||||
"role": "reviewer",
|
||||
"username": "sysadmin",
|
||||
"execution_profile": "prgs-reviewer",
|
||||
"audit_label": "prgs-reviewer",
|
||||
# no base_url on purpose: must fall back to context gitea
|
||||
"auth": {"type": "keychain", "id": "prgs-gitea-reviewer-token"},
|
||||
"allowed_operations": [
|
||||
"read", "review", "comment", "approve",
|
||||
"request_changes", "merge",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"branch", "commit", "push", "open_pr",
|
||||
],
|
||||
},
|
||||
"retired-author": {
|
||||
"enabled": False,
|
||||
"context": "prgs",
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "retired-token-ref"},
|
||||
"allowed_operations": ["read"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
"lab-author": {
|
||||
"enabled": True,
|
||||
"context": "lab",
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"base_url": "http://localhost:3000",
|
||||
"auth": {"type": "keychain", "id": "lab-gitea-author-token"},
|
||||
"allowed_operations": ["read"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
},
|
||||
"projects": {
|
||||
"/repo/one": {
|
||||
"enabled": True,
|
||||
"context": "prgs",
|
||||
"default_owner": "Scaled-Tech-Consulting",
|
||||
"default_repo": "One",
|
||||
"default_author_profile": "prgs-author",
|
||||
"default_reviewer_profile": "prgs-reviewer",
|
||||
},
|
||||
"/repo/lab": {
|
||||
"enabled": False,
|
||||
"context": "lab",
|
||||
},
|
||||
},
|
||||
"rules": {
|
||||
"disabled_behavior": "report in audits, never act",
|
||||
"no_silent_fallback": True,
|
||||
"tokens_in_json": False,
|
||||
"token_storage": "keychain",
|
||||
"hide_service_urls_from_llm": True,
|
||||
"hide_keychain_ids_from_llm": True,
|
||||
"mcp_resolves_endpoints": True,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def write_config(data):
|
||||
"""Write *data* to a temp JSON file and return its path."""
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||||
json.dump(data, fh)
|
||||
return path
|
||||
|
||||
|
||||
def load(data):
|
||||
"""Load *data* through gitea_config via a temp file, then clean up."""
|
||||
path = write_config(data)
|
||||
try:
|
||||
return gitea_config.load_config(path)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
class LoadContextsShapeTests(unittest.TestCase):
|
||||
def test_contexts_shape_loads(self):
|
||||
config = load(contexts_config())
|
||||
self.assertEqual(config["version"], 2)
|
||||
self.assertIn("prgs-author", config["profiles"])
|
||||
self.assertIn("prgs-reviewer", config["profiles"])
|
||||
|
||||
def test_active_profile_resolved_from_env(self):
|
||||
path = write_config(contexts_config())
|
||||
try:
|
||||
with patch.dict(os.environ, {
|
||||
gitea_config.ENV_CONFIG_PATH: path,
|
||||
gitea_config.ENV_PROFILE: "prgs-author",
|
||||
}):
|
||||
profile = gitea_config.resolve_profile()
|
||||
finally:
|
||||
os.unlink(path)
|
||||
self.assertEqual(profile["username"], "jcwalker3")
|
||||
self.assertEqual(profile["base_url"], "https://gitea.prgs.cc")
|
||||
self.assertEqual(profile["context"], "prgs")
|
||||
|
||||
def test_base_url_falls_back_to_context_gitea(self):
|
||||
profile = gitea_config.select_profile(load(contexts_config()),
|
||||
"prgs-reviewer")
|
||||
self.assertEqual(profile["base_url"], "https://gitea.prgs.cc")
|
||||
|
||||
def test_profile_without_any_base_url_is_refused(self):
|
||||
data = contexts_config()
|
||||
del data["profiles"]["prgs-author"]["base_url"]
|
||||
data["contexts"]["prgs"]["gitea"]["enabled"] = False
|
||||
config = load(data)
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_config.select_profile(config, "prgs-author")
|
||||
|
||||
def test_v1_config_still_loads(self):
|
||||
config = load({
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs": {
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "prgs-gitea-token"},
|
||||
},
|
||||
},
|
||||
})
|
||||
profile = gitea_config.select_profile(config, "prgs")
|
||||
self.assertEqual(profile["base_url"], "https://gitea.prgs.cc")
|
||||
|
||||
def test_mixed_contexts_and_environments_rejected(self):
|
||||
data = contexts_config()
|
||||
data["environments"] = {"x": {"services": {}}}
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
load(data)
|
||||
|
||||
def test_missing_enabled_flag_is_refused(self):
|
||||
data = contexts_config()
|
||||
del data["profiles"]["prgs-author"]["enabled"]
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
load(data)
|
||||
self.assertIn("enabled", str(ctx.exception))
|
||||
|
||||
|
||||
class DisabledRefusalTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.config = load(contexts_config())
|
||||
|
||||
def test_disabled_profile_refused(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.select_profile(self.config, "retired-author")
|
||||
self.assertIn("disabled", str(ctx.exception))
|
||||
|
||||
def test_profile_in_disabled_context_refused(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.select_profile(self.config, "lab-author")
|
||||
self.assertIn("disabled", str(ctx.exception))
|
||||
|
||||
def test_enabled_profile_still_selectable(self):
|
||||
profile = gitea_config.select_profile(self.config, "prgs-author")
|
||||
self.assertEqual(profile["context"], "prgs")
|
||||
|
||||
def test_disabled_service_refused(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.resolve_service(self.config, "prgs", "sentry")
|
||||
self.assertIn("disabled", str(ctx.exception))
|
||||
|
||||
def test_enabled_service_resolves_internally_with_auth_reference(self):
|
||||
# Internal resolution keeps the URL + auth reference for MCP's own use;
|
||||
# they must never appear in LLM-facing (audit/summary) output.
|
||||
service = gitea_config.resolve_service(self.config, "prgs", "jenkins")
|
||||
self.assertEqual(service["base_url"], "https://jenkins.prgs.cc")
|
||||
self.assertEqual(service["auth"], {"type": "keychain",
|
||||
"id": "prgs-jenkins-token"})
|
||||
self.assertNotIn("token", service)
|
||||
|
||||
def test_service_in_disabled_context_refused(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.resolve_service(self.config, "lab", "jenkins")
|
||||
self.assertIn("disabled", str(ctx.exception))
|
||||
|
||||
def test_unknown_service_fails_closed(self):
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_config.resolve_service(self.config, "prgs", "nope")
|
||||
|
||||
|
||||
class ProjectMappingTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.config = load(contexts_config())
|
||||
|
||||
def test_project_maps_to_context(self):
|
||||
project = gitea_config.project_for_path(self.config, "/repo/one")
|
||||
self.assertEqual(project["context"], "prgs")
|
||||
self.assertEqual(project["default_reviewer_profile"], "prgs-reviewer")
|
||||
|
||||
def test_unknown_project_returns_none(self):
|
||||
self.assertIsNone(
|
||||
gitea_config.project_for_path(self.config, "/repo/unknown"))
|
||||
|
||||
def test_disabled_project_refused(self):
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
gitea_config.project_for_path(self.config, "/repo/lab")
|
||||
self.assertIn("disabled", str(ctx.exception))
|
||||
|
||||
|
||||
class SecretHandlingTests(unittest.TestCase):
|
||||
def test_inline_profile_token_rejected(self):
|
||||
data = contexts_config()
|
||||
data["profiles"]["prgs-author"]["token"] = FAKE_TOKEN
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
load(data)
|
||||
self.assertNotIn(FAKE_TOKEN, str(ctx.exception))
|
||||
|
||||
def test_inline_service_token_rejected(self):
|
||||
data = contexts_config()
|
||||
data["contexts"]["prgs"]["services"]["jenkins"]["token"] = FAKE_TOKEN
|
||||
with self.assertRaises(gitea_config.ConfigError) as ctx:
|
||||
load(data)
|
||||
self.assertNotIn(FAKE_TOKEN, str(ctx.exception))
|
||||
|
||||
def test_selected_profile_resolves_token_via_keychain(self):
|
||||
profile = gitea_config.select_profile(load(contexts_config()),
|
||||
"prgs-author")
|
||||
token = gitea_config.resolve_token(
|
||||
profile, keychain_lookup=lambda item_id: FAKE_TOKEN
|
||||
if item_id == "prgs-gitea-author-token" else None)
|
||||
self.assertEqual(token, FAKE_TOKEN)
|
||||
|
||||
|
||||
class AuditTests(unittest.TestCase):
|
||||
"""LLM-facing audit output: enabled/disabled state only — no endpoint
|
||||
URLs, no keychain ids, no token values. Admin opt-in reveals endpoints
|
||||
and auth source names (never token values)."""
|
||||
|
||||
def setUp(self):
|
||||
self.config = load(contexts_config())
|
||||
|
||||
def test_audit_reports_enabled_and_disabled(self):
|
||||
report = gitea_config.audit_config(self.config)
|
||||
profiles = {p["name"]: p for p in report["profiles"]}
|
||||
self.assertTrue(profiles["prgs-author"]["enabled"])
|
||||
self.assertFalse(profiles["retired-author"]["enabled"])
|
||||
services = {(s["context"], s["name"]): s for s in report["services"]}
|
||||
self.assertTrue(services[("prgs", "jenkins")]["enabled"])
|
||||
self.assertFalse(services[("prgs", "sentry")]["enabled"])
|
||||
self.assertFalse(services[("lab", "jenkins")]["enabled"])
|
||||
|
||||
def test_audit_hides_urls_keychain_ids_and_tokens_by_default(self):
|
||||
rendered = json.dumps(gitea_config.audit_config(self.config))
|
||||
for leaked in ("https://", "http://", "prgs-gitea-author-token",
|
||||
"prgs-jenkins-token", "base_url", FAKE_TOKEN):
|
||||
self.assertNotIn(leaked, rendered)
|
||||
# Auth is reported as a status, not a reference.
|
||||
report = gitea_config.audit_config(self.config)
|
||||
profiles = {p["name"]: p for p in report["profiles"]}
|
||||
self.assertEqual(profiles["prgs-author"]["auth"], "keychain")
|
||||
|
||||
def test_audit_admin_optin_reveals_endpoints_but_never_tokens(self):
|
||||
report = gitea_config.audit_config(self.config, reveal_endpoints=True)
|
||||
rendered = json.dumps(report)
|
||||
self.assertIn("https://jenkins.prgs.cc", rendered)
|
||||
self.assertIn("keychain:prgs-gitea-author-token", rendered)
|
||||
self.assertNotIn(FAKE_TOKEN, rendered)
|
||||
|
||||
def test_audit_works_for_v1_config(self):
|
||||
report = gitea_config.audit_config({
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs": {
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain", "id": "prgs-gitea-token"},
|
||||
},
|
||||
},
|
||||
})
|
||||
profiles = {p["name"]: p for p in report["profiles"]}
|
||||
self.assertTrue(profiles["prgs"]["enabled"])
|
||||
self.assertEqual(profiles["prgs"]["auth"], "keychain")
|
||||
self.assertNotIn("https://", json.dumps(report))
|
||||
|
||||
|
||||
class ServiceSummaryTests(unittest.TestCase):
|
||||
"""Safe one-line summaries for LLM sessions: label + state only."""
|
||||
|
||||
def setUp(self):
|
||||
self.config = load(contexts_config())
|
||||
|
||||
def test_summaries_show_state_without_urls_or_ids(self):
|
||||
lines = gitea_config.service_summaries(
|
||||
self.config, auth_check=lambda service: True)
|
||||
text = "\n".join(lines)
|
||||
self.assertIn("PRGS Jenkins: enabled, read-only, authenticated", text)
|
||||
self.assertIn("PRGS Sentry: disabled", text)
|
||||
self.assertIn("Lab Jenkins: disabled", text)
|
||||
for leaked in ("https://", "http://", "keychain",
|
||||
"prgs-jenkins-token"):
|
||||
self.assertNotIn(leaked, text)
|
||||
|
||||
def test_summary_reports_missing_auth_without_secrets(self):
|
||||
lines = gitea_config.service_summaries(
|
||||
self.config, auth_check=lambda service: False)
|
||||
text = "\n".join(lines)
|
||||
self.assertIn("PRGS Jenkins: enabled, read-only, no credential", text)
|
||||
|
||||
|
||||
class NoSilentFallbackTests(unittest.TestCase):
|
||||
def test_broken_config_fails_auth_instead_of_falling_back(self):
|
||||
"""With GITEA_MCP_CONFIG set but unloadable, auth must fail closed."""
|
||||
path = write_config({"version": 2}) # invalid: no contexts/environments
|
||||
env = {
|
||||
gitea_config.ENV_CONFIG_PATH: path,
|
||||
gitea_config.ENV_PROFILE: "prgs-author",
|
||||
}
|
||||
try:
|
||||
with patch.dict(os.environ, env, clear=False), \
|
||||
patch.object(gitea_auth, "get_credentials",
|
||||
return_value=(None, None)):
|
||||
for var in ("GITEA_TOKEN", "GITEA_TOKEN_PRGS",
|
||||
"GITEA_TOKEN_DADESCHOOLS"):
|
||||
os.environ.pop(var, None)
|
||||
with self.assertRaises(gitea_config.ConfigError):
|
||||
gitea_auth.get_auth_header("https://gitea.prgs.cc")
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
def test_env_only_users_unaffected(self):
|
||||
"""Without GITEA_MCP_CONFIG, a missing token still degrades quietly."""
|
||||
env = dict(os.environ)
|
||||
env.pop(gitea_config.ENV_CONFIG_PATH, None)
|
||||
with patch.dict(os.environ, env, clear=True), \
|
||||
patch.object(gitea_auth, "get_credentials",
|
||||
return_value=(None, None)):
|
||||
for var in ("GITEA_TOKEN", "GITEA_TOKEN_PRGS",
|
||||
"GITEA_TOKEN_DADESCHOOLS"):
|
||||
os.environ.pop(var, None)
|
||||
self.assertIsNone(
|
||||
gitea_auth.get_auth_header("https://gitea.prgs.cc"))
|
||||
|
||||
|
||||
class ValidateConfigTests(unittest.TestCase):
|
||||
def test_valid_contexts_config_has_no_problems(self):
|
||||
self.assertEqual(gitea_config.validate_config(contexts_config()), [])
|
||||
|
||||
def test_repo_example_file_validates(self):
|
||||
example = __import__("pathlib").Path(__file__).resolve().parent.parent \
|
||||
/ "gitea-mcp.v2-contexts.example.json"
|
||||
with open(example, encoding="utf-8") as fh:
|
||||
self.assertEqual(gitea_config.validate_config(json.load(fh)), [])
|
||||
|
||||
def test_broken_contexts_config_reports_problems(self):
|
||||
data = contexts_config()
|
||||
data["profiles"]["prgs-author"]["context"] = "nope"
|
||||
problems = gitea_config.validate_config(data)
|
||||
self.assertTrue(problems)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+145
-3
@@ -3,6 +3,7 @@
|
||||
Each tool is tested by calling the underlying function directly (not through
|
||||
the MCP protocol) with mocked API responses.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
@@ -880,7 +881,9 @@ class TestWhoami(unittest.TestCase):
|
||||
self.assertEqual(result["username"], "reviewer-bot")
|
||||
self.assertEqual(result["display_name"], "Reviewer Bot")
|
||||
self.assertEqual(result["user_id"], 42)
|
||||
self.assertEqual(result["server"], "https://gitea.prgs.cc")
|
||||
# Endpoint URLs are hidden from normal LLM-facing output (#120);
|
||||
# the logical remote name is the addressing surface.
|
||||
self.assertNotIn("server", result)
|
||||
self.assertEqual(result["remote"], "prgs")
|
||||
# Read-only: GET against the authenticated-user endpoint.
|
||||
call_args = mock_api.call_args
|
||||
@@ -1035,8 +1038,12 @@ class TestProfileDiscovery(unittest.TestCase):
|
||||
self.assertEqual(result["allowed_operations"], ["read", "review", "approve"])
|
||||
self.assertEqual(result["authenticated_username"], "reviewer-bot")
|
||||
self.assertEqual(result["identity_status"], "verified")
|
||||
self.assertEqual(result["server"], "https://gitea.prgs.cc")
|
||||
self.assertEqual(result["token_source_name"], "GITEA_TOKEN")
|
||||
# Endpoint URLs and token source names are hidden from normal
|
||||
# LLM-facing output (#120); auth is reported as a status only.
|
||||
self.assertNotIn("server", result)
|
||||
self.assertNotIn("base_url", result)
|
||||
self.assertNotIn("token_source_name", result)
|
||||
self.assertEqual(result["auth_status"], "configured")
|
||||
# Read-only: only a GET to the user endpoint was issued.
|
||||
self.assertEqual(mock_api.call_args[0][0], "GET")
|
||||
self.assertTrue(mock_api.call_args[0][1].endswith("/api/v1/user"))
|
||||
@@ -1669,3 +1676,138 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
# branch name fallback
|
||||
self.assertEqual(extract_linked_issue_numbers("", branch_name="issue-123"), [123])
|
||||
self.assertEqual(extract_linked_issue_numbers("", branch_name="feat/issue-123-foo"), [123])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoint/keychain redaction in LLM-facing output — issue #120
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestEndpointRedaction(unittest.TestCase):
|
||||
"""Normal MCP output hides endpoint URLs and keychain ids; the admin
|
||||
opt-in (GITEA_MCP_REVEAL_ENDPOINTS) restores them for local diagnostics
|
||||
without ever revealing token values."""
|
||||
|
||||
def _contexts_config_file(self):
|
||||
import tempfile
|
||||
config = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"prgs": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "kind": "gitea",
|
||||
"base_url": "https://gitea.prgs.cc"},
|
||||
"services": {
|
||||
"jenkins": {
|
||||
"enabled": True, "kind": "jenkins",
|
||||
"label": "PRGS Jenkins",
|
||||
"base_url": "https://jenkins.prgs.cc",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "prgs-jenkins-token"},
|
||||
"capabilities": ["read"],
|
||||
},
|
||||
"sentry": {
|
||||
"enabled": False, "kind": "sentry",
|
||||
"label": "PRGS Sentry", "base_url": "",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "prgs-sentry-token"},
|
||||
"capabilities": ["read"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"profiles": {
|
||||
"prgs-author": {
|
||||
"enabled": True, "context": "prgs", "role": "author",
|
||||
"username": "jcwalker3",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"auth": {"type": "keychain",
|
||||
"id": "prgs-gitea-author-token"},
|
||||
"allowed_operations": ["read"],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
},
|
||||
"projects": {},
|
||||
"rules": {"hide_service_urls_from_llm": True,
|
||||
"hide_keychain_ids_from_llm": True,
|
||||
"mcp_resolves_endpoints": True},
|
||||
}
|
||||
fd, path = tempfile.mkstemp(suffix=".json")
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
||||
json.dump(config, fh)
|
||||
return path
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_whoami_hides_endpoint_url_by_default(self, _auth, mock_api):
|
||||
mock_api.return_value = {"id": 1, "login": "someone"}
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
result = gitea_whoami(remote="prgs")
|
||||
self.assertNotIn("server", result)
|
||||
self.assertNotIn("https://", repr(result))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_whoami_reveals_endpoint_with_admin_optin(self, _auth, mock_api):
|
||||
mock_api.return_value = {"id": 1, "login": "someone"}
|
||||
with patch.dict(os.environ,
|
||||
{"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||
result = gitea_whoami(remote="prgs")
|
||||
self.assertEqual(result["server"], "https://gitea.prgs.cc")
|
||||
|
||||
def test_get_profile_hides_url_and_token_source_by_default(self):
|
||||
env = {
|
||||
"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_BASE_URL": "https://gitea.example.invalid",
|
||||
"GITEA_TOKEN_SOURCE": "keychain:some-item-id",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
result = gitea_get_profile(remote="prgs",
|
||||
resolve_identity=False)
|
||||
blob = repr(result)
|
||||
for leaked in ("https://", "keychain:", "some-item-id",
|
||||
"base_url", "server", "token_source_name"):
|
||||
self.assertNotIn(leaked, blob)
|
||||
self.assertEqual(result["auth_status"], "configured")
|
||||
|
||||
def test_get_profile_reports_unconfigured_auth(self):
|
||||
with patch.dict(os.environ,
|
||||
{"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
|
||||
result = gitea_get_profile(remote="prgs",
|
||||
resolve_identity=False)
|
||||
self.assertEqual(result["auth_status"], "unconfigured")
|
||||
|
||||
def test_get_profile_reveals_with_admin_optin(self):
|
||||
env = {
|
||||
"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_TOKEN_SOURCE": "keychain:some-item-id",
|
||||
"GITEA_MCP_REVEAL_ENDPOINTS": "1",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
result = gitea_get_profile(remote="prgs",
|
||||
resolve_identity=False)
|
||||
self.assertEqual(result["server"], "https://gitea.prgs.cc")
|
||||
self.assertEqual(result["token_source_name"], "keychain:some-item-id")
|
||||
|
||||
def test_audit_tool_reports_state_without_urls_or_ids(self):
|
||||
from mcp_server import gitea_audit_config
|
||||
path = self._contexts_config_file()
|
||||
try:
|
||||
env = {"GITEA_MCP_CONFIG": path,
|
||||
"GITEA_MCP_PROFILE": "prgs-author"}
|
||||
with patch.dict(os.environ, env, clear=True), \
|
||||
patch("gitea_config._keychain_token", return_value="x"):
|
||||
result = gitea_audit_config()
|
||||
finally:
|
||||
os.unlink(path)
|
||||
blob = json.dumps(result)
|
||||
self.assertIn("PRGS Jenkins: enabled, read-only, authenticated",
|
||||
result["summaries"])
|
||||
self.assertIn("PRGS Sentry: disabled", result["summaries"])
|
||||
for leaked in ("https://", "http://", "prgs-jenkins-token",
|
||||
"prgs-gitea-author-token", "base_url"):
|
||||
self.assertNotIn(leaked, blob)
|
||||
|
||||
def test_audit_tool_without_config_reports_off(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
from mcp_server import gitea_audit_config
|
||||
result = gitea_audit_config()
|
||||
self.assertFalse(result["configured"])
|
||||
|
||||
@@ -0,0 +1,239 @@
|
||||
"""Operation-name normalization table and enforcement tests — issue #106.
|
||||
|
||||
Covers the required matrix from #106:
|
||||
|
||||
- fully qualified allowed / forbidden operations
|
||||
- legacy unqualified allowed / forbidden operations
|
||||
- unknown operations (fail closed)
|
||||
- ambiguous operations (fail closed)
|
||||
- service mismatch (cross-service names never accepted by the wrong service)
|
||||
- forbidden-overrides-allowed
|
||||
- empty / missing allowed list
|
||||
- duplicate operations after normalization
|
||||
- no silent permission widening
|
||||
- eligibility enforcement normalizes before checking
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_config # noqa: E402
|
||||
from gitea_config import ( # noqa: E402
|
||||
ConfigError,
|
||||
check_operation,
|
||||
normalize_operation,
|
||||
)
|
||||
from mcp_server import gitea_check_pr_eligibility # noqa: E402
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# normalize_operation — canonical table
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestNormalizeOperation(unittest.TestCase):
|
||||
|
||||
def test_fully_qualified_gitea_op_unchanged(self):
|
||||
self.assertEqual(normalize_operation("gitea.pr.merge"), "gitea.pr.merge")
|
||||
|
||||
def test_legacy_aliases_map_to_canonical_names(self):
|
||||
expected = {
|
||||
"merge": "gitea.pr.merge",
|
||||
"approve": "gitea.pr.approve",
|
||||
"request_changes": "gitea.pr.request_changes",
|
||||
"review": "gitea.pr.review",
|
||||
"comment": "gitea.pr.comment",
|
||||
"read": "gitea.read",
|
||||
}
|
||||
for legacy, canonical in expected.items():
|
||||
self.assertEqual(normalize_operation(legacy), canonical)
|
||||
|
||||
def test_contexts_shape_author_verbs(self):
|
||||
self.assertEqual(normalize_operation("branch"), "gitea.branch.create")
|
||||
self.assertEqual(normalize_operation("commit"), "gitea.repo.commit")
|
||||
self.assertEqual(normalize_operation("push"), "gitea.branch.push")
|
||||
self.assertEqual(normalize_operation("open_pr"), "gitea.pr.create")
|
||||
|
||||
def test_unknown_unqualified_op_fails_closed(self):
|
||||
with self.assertRaises(ConfigError):
|
||||
normalize_operation("frobnicate")
|
||||
|
||||
def test_ambiguous_dotted_op_fails_closed(self):
|
||||
# Dotted but neither gitea-prefixed nor an explicit alias: refuse to
|
||||
# guess which namespace was meant.
|
||||
with self.assertRaises(ConfigError):
|
||||
normalize_operation("build.read")
|
||||
|
||||
def test_cross_service_name_rejected_by_wrong_service(self):
|
||||
with self.assertRaises(ConfigError):
|
||||
normalize_operation("jenkins.read", service="gitea")
|
||||
with self.assertRaises(ConfigError):
|
||||
normalize_operation("gitea.read", service="jenkins")
|
||||
|
||||
def test_non_gitea_single_word_namespaced_to_service(self):
|
||||
self.assertEqual(normalize_operation("read", service="jenkins"),
|
||||
"jenkins.read")
|
||||
|
||||
def test_non_gitea_qualified_own_prefix_unchanged(self):
|
||||
self.assertEqual(
|
||||
normalize_operation("jenkins.build.read", service="jenkins"),
|
||||
"jenkins.build.read",
|
||||
)
|
||||
|
||||
def test_empty_and_non_string_fail_closed(self):
|
||||
for bad in ("", None, 3, ["merge"]):
|
||||
with self.assertRaises(ConfigError):
|
||||
normalize_operation(bad)
|
||||
|
||||
def test_gitea_alias_not_applied_to_other_services(self):
|
||||
# "merge" on jenkins must not resolve to the *gitea* merge permission.
|
||||
self.assertEqual(normalize_operation("merge", service="jenkins"),
|
||||
"jenkins.merge")
|
||||
|
||||
def test_table_is_documented_and_matches_normalization(self):
|
||||
table = gitea_config.GITEA_OPERATION_ALIASES
|
||||
self.assertIsInstance(table, dict)
|
||||
self.assertTrue(table)
|
||||
for legacy, canonical in table.items():
|
||||
self.assertEqual(normalize_operation(legacy), canonical)
|
||||
self.assertTrue(canonical.startswith("gitea."))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# check_operation — enforcement semantics (normalize BEFORE checking)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCheckOperation(unittest.TestCase):
|
||||
|
||||
def test_fully_qualified_allowed(self):
|
||||
ok, reason = check_operation("gitea.pr.merge", ["gitea.pr.merge"])
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(reason, "allowed")
|
||||
|
||||
def test_fully_qualified_forbidden(self):
|
||||
ok, reason = check_operation(
|
||||
"gitea.pr.merge", ["gitea.pr.merge"], ["gitea.pr.merge"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "forbidden")
|
||||
|
||||
def test_legacy_unqualified_allowed(self):
|
||||
ok, reason = check_operation("merge", ["gitea.pr.merge"])
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(reason, "allowed")
|
||||
|
||||
def test_legacy_unqualified_forbidden(self):
|
||||
ok, reason = check_operation("merge", ["gitea.pr.merge"], ["merge"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "forbidden")
|
||||
|
||||
def test_unknown_operation_fails_closed(self):
|
||||
ok, reason = check_operation("frobnicate", ["gitea.read"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "invalid-operation")
|
||||
|
||||
def test_ambiguous_operation_fails_closed(self):
|
||||
ok, reason = check_operation("build.read", ["gitea.read"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "invalid-operation")
|
||||
|
||||
def test_service_mismatch_rejected(self):
|
||||
ok, reason = check_operation("jenkins.read", ["gitea.read"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "invalid-operation")
|
||||
|
||||
def test_forbidden_overrides_allowed_across_spellings(self):
|
||||
# Allowed via legacy spelling, forbidden via canonical spelling: the
|
||||
# forbidden entry must win after both normalize to the same op.
|
||||
ok, reason = check_operation("merge", ["merge"], ["gitea.pr.merge"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "forbidden")
|
||||
|
||||
def test_empty_allowed_list_denies(self):
|
||||
ok, reason = check_operation("gitea.read", [])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "no-allowed-operations")
|
||||
|
||||
def test_missing_allowed_list_denies(self):
|
||||
ok, reason = check_operation("gitea.read", None)
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "no-allowed-operations")
|
||||
|
||||
def test_duplicates_after_normalization_are_harmless(self):
|
||||
ok, reason = check_operation(
|
||||
"merge", ["merge", "gitea.pr.merge", "merge"])
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(reason, "allowed")
|
||||
|
||||
def test_unnormalizable_allowed_entry_grants_nothing(self):
|
||||
# A junk allowed entry must not widen permissions to anything.
|
||||
ok, reason = check_operation("gitea.read", ["frobnicate"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "not-allowed")
|
||||
|
||||
def test_unnormalizable_forbidden_entry_fails_closed(self):
|
||||
# If a forbidden entry cannot be understood, deny rather than risk
|
||||
# silently narrowing the forbidden set (which would widen permissions).
|
||||
ok, reason = check_operation(
|
||||
"gitea.read", ["gitea.read"], ["frobnicate"])
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(reason, "invalid-forbidden-entry")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Eligibility enforcement — normalization happens before checking (#106)
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestEligibilityNormalizesOperations(unittest.TestCase):
|
||||
|
||||
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
||||
return {
|
||||
"user": {"login": author},
|
||||
"state": state,
|
||||
"head": {"sha": sha},
|
||||
"mergeable": mergeable,
|
||||
}
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_namespaced_profile_ops_allow_legacy_action(self, _auth, mock_api):
|
||||
# JSON-config profiles carry canonical namespaced ops; the raw action
|
||||
# "merge" must still match them after normalization.
|
||||
mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-merger",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.merge"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_check_pr_eligibility(pr_number=9, action="merge",
|
||||
remote="prgs")
|
||||
self.assertTrue(r["eligible"])
|
||||
self.assertNotIn("profile is not allowed to merge", r["reasons"])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_namespaced_forbidden_op_blocks_legacy_action(self, _auth, mock_api):
|
||||
mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-merger",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.merge",
|
||||
"GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.merge"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_check_pr_eligibility(pr_number=9, action="merge",
|
||||
remote="prgs")
|
||||
self.assertFalse(r["eligible"])
|
||||
self.assertIn("profile forbids 'merge'", r["reasons"])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_legacy_env_ops_still_work(self, _auth, mock_api):
|
||||
# v1/env behaviour stays compatible: unqualified env ops keep working.
|
||||
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
r = gitea_check_pr_eligibility(pr_number=5, action="review",
|
||||
remote="prgs")
|
||||
self.assertTrue(r["eligible"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user