Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 23aa2fb192 | |||
| cd633e2c2b | |||
| 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/
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
+634
-10
@@ -54,11 +54,67 @@ 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")
|
||||
|
||||
# ── Minimal operation normalization (#103) ─────────────────────────────────────
|
||||
# Only what the #103 invariants need. The full normalization table, deprecation
|
||||
# handling, and enforcement test matrix belong to issue #106 — do not grow this
|
||||
# beyond invariant safety here.
|
||||
_MINIMAL_GITEA_OP_MAP = {
|
||||
"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_op(service, op, addr):
|
||||
"""Normalize *op* for *service*, or fail closed (#103 minimal subset).
|
||||
|
||||
- already namespaced for this service (``{service}.*``) → unchanged
|
||||
- known unqualified Gitea ops → mapped via ``_MINIMAL_GITEA_OP_MAP``
|
||||
- unqualified single-word ops on non-Gitea services → ``{service}.{op}``
|
||||
- anything else (foreign prefixes, unknown unqualified names) → ConfigError
|
||||
"""
|
||||
if not isinstance(op, str) or not op:
|
||||
raise ConfigError(f"identity '{addr}' has an empty or non-string operation")
|
||||
if op.startswith(service + "."):
|
||||
return op
|
||||
if service == "gitea" and op in _MINIMAL_GITEA_OP_MAP:
|
||||
return _MINIMAL_GITEA_OP_MAP[op]
|
||||
if service != "gitea" and "." not in op:
|
||||
return f"{service}.{op}"
|
||||
raise ConfigError(
|
||||
f"identity '{addr}' has operation {op!r} that cannot be normalized "
|
||||
f"safely for service '{service}' (fail closed; full table is issue #106)"
|
||||
)
|
||||
|
||||
# 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 +164,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 +737,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 +889,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 +1054,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)
|
||||
|
||||
+61
-9
@@ -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. "
|
||||
@@ -1382,21 +1392,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 +1442,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 +1464,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,6 +1491,7 @@ def gitea_get_profile(
|
||||
return result
|
||||
|
||||
h = host or REMOTES[remote]["host"]
|
||||
if reveal:
|
||||
result["server"] = f"https://{h}"
|
||||
|
||||
if resolve_identity:
|
||||
@@ -1487,6 +1512,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,
|
||||
|
||||
Executable
+284
@@ -0,0 +1,284 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Migration helper to convert profiles.json from version 1 to version 2 environments shape.
|
||||
|
||||
This script preserves existing keychain references (auth.id) and maps old profile
|
||||
names as aliases so that existing IDE configurations continue to function.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
import shutil
|
||||
import tempfile
|
||||
|
||||
# Resolve path to import gitea_config
|
||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
if PROJECT_ROOT not in sys.path:
|
||||
sys.path.insert(0, PROJECT_ROOT)
|
||||
|
||||
import gitea_config
|
||||
|
||||
|
||||
AUTHOR_DEFAULT_ALLOWED = ["read", "branch", "commit", "push", "open_pr", "comment"]
|
||||
AUTHOR_DEFAULT_FORBIDDEN = ["approve", "request_changes", "merge"]
|
||||
REVIEWER_DEFAULT_ALLOWED = [
|
||||
"read", "review", "comment", "approve", "request_changes", "merge"
|
||||
]
|
||||
REVIEWER_DEFAULT_FORBIDDEN = ["branch", "commit", "push", "open_pr"]
|
||||
|
||||
|
||||
def infer_role(name, execution_profile):
|
||||
"""Return the unambiguous role for a legacy profile name, or None."""
|
||||
haystack = f"{name} {execution_profile or ''}".lower()
|
||||
has_author = "author" in haystack
|
||||
has_reviewer = "reviewer" in haystack
|
||||
if has_author == has_reviewer:
|
||||
return None
|
||||
return "reviewer" if has_reviewer else "author"
|
||||
|
||||
|
||||
def migration_summary(v2_data):
|
||||
"""Return a redacted summary of the migrated config."""
|
||||
environments = v2_data.get("environments", {})
|
||||
service_count = 0
|
||||
identity_count = 0
|
||||
for env in environments.values():
|
||||
services = env.get("services", {})
|
||||
service_count += len(services)
|
||||
for service in services.values():
|
||||
identity_count += len(service.get("identities", {}))
|
||||
return {
|
||||
"version": v2_data.get("version"),
|
||||
"environments": len(environments),
|
||||
"services": service_count,
|
||||
"identities": identity_count,
|
||||
"aliases": len(v2_data.get("aliases", {})),
|
||||
}
|
||||
|
||||
|
||||
def migrate_v1_to_v2(v1_data):
|
||||
"""Convert version 1 profiles.json format to version 2 environments format."""
|
||||
environments = {}
|
||||
aliases = {}
|
||||
|
||||
profiles = v1_data.get("profiles", {})
|
||||
if not isinstance(profiles, dict):
|
||||
raise ValueError("Malformed input: 'profiles' field must be a JSON object")
|
||||
|
||||
for name, prof in profiles.items():
|
||||
if not isinstance(prof, dict):
|
||||
raise ValueError(f"Malformed input: profile '{name}' must be a JSON object")
|
||||
|
||||
# Infer environment and identity name
|
||||
if "-" in name:
|
||||
parts = name.split("-", 1)
|
||||
env_name = parts[0]
|
||||
ident_name = parts[1]
|
||||
else:
|
||||
env_name = name
|
||||
ident_name = "author"
|
||||
|
||||
# Determine role and identity based on name / execution_profile.
|
||||
# Ambiguous profiles may still migrate only when they carry explicit
|
||||
# permissions; otherwise role-based defaults could widen permissions.
|
||||
exec_prof = prof.get("execution_profile") or ""
|
||||
role = infer_role(name, exec_prof)
|
||||
if role == "reviewer":
|
||||
ident_name = "reviewer"
|
||||
elif role == "author":
|
||||
ident_name = "author"
|
||||
else:
|
||||
role = prof.get("role")
|
||||
if role not in (None, "author", "reviewer"):
|
||||
raise ValueError(
|
||||
f"Profile '{name}' has unsupported role {role!r}"
|
||||
)
|
||||
|
||||
# Construct identity block
|
||||
identity_data = {
|
||||
"username": prof.get("username"),
|
||||
"auth": prof.get("auth"),
|
||||
}
|
||||
if role:
|
||||
identity_data["role"] = role
|
||||
if prof.get("execution_profile"):
|
||||
identity_data["execution_profile"] = prof["execution_profile"]
|
||||
|
||||
# Set audit label (default to old name to preserve context)
|
||||
identity_data["audit_label"] = prof.get("audit_label") or name
|
||||
|
||||
has_allowed = "allowed_operations" in prof
|
||||
has_forbidden = "forbidden_operations" in prof
|
||||
if has_allowed != has_forbidden:
|
||||
raise ValueError(
|
||||
f"Profile '{name}' must define both allowed_operations and "
|
||||
"forbidden_operations, or neither (fail closed)"
|
||||
)
|
||||
if has_allowed:
|
||||
allowed = prof.get("allowed_operations")
|
||||
forbidden = prof.get("forbidden_operations")
|
||||
if not isinstance(allowed, list) or not isinstance(forbidden, list):
|
||||
raise ValueError(
|
||||
f"Profile '{name}' operation fields must be lists"
|
||||
)
|
||||
identity_data["allowed_operations"] = list(allowed)
|
||||
identity_data["forbidden_operations"] = list(forbidden)
|
||||
elif role == "author":
|
||||
identity_data["allowed_operations"] = list(AUTHOR_DEFAULT_ALLOWED)
|
||||
identity_data["forbidden_operations"] = list(AUTHOR_DEFAULT_FORBIDDEN)
|
||||
elif role == "reviewer":
|
||||
identity_data["allowed_operations"] = list(REVIEWER_DEFAULT_ALLOWED)
|
||||
identity_data["forbidden_operations"] = list(REVIEWER_DEFAULT_FORBIDDEN)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Profile '{name}' has no explicit operation lists and no "
|
||||
"unambiguous author/reviewer role marker (fail closed)"
|
||||
)
|
||||
|
||||
# Nest inside environments/services structure
|
||||
env = environments.setdefault(env_name, {})
|
||||
services = env.setdefault("services", {})
|
||||
gitea_svc = services.setdefault("gitea", {})
|
||||
|
||||
# Copy service-level attributes
|
||||
if prof.get("base_url"):
|
||||
gitea_svc["base_url"] = prof["base_url"]
|
||||
if prof.get("default_owner"):
|
||||
gitea_svc["default_owner"] = prof["default_owner"]
|
||||
if prof.get("default_repo"):
|
||||
gitea_svc["default_repo"] = prof["default_repo"]
|
||||
|
||||
identities = gitea_svc.setdefault("identities", {})
|
||||
identities[ident_name] = identity_data
|
||||
|
||||
# Alias resolution targets
|
||||
alias_target = f"{env_name}.gitea.{ident_name}"
|
||||
if name != alias_target:
|
||||
aliases[name] = alias_target
|
||||
|
||||
# Extra convenience alias for standard old-profile compatibility (e.g. prgs-author)
|
||||
convenience_alias = f"{env_name}-{ident_name}"
|
||||
if convenience_alias != alias_target and convenience_alias not in aliases:
|
||||
aliases[convenience_alias] = alias_target
|
||||
|
||||
v2_data = {
|
||||
"version": 2,
|
||||
"environments": environments,
|
||||
"aliases": aliases
|
||||
}
|
||||
return v2_data
|
||||
|
||||
|
||||
def validate_v2_data(v2_data):
|
||||
"""Validate generated v2 structure using gitea_config parser."""
|
||||
fd, temp_path = tempfile.mkstemp(suffix=".json")
|
||||
os.close(fd)
|
||||
try:
|
||||
with open(temp_path, "w") as f:
|
||||
json.dump(v2_data, f)
|
||||
# Attempt to load using load_config to run all validation rules
|
||||
gitea_config.load_config(temp_path)
|
||||
return True
|
||||
except Exception as e:
|
||||
raise ValueError(f"Generated v2 config failed validation: {e}")
|
||||
finally:
|
||||
try:
|
||||
os.remove(temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Migrate profiles.json from version 1 to version 2 environments shape."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-i", "--input",
|
||||
default=gitea_config.DEFAULT_CONFIG_PATH,
|
||||
help="Path to the version 1 profiles.json file (default: ~/.config/gitea-tools/profiles.json)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"-o", "--output",
|
||||
help="Path to write the migrated version 2 profiles.json file (default: overwrite input)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"-w", "--write",
|
||||
action="store_true",
|
||||
help="Actually write the migrated config and create a backup (default is dry-run)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--backup",
|
||||
help="Path to write the backup file (default: <input_path>.bak)"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = os.path.abspath(args.input)
|
||||
output_path = os.path.abspath(args.output or input_path)
|
||||
backup_path = args.backup or f"{input_path}.bak"
|
||||
|
||||
if not os.path.isfile(input_path):
|
||||
print(f"Error: Input file not found: {input_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
with open(input_path, "r") as f:
|
||||
v1_data = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Error: Input file is not valid JSON: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"Error reading input file: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Validate version
|
||||
version = v1_data.get("version")
|
||||
if version is not None and version != 1:
|
||||
print(f"Error: Unsupported profiles.json version: {version}. Expected version 1.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
v2_data = migrate_v1_to_v2(v1_data)
|
||||
validate_v2_data(v2_data)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if not args.write:
|
||||
print("=== DRY-RUN MODE (No files modified) ===")
|
||||
print("Generated v2 config validated successfully.")
|
||||
print("Only aggregate counts are shown.")
|
||||
summary = migration_summary(v2_data)
|
||||
print("Summary:")
|
||||
print(f" version: {summary['version']}")
|
||||
print(f" environments: {summary['environments']}")
|
||||
print(f" services: {summary['services']}")
|
||||
print(f" identities: {summary['identities']}")
|
||||
print(f" aliases: {summary['aliases']}")
|
||||
sys.exit(0)
|
||||
|
||||
# Write Mode: Create Backup first
|
||||
try:
|
||||
print(f"Creating backup: {backup_path}")
|
||||
shutil.copy2(input_path, backup_path)
|
||||
except Exception as e:
|
||||
print(f"Error creating backup: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Write migrated config
|
||||
try:
|
||||
print(f"Writing migrated version 2 config: {output_path}")
|
||||
# Ensure target directory exists
|
||||
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
||||
with open(output_path, "w") as f:
|
||||
json.dump(v2_data, f, indent=2)
|
||||
f.write("\n")
|
||||
print("Migration completed successfully!")
|
||||
except Exception as e:
|
||||
print(f"Error writing output file: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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,299 @@
|
||||
"""Unit tests for migrate_profiles.py migration helper."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import unittest
|
||||
import tempfile
|
||||
import shutil
|
||||
from unittest.mock import patch
|
||||
from io import StringIO
|
||||
|
||||
# Add project root to sys.path
|
||||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if PROJECT_ROOT not in sys.path:
|
||||
sys.path.insert(0, PROJECT_ROOT)
|
||||
|
||||
import migrate_profiles
|
||||
|
||||
|
||||
class TestMigrateProfiles(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.mkdtemp()
|
||||
self.v1_content = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "jcwalker3",
|
||||
"auth": {
|
||||
"type": "keychain",
|
||||
"id": "redacted-author-ref"
|
||||
},
|
||||
"default_owner": "Scaled-Tech-Consulting",
|
||||
"execution_profile": "personal-prgs",
|
||||
"allowed_operations": ["read", "comment"],
|
||||
"forbidden_operations": ["approve", "merge"]
|
||||
},
|
||||
"mdcps": {
|
||||
"base_url": "redacted-mdcps-service",
|
||||
"username": "913443",
|
||||
"auth": {
|
||||
"type": "keychain",
|
||||
"id": "redacted-mdcps-ref"
|
||||
},
|
||||
"default_owner": "Contractor",
|
||||
"execution_profile": "mdcps",
|
||||
"allowed_operations": ["read"],
|
||||
"forbidden_operations": ["merge"]
|
||||
},
|
||||
"prgs-reviewer": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "sysadmin",
|
||||
"auth": {
|
||||
"type": "keychain",
|
||||
"id": "redacted-reviewer-ref"
|
||||
},
|
||||
"default_owner": "Scaled-Tech-Consulting",
|
||||
"execution_profile": "prgs-reviewer",
|
||||
"allowed_operations": [
|
||||
"read", "review", "comment", "approve",
|
||||
"request_changes", "merge"
|
||||
],
|
||||
"forbidden_operations": ["branch", "commit", "push", "open_pr"]
|
||||
}
|
||||
}
|
||||
}
|
||||
self.input_file = os.path.join(self.temp_dir, "profiles.json")
|
||||
with open(self.input_file, "w") as f:
|
||||
json.dump(self.v1_content, f)
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.temp_dir)
|
||||
|
||||
def test_migration_logic(self):
|
||||
"""Test the structural transformation and capability mapping."""
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
||||
self.assertEqual(v2_data["version"], 2)
|
||||
|
||||
# Check environment structure
|
||||
envs = v2_data["environments"]
|
||||
self.assertIn("prgs", envs)
|
||||
self.assertIn("mdcps", envs)
|
||||
|
||||
# Check service and identity structure
|
||||
prgs_gitea = envs["prgs"]["services"]["gitea"]
|
||||
self.assertEqual(prgs_gitea["base_url"], "redacted-prgs-service")
|
||||
self.assertEqual(prgs_gitea["default_owner"], "Scaled-Tech-Consulting")
|
||||
|
||||
author = prgs_gitea["identities"]["author"]
|
||||
self.assertEqual(author["username"], "jcwalker3")
|
||||
self.assertEqual(author["auth"]["id"], "redacted-author-ref")
|
||||
self.assertEqual(author["allowed_operations"], ["read", "comment"])
|
||||
self.assertEqual(author["forbidden_operations"], ["approve", "merge"])
|
||||
|
||||
reviewer = prgs_gitea["identities"]["reviewer"]
|
||||
self.assertEqual(reviewer["role"], "reviewer")
|
||||
self.assertEqual(reviewer["username"], "sysadmin")
|
||||
self.assertEqual(reviewer["auth"]["id"], "redacted-reviewer-ref")
|
||||
self.assertIn("merge", reviewer["allowed_operations"])
|
||||
|
||||
def test_alias_generation(self):
|
||||
"""Test that aliases are correctly generated to support old profile names."""
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
||||
aliases = v2_data["aliases"]
|
||||
|
||||
self.assertEqual(aliases["prgs"], "prgs.gitea.author")
|
||||
self.assertEqual(aliases["prgs-author"], "prgs.gitea.author")
|
||||
self.assertEqual(aliases["prgs-reviewer"], "prgs.gitea.reviewer")
|
||||
self.assertEqual(aliases["mdcps"], "mdcps.gitea.author")
|
||||
|
||||
def test_no_secret_behavior(self):
|
||||
"""Ensure secrets are never extracted, printed, or processed."""
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
||||
# Check that auth structures only contain keychain references, not credentials
|
||||
for env in v2_data["environments"].values():
|
||||
for svc in env["services"].values():
|
||||
for ident in svc["identities"].values():
|
||||
auth = ident["auth"]
|
||||
self.assertEqual(auth["type"], "keychain")
|
||||
self.assertIn("id", auth)
|
||||
self.assertNotIn("token", auth)
|
||||
self.assertNotIn("password", auth)
|
||||
|
||||
def test_validation(self):
|
||||
"""Test that the generated v2 configuration validates against Gitea-Tools v2 loader."""
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
|
||||
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
|
||||
|
||||
@patch("sys.stdout", new_callable=StringIO)
|
||||
def test_dry_run_default(self, mock_stdout):
|
||||
"""Verify that running without -w prints generated config without modifying files."""
|
||||
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
|
||||
test_args = [
|
||||
"migrate_profiles.py",
|
||||
"-i", self.input_file,
|
||||
"-o", output_file
|
||||
]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
self.assertFalse(os.path.exists(output_file))
|
||||
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
|
||||
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
self.assertIn("DRY-RUN MODE", stdout_output)
|
||||
self.assertIn("version", stdout_output)
|
||||
self.assertIn("identities", stdout_output)
|
||||
self.assertIn("aliases", stdout_output)
|
||||
self.assertNotIn("redacted-prgs-service", stdout_output)
|
||||
self.assertNotIn("redacted-mdcps-service", stdout_output)
|
||||
self.assertNotIn("redacted-author-ref", stdout_output)
|
||||
self.assertNotIn("redacted-mdcps-ref", stdout_output)
|
||||
self.assertNotIn("redacted-reviewer-ref", stdout_output)
|
||||
self.assertNotIn("keychain", stdout_output)
|
||||
self.assertNotIn("auth", stdout_output)
|
||||
|
||||
def test_dry_run_hides_token_like_values(self):
|
||||
"""Verify dry-run summary does not expose token-like auth metadata."""
|
||||
sensitive = json.loads(json.dumps(self.v1_content))
|
||||
sensitive["profiles"]["prgs"]["auth"]["id"] = "super-secret-token-value"
|
||||
with open(self.input_file, "w") as f:
|
||||
json.dump(sensitive, f)
|
||||
|
||||
test_args = ["migrate_profiles.py", "-i", self.input_file]
|
||||
with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
stdout_output = mock_stdout.getvalue()
|
||||
self.assertNotIn("super-secret-token-value", stdout_output)
|
||||
self.assertNotIn("token", stdout_output.lower())
|
||||
|
||||
def test_explicit_operations_are_preserved(self):
|
||||
"""Explicit v1 permissions must not be replaced by role defaults."""
|
||||
v1_data = json.loads(json.dumps(self.v1_content))
|
||||
v1_data["profiles"]["prgs-reviewer"]["allowed_operations"] = ["read"]
|
||||
v1_data["profiles"]["prgs-reviewer"]["forbidden_operations"] = ["merge"]
|
||||
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
reviewer = (
|
||||
v2_data["environments"]["prgs"]["services"]["gitea"]
|
||||
["identities"]["reviewer"]
|
||||
)
|
||||
self.assertEqual(reviewer["allowed_operations"], ["read"])
|
||||
self.assertEqual(reviewer["forbidden_operations"], ["merge"])
|
||||
|
||||
def test_inferred_role_defaults_only_when_unambiguous(self):
|
||||
"""Role defaults are allowed only for clear author/reviewer profiles."""
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs-author": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "jcwalker3",
|
||||
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
||||
"execution_profile": "prgs-author",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
v2_data = migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
author = (
|
||||
v2_data["environments"]["prgs"]["services"]["gitea"]
|
||||
["identities"]["author"]
|
||||
)
|
||||
self.assertEqual(
|
||||
author["allowed_operations"],
|
||||
migrate_profiles.AUTHOR_DEFAULT_ALLOWED,
|
||||
)
|
||||
self.assertEqual(
|
||||
author["forbidden_operations"],
|
||||
migrate_profiles.AUTHOR_DEFAULT_FORBIDDEN,
|
||||
)
|
||||
|
||||
def test_ambiguous_permission_source_fails_closed(self):
|
||||
"""A profile without explicit permissions or clear role must not widen."""
|
||||
v1_data = {
|
||||
"version": 1,
|
||||
"profiles": {
|
||||
"prgs": {
|
||||
"base_url": "redacted-prgs-service",
|
||||
"username": "jcwalker3",
|
||||
"auth": {"type": "keychain", "id": "hidden-author-ref"},
|
||||
"execution_profile": "personal-prgs",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "fail closed"):
|
||||
migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
|
||||
def test_partial_permission_source_fails_closed(self):
|
||||
"""Allowed without forbidden, or vice versa, is ambiguous."""
|
||||
v1_data = json.loads(json.dumps(self.v1_content))
|
||||
del v1_data["profiles"]["prgs"]["forbidden_operations"]
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "fail closed"):
|
||||
migrate_profiles.migrate_v1_to_v2(v1_data)
|
||||
|
||||
def test_write_mode_and_backup(self):
|
||||
"""Verify that write mode creates a backup and correctly saves the validated config."""
|
||||
output_file = os.path.join(self.temp_dir, "migrated.json")
|
||||
backup_file = os.path.join(self.temp_dir, "profiles_backup.json.bak")
|
||||
|
||||
test_args = [
|
||||
"migrate_profiles.py",
|
||||
"-i", self.input_file,
|
||||
"-o", output_file,
|
||||
"--backup", backup_file,
|
||||
"-w"
|
||||
]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
migrate_profiles.main()
|
||||
|
||||
# Verify backup exists and matches original v1 config
|
||||
self.assertTrue(os.path.exists(backup_file))
|
||||
with open(backup_file, "r") as f:
|
||||
backup_data = json.load(f)
|
||||
self.assertEqual(backup_data["version"], 1)
|
||||
self.assertIn("prgs", backup_data["profiles"])
|
||||
|
||||
# Verify migrated v2 config exists and validates
|
||||
self.assertTrue(os.path.exists(output_file))
|
||||
with open(output_file, "r") as f:
|
||||
v2_data = json.load(f)
|
||||
self.assertEqual(v2_data["version"], 2)
|
||||
self.assertIn("environments", v2_data)
|
||||
self.assertEqual(v2_data["aliases"]["prgs"], "prgs.gitea.author")
|
||||
|
||||
def test_malformed_input_fails_safely(self):
|
||||
"""Test that malformed JSON or invalid version numbers cause a clean exit with code 1."""
|
||||
bad_json_file = os.path.join(self.temp_dir, "bad.json")
|
||||
with open(bad_json_file, "w") as f:
|
||||
f.write("{invalid-json}")
|
||||
|
||||
test_args = ["migrate_profiles.py", "-i", bad_json_file]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 1)
|
||||
|
||||
bad_version_file = os.path.join(self.temp_dir, "bad_version.json")
|
||||
with open(bad_version_file, "w") as f:
|
||||
json.dump({"version": 3, "profiles": {}}, f)
|
||||
|
||||
test_args = ["migrate_profiles.py", "-i", bad_version_file]
|
||||
with patch.object(sys, "argv", test_args):
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
migrate_profiles.main()
|
||||
self.assertEqual(cm.exception.code, 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user