Compare commits

..

9 Commits

Author SHA1 Message Date
a32ce4c8f9 feat(869-c4): activation version-coupling assertion (Part of #869)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
Part of #869

Mos (id-11) Gate-16 merge: independent APPROVE @90eb48fa (fail-closed identity locks byte-unchanged verified), author id2 != approver id11, clean mosaic-coder author, CI green wp1992. #869 Point-1 CODE COMPLETE (C1/C3/C5/C2/C4).

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 19:07:27 +00:00
d351caad36 feat(869-c2): install-ordering enforcement-hook guard (Part of #869)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
Part of #869

Mos (id-11) Gate-16 merge: independent APPROVE @b6f36564 (8/8, verified vs real production settings template), author id2 != approver id11, clean mosaic-coder author, CI green wp1988.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 18:48:33 +00:00
76b86a246e feat(869-c5): mosaic doctor activation-check (Part of #869)
Some checks failed
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was canceled
Part of #869

Mos (id-11) Gate-16 merge: independent APPROVE @e75e3238 (8/8), author id2 != approver id11, clean mosaic-coder author, CI green wp1987.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 18:38:21 +00:00
4422231bdb feat: per-agent Gitea identity resolution (#873)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
Closes #873

Mos (id-11) Gate-16 merge: independent APPROVE @4b472a22 (author-blocker dissolved via (a) re-author, identical tree hash to tech-approved head), author id2 != approver id11, clean mosaic-coder commit-author, CI green wp1985. Framework train COMPLETE 6/6.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 18:09:34 +00:00
8504216964 fix(pr-review): case-insensitive _belongs slug compare (#875)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
Closes #875

Mos (id-11) Gate-16 merge: independent APPROVE @9d8d58ae, author id2 != approver id11, clean mosaic-coder commit-author, CI green wp1982.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 17:53:26 +00:00
7edc9b3121 fix(gitea): direct REST comment/review with fail-closed read-back (#865)
All checks were successful
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was successful
Closes #865

Mos (id-11) Gate-16 merge: fresh confirmatory independent APPROVE @8ac7e70f (1241-case fuzz 0 fail-open), author id2 != approver id11, clean commit-author, CI green wp1966.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 17:25:32 +00:00
2f50c0876b feat(869-c3): lease-broker supervisor unit (Part of #869)
Some checks failed
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was canceled
Part of #869

Mos (id-11) Gate-16 merge: independent APPROVE @75235ef8 (9/9, no live host mutation), author id2 != approver id11, CI green wp1971.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 17:19:48 +00:00
db90da347e feat(869-c1): activation-capability probe (Part of #869)
Some checks failed
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was canceled
Part of #869

Mos (id-11) Gate-16 merge: independent 3-round APPROVE @c5a2bcc5, author id2 != approver id11, CI green wp1973.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 17:14:57 +00:00
48fd1df28a fix(ci-queue-wait): treat absent branch (404) as queue-clear (#872)
Some checks failed
ci/woodpecker/push/publish Pipeline was successful
ci/woodpecker/push/ci Pipeline was canceled
Closes #872

Mos (id-11) Gate-16 merge: independent review APPROVE @23cbdaf8, author jason.woltje(id2) != approver Mos(id11), CI green wp1974.

Co-authored-by: jason.woltje <jason@diversecanvas.com>
Co-committed-by: jason.woltje <jason@diversecanvas.com>
2026-07-23 17:08:57 +00:00
32 changed files with 7051 additions and 257 deletions

View File

@@ -18,12 +18,33 @@ set -Eeuo pipefail
# MOSAIC_INSTALL_MODE — prompt|keep|overwrite (default: prompt) # MOSAIC_INSTALL_MODE — prompt|keep|overwrite (default: prompt)
# MOSAIC_ALLOW_MISSING_SEQUENTIAL_THINKING — 1 to bypass MCP check # MOSAIC_ALLOW_MISSING_SEQUENTIAL_THINKING — 1 to bypass MCP check
# MOSAIC_SKIP_SKILLS_SYNC — 1 to skip skill sync # MOSAIC_SKIP_SKILLS_SYNC — 1 to skip skill sync
#
# Flags (CLI args, NOT environment variables — see #869 Point-1 C2):
# --allow-inactive-enforcement Explicit, per-invocation opt-out that lets the
# lease-enforcement hooks (mutator-gate.py,
# receipt-observer-client.py) be wired into
# ~/.claude/settings.json even when this host
# cannot confirm it can ACTIVATE them. Loud on
# use (see mosaic-link-runtime-assets). Default
# (flag absent) is fail-loud: the enforcement
# hooks are NOT wired and the framework's
# runtime-asset-link step reports a failure.
# ────────────────────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────────────────────
SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
TARGET_DIR="${MOSAIC_HOME:-$HOME/.config/mosaic}" TARGET_DIR="${MOSAIC_HOME:-$HOME/.config/mosaic}"
INSTALL_MODE="${MOSAIC_INSTALL_MODE:-prompt}" INSTALL_MODE="${MOSAIC_INSTALL_MODE:-prompt}"
# Deliberately parsed from "$@" (a real, explicit, per-invocation argument) —
# never an environment variable — so this opt-out can never sit silently
# inherited in a shell profile. See #869 Point-1 C2.
ALLOW_INACTIVE_ENFORCEMENT=0
for _arg in "$@"; do
case "$_arg" in
--allow-inactive-enforcement) ALLOW_INACTIVE_ENFORCEMENT=1 ;;
esac
done
# Shared framework path-ownership manifest reader (#791). Parity with # Shared framework path-ownership manifest reader (#791). Parity with
# packages/mosaic/src/framework/manifest.ts — both consume framework-manifest.txt. # packages/mosaic/src/framework/manifest.ts — both consume framework-manifest.txt.
# Sourcing does not run its CLI dispatch (guarded by BASH_SOURCE==$0). # Sourcing does not run its CLI dispatch (guarded by BASH_SOURCE==$0).
@@ -639,6 +660,10 @@ reconcile_framework_files
# Ensure tool scripts are executable # Ensure tool scripts are executable
find "$TARGET_DIR/tools" -name "*.sh" -exec chmod +x {} + 2>/dev/null || true find "$TARGET_DIR/tools" -name "*.sh" -exec chmod +x {} + 2>/dev/null || true
find "$TARGET_DIR/tools/_scripts" -type f -exec chmod +x {} + 2>/dev/null || true find "$TARGET_DIR/tools/_scripts" -type f -exec chmod +x {} + 2>/dev/null || true
# git-credential-mosaic (per-agent Gitea identity helper) ships without a .sh
# suffix — git resolves credential helpers by exact name/path, not extension —
# so the *.sh glob above does not cover it; chmod it explicitly.
[[ -f "$TARGET_DIR/tools/git/git-credential-mosaic" ]] && chmod +x "$TARGET_DIR/tools/git/git-credential-mosaic" 2>/dev/null || true
ok "Framework synced to $TARGET_DIR" ok "Framework synced to $TARGET_DIR"
@@ -666,10 +691,15 @@ step "Post-install tasks"
SCRIPTS="$TARGET_DIR/tools/_scripts" SCRIPTS="$TARGET_DIR/tools/_scripts"
if [[ -x "$SCRIPTS/mosaic-link-runtime-assets" ]]; then if [[ -x "$SCRIPTS/mosaic-link-runtime-assets" ]]; then
if "$SCRIPTS/mosaic-link-runtime-assets" >/dev/null 2>&1; then link_args=()
[[ "$ALLOW_INACTIVE_ENFORCEMENT" == "1" ]] && link_args+=(--allow-inactive-enforcement)
# stdout is suppressed as before, but stderr is left connected: the
# install-ordering guard's FAIL LOUD message (#869 Point-1 C2) must reach
# the operator, not be swallowed silently.
if "$SCRIPTS/mosaic-link-runtime-assets" "${link_args[@]}" >/dev/null; then
ok "Runtime assets linked" ok "Runtime assets linked"
else else
warn "Runtime asset linking failed (non-fatal)" warn "Runtime asset linking failed (non-fatal) — see message above for details."
fi fi
fi fi

View File

@@ -4,6 +4,22 @@ set -euo pipefail
MOSAIC_HOME="${MOSAIC_HOME:-$HOME/.config/mosaic}" MOSAIC_HOME="${MOSAIC_HOME:-$HOME/.config/mosaic}"
backup_stamp="$(date +%Y%m%d%H%M%S)" backup_stamp="$(date +%Y%m%d%H%M%S)"
# ─── Install-ordering guard opt-out (#869 Point-1 C2) ───────────────────────
# Explicit, per-invocation CLI flag ONLY — deliberately NOT read from an
# environment variable, so it can never sit as a silently-inherited default in
# a shell profile or CI env. Absent (the default) => hard fail-loud path.
allow_inactive_enforcement=0
for arg in "$@"; do
case "$arg" in
--allow-inactive-enforcement) allow_inactive_enforcement=1 ;;
esac
done
# Tracks whether the Claude settings install-ordering guard (below) reported a
# degraded (enforcement-not-wired) outcome, so this script's own exit status
# reflects it even though the rest of the runtime-asset sync must still run.
guard_degraded=0
copy_file_managed() { copy_file_managed() {
local src="$1" local src="$1"
local dst="$2" local dst="$2"
@@ -24,6 +40,103 @@ copy_file_managed() {
cp "$src" "$dst" cp "$src" "$dst"
} }
# ─── Install-ordering guard for settings.json (#869 Point-1 C2) ─────────────
#
# settings.json is where #828's enforcement hooks (PreToolUse mutator-gate.py,
# Stop receipt-observer-client.py) get wired unconditionally. Before copying
# it, delegate to `mosaic __link-claude-settings` (packages/mosaic/src/commands/
# install-ordering-guard.ts) so the wiring decision is made by importing the
# C1 activation probe (`leaseEnforcementActivatable()`) directly, rather than
# re-implementing the capability/supervisor checks in shell. That subcommand:
# - activatable -> writes settings.json with hooks intact, exits 0
# - NOT activatable -> writes settings.json with hooks STRIPPED,
# prints an actionable message, exits 1
# - NOT activatable + opt-out -> writes settings.json with hooks intact,
# prints a loud warning, exits 0
# The `mosaic` CLI is expected on PATH at this point ("No executables are
# placed on PATH — the mosaic npm CLI is the only binary", per install.sh).
# If it is not resolvable at all, that is itself strong evidence the
# activation half is absent, so the same fail-loud default applies via a
# minimal python3 fallback (this repo already depends on python3 for the
# lease broker itself).
copy_claude_settings_guarded() {
local src="$1"
local dst="$2"
local guard_args=(__link-claude-settings "$src" "$dst")
if [[ "$allow_inactive_enforcement" == "1" ]]; then
guard_args+=(--allow-inactive-enforcement)
fi
if command -v mosaic >/dev/null 2>&1; then
if mosaic "${guard_args[@]}"; then
return 0
fi
echo "[mosaic-link] Enforcement hooks were NOT wired into $dst (see message above)." >&2
guard_degraded=1
return 0
fi
echo "[mosaic-link] ERROR: 'mosaic' CLI not found on PATH — cannot confirm lease-enforcement" >&2
echo "[mosaic-link] activation capability. enforcement requested but activation half absent —" >&2
echo "[mosaic-link] needs a published CLI carrying launch-runtime activation + a broker" >&2
echo "[mosaic-link] supervisor; refusing to wire a dead gate (see #869)." >&2
if [[ "$allow_inactive_enforcement" == "1" ]]; then
echo "[mosaic-link] WARNING: --allow-inactive-enforcement set — wiring $dst AS-IS (with" >&2
echo "[mosaic-link] enforcement hooks) despite being unable to confirm activation." >&2
copy_file_managed "$src" "$dst"
return 0
fi
mkdir -p "$(dirname "$dst")"
if command -v python3 >/dev/null 2>&1; then
python3 - "$src" "$dst" <<'PYEOF'
import json, sys
src, dest = sys.argv[1], sys.argv[2]
with open(src) as f:
data = json.load(f)
hooks = data.get("hooks", {})
pre = hooks.get("PreToolUse", [])
hooks["PreToolUse"] = [
t for t in pre
if not any("mutator-gate.py" in h.get("command", "") for h in t.get("hooks", []))
]
if not hooks["PreToolUse"]:
del hooks["PreToolUse"]
stop = hooks.get("Stop", [])
new_stop = []
for t in stop:
kept = [h for h in t.get("hooks", []) if "receipt-observer-client.py" not in h.get("command", "")]
if kept:
t = dict(t)
t["hooks"] = kept
new_stop.append(t)
if new_stop:
hooks["Stop"] = new_stop
elif "Stop" in hooks:
del hooks["Stop"]
if hooks:
data["hooks"] = hooks
else:
data.pop("hooks", None)
with open(dest, "w") as f:
json.dump(data, f, indent=2)
f.write("\n")
PYEOF
else
cp "$src" "$dst"
fi
guard_degraded=1
return 0
}
remove_legacy_path() { remove_legacy_path() {
local p="$1" local p="$1"
@@ -110,6 +223,13 @@ for runtime_file in \
fi fi
src="$MOSAIC_HOME/runtime/claude/$runtime_file" src="$MOSAIC_HOME/runtime/claude/$runtime_file"
[[ -f "$src" ]] || continue [[ -f "$src" ]] || continue
if [[ "$runtime_file" == "settings.json" ]]; then
# Install-ordering guard (#869 Point-1 C2): gate enforcement-hook wiring
# on confirmed activation instead of the plain copy_file_managed used for
# every other runtime file. See copy_claude_settings_guarded() above.
copy_claude_settings_guarded "$src" "$HOME/.claude/$runtime_file"
continue
fi
copy_file_managed "$src" "$HOME/.claude/$runtime_file" copy_file_managed "$src" "$HOME/.claude/$runtime_file"
done done
@@ -167,3 +287,12 @@ fi
echo "[mosaic-link] Runtime assets synced (non-symlink mode)" echo "[mosaic-link] Runtime assets synced (non-symlink mode)"
echo "[mosaic-link] Canonical source: $MOSAIC_HOME" echo "[mosaic-link] Canonical source: $MOSAIC_HOME"
# Propagate the install-ordering guard's outcome (#869 Point-1 C2): every
# other runtime asset above is best-effort/non-fatal, but a degraded
# (enforcement-not-wired) settings.json must make THIS script's own exit
# status non-zero so callers (framework/install.sh, finalize.ts) can surface
# it — never silently.
if [[ "$guard_degraded" == "1" ]]; then
exit 1
fi

View File

@@ -0,0 +1,180 @@
#!/usr/bin/env bash
# Regression harness for issue #869 Point-1 C2 — the install-ordering guard
# wired into mosaic-link-runtime-assets.
#
# Root cause under test: mosaic-link-runtime-assets copies
# runtime/claude/settings.json (which embeds the PreToolUse mutator-gate.py
# hook and the Stop receipt-observer-client.py hook) straight into
# ~/.claude/settings.json, unconditionally. If the lease-broker activation
# half cannot be confirmed on this host, wiring those hooks bricks it with a
# fail-closed gate that can never be satisfied.
#
# This harness never invokes a real `mosaic` CLI build — it stubs the
# `__link-claude-settings` contract with a fake `mosaic` on PATH so the shell
# WIRING (does mosaic-link-runtime-assets call out correctly? does it
# propagate a degraded outcome? does it still copy every other runtime file?
# does --allow-inactive-enforcement forward through?) is exercised
# independently of the TS guard's own logic (already covered by
# install-ordering-guard.spec.ts). It also exercises the no-mosaic-on-PATH
# python3 fallback directly.
#
# Scenarios:
# 1. probe=true (fake mosaic exits 0) -> settings.json copied, script exits 0.
# 2. probe=false (fake mosaic exits 1) -> script exits 1 (guard_degraded
# propagated), but every OTHER runtime file is still copied.
# 3. probe=false + --allow-inactive-enforcement -> the flag is forwarded to
# the fake mosaic stub.
# 4. No `mosaic` on PATH at all (activation unconfirmable) -> the python3
# fallback strips the enforcement hooks itself and the script exits 1.
# 5. No `mosaic` on PATH + --allow-inactive-enforcement -> the python3
# fallback wires the hooks AS-IS and the script exits 0.
set -uo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LINK_SCRIPT="$SCRIPT_DIR/mosaic-link-runtime-assets"
TMP_ROOT=$(mktemp -d)
trap 'rm -rf "$TMP_ROOT"' EXIT
fail=0
fail_msg() {
echo "FAIL: $*" >&2
fail=1
}
FIXTURE_SETTINGS='{
"model": "opus",
"hooks": {
"PreToolUse": [
{ "matcher": ".*", "hooks": [ { "type": "command", "command": "python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude" } ] },
{ "matcher": "Write|Edit|MultiEdit", "hooks": [ { "type": "command", "command": "~/.config/mosaic/tools/qa/prevent-memory-write.sh" } ] }
],
"Stop": [
{ "hooks": [
{ "type": "command", "command": "python3 ~/.config/mosaic/tools/lease-broker/receipt-observer-client.py --runtime claude" },
{ "type": "command", "command": "~/.config/mosaic/tools/qa/reflect-stop-hook.sh" }
] }
]
}
}'
# Sets up a fresh $MOSAIC_HOME/runtime/claude/{settings.json,CLAUDE.md,
# hooks-config.json,context7-integration.md} + fresh $HOME, echoes both paths
# space-separated for the caller to `read`.
new_scenario_dirs() {
local scenario="$1"
local base="$TMP_ROOT/$scenario"
local mosaic_home="$base/mosaic-home"
local home="$base/home"
mkdir -p "$mosaic_home/runtime/claude" "$home"
printf '%s' "$FIXTURE_SETTINGS" > "$mosaic_home/runtime/claude/settings.json"
echo "claude.md fixture" > "$mosaic_home/runtime/claude/CLAUDE.md"
echo '{"hooks":{}}' > "$mosaic_home/runtime/claude/hooks-config.json"
echo "context7 fixture" > "$mosaic_home/runtime/claude/context7-integration.md"
echo "$mosaic_home" "$home"
}
settings_has_marker() {
local file="$1" marker="$2"
[[ -f "$file" ]] && grep -q "$marker" "$file"
}
# A fake `mosaic` binary implementing only the __link-claude-settings contract
# this harness needs: writes dest verbatim (fixture is unmodified either way —
# this stub only exercises the CALL CONTRACT, not the TS strip logic, which
# has its own vitest coverage) and exits with the code the scenario wants.
# Records the args it was called with so the harness can assert forwarding.
make_fake_mosaic() {
local bin_dir="$1" exit_code="$2"
mkdir -p "$bin_dir"
cat > "$bin_dir/mosaic" <<EOF
#!/usr/bin/env bash
set -euo pipefail
echo "\$@" > "$bin_dir/mosaic.args"
if [[ "\$1" == "__link-claude-settings" ]]; then
cp "\$2" "\$3"
exit $exit_code
fi
exit 0
EOF
chmod +x "$bin_dir/mosaic"
}
# --- Scenario 1: probe=true (fake mosaic exits 0) ---------------------------
read -r MOSAIC_HOME_1 HOME_1 < <(new_scenario_dirs scenario1)
BIN_1="$TMP_ROOT/scenario1/bin"
make_fake_mosaic "$BIN_1" 0
OUTPUT=$(MOSAIC_HOME="$MOSAIC_HOME_1" HOME="$HOME_1" PATH="$BIN_1:$PATH" "$LINK_SCRIPT" 2>&1)
STATUS=$?
[[ "$STATUS" -eq 0 ]] || fail_msg "scenario1 (probe=true): expected exit 0, got $STATUS. Output: $OUTPUT"
[[ -f "$HOME_1/.claude/settings.json" ]] || fail_msg "scenario1: settings.json was not copied"
# --- Scenario 2: probe=false (fake mosaic exits 1) --------------------------
read -r MOSAIC_HOME_2 HOME_2 < <(new_scenario_dirs scenario2)
BIN_2="$TMP_ROOT/scenario2/bin"
make_fake_mosaic "$BIN_2" 1
OUTPUT=$(MOSAIC_HOME="$MOSAIC_HOME_2" HOME="$HOME_2" PATH="$BIN_2:$PATH" "$LINK_SCRIPT" 2>&1)
STATUS=$?
[[ "$STATUS" -ne 0 ]] || fail_msg "scenario2 (probe=false, default): expected non-zero exit, got 0. Output: $OUTPUT"
[[ -f "$HOME_2/.claude/CLAUDE.md" ]] || fail_msg "scenario2: CLAUDE.md was NOT copied even though it is independent of the settings.json guard"
[[ -f "$HOME_2/.claude/hooks-config.json" ]] || fail_msg "scenario2: hooks-config.json was NOT copied"
[[ -f "$HOME_2/.claude/context7-integration.md" ]] || fail_msg "scenario2: context7-integration.md was NOT copied"
case "$OUTPUT" in
*"NOT be wired"*|*"NOT wired"*) ;;
*) fail_msg "scenario2: expected an actionable degraded-wiring message in output, got: $OUTPUT" ;;
esac
# --- Scenario 3: probe=false + --allow-inactive-enforcement forwards the flag
read -r MOSAIC_HOME_3 HOME_3 < <(new_scenario_dirs scenario3)
BIN_3="$TMP_ROOT/scenario3/bin"
make_fake_mosaic "$BIN_3" 0
MOSAIC_HOME="$MOSAIC_HOME_3" HOME="$HOME_3" PATH="$BIN_3:$PATH" "$LINK_SCRIPT" --allow-inactive-enforcement >/dev/null 2>&1
RECORDED_ARGS="$(cat "$BIN_3/mosaic.args" 2>/dev/null || true)"
case "$RECORDED_ARGS" in
*"--allow-inactive-enforcement"*) ;;
*) fail_msg "scenario3: --allow-inactive-enforcement was not forwarded to the mosaic CLI invocation (got: '$RECORDED_ARGS')" ;;
esac
# --- Scenario 4: no `mosaic` on PATH at all -> python3 fallback strips hooks
read -r MOSAIC_HOME_4 HOME_4 < <(new_scenario_dirs scenario4)
EMPTY_BIN="$TMP_ROOT/scenario4/empty-bin"
mkdir -p "$EMPTY_BIN"
# A PATH containing only python3 (for the fallback) + core utils, no mosaic.
FALLBACK_PATH="$EMPTY_BIN:/usr/bin:/bin"
OUTPUT=$(MOSAIC_HOME="$MOSAIC_HOME_4" HOME="$HOME_4" PATH="$FALLBACK_PATH" "$LINK_SCRIPT" 2>&1)
STATUS=$?
[[ "$STATUS" -ne 0 ]] || fail_msg "scenario4 (no mosaic on PATH, default): expected non-zero exit, got 0. Output: $OUTPUT"
if settings_has_marker "$HOME_4/.claude/settings.json" "mutator-gate.py"; then
fail_msg "scenario4: mutator-gate.py hook was wired even though mosaic could not be resolved (activation unconfirmable)"
fi
if settings_has_marker "$HOME_4/.claude/settings.json" "receipt-observer-client.py"; then
fail_msg "scenario4: receipt-observer-client.py hook was wired even though mosaic could not be resolved"
fi
if ! settings_has_marker "$HOME_4/.claude/settings.json" "prevent-memory-write.sh"; then
fail_msg "scenario4: the unrelated prevent-memory-write.sh hook was incorrectly dropped too"
fi
# --- Scenario 5: no `mosaic` on PATH + --allow-inactive-enforcement --------
read -r MOSAIC_HOME_5 HOME_5 < <(new_scenario_dirs scenario5)
OUTPUT=$(MOSAIC_HOME="$MOSAIC_HOME_5" HOME="$HOME_5" PATH="$FALLBACK_PATH" "$LINK_SCRIPT" --allow-inactive-enforcement 2>&1)
STATUS=$?
[[ "$STATUS" -eq 0 ]] || fail_msg "scenario5 (no mosaic, opt-out): expected exit 0, got $STATUS. Output: $OUTPUT"
if ! settings_has_marker "$HOME_5/.claude/settings.json" "mutator-gate.py"; then
fail_msg "scenario5: mutator-gate.py hook should have been wired (explicit opt-out set)"
fi
case "$OUTPUT" in
*"WARNING"*"--allow-inactive-enforcement"*) ;;
*) fail_msg "scenario5: expected a loud WARNING mentioning --allow-inactive-enforcement, got: $OUTPUT" ;;
esac
if [[ "$fail" -eq 0 ]]; then
echo "install-ordering-guard regression passed (5/5 scenarios)"
fi
exit "$fail"

View File

@@ -4,6 +4,93 @@ These scripts provide host-aware GitHub and Gitea issue, pull-request, milestone
## Durable review provenance ## Durable review provenance
A successful provider write command—or a wrapper message based only on that command's exit code—is **not** durable review provenance. Review comments count as durable provenance only after the wrapper reads the created provider record back and verifies that it belongs to the intended repository and pull request and contains the exact submitted body (or verifies the provider-returned record ID). A successful provider write command—or a wrapper message based only on that command's exit code—is **not** durable review provenance. Review comments, approvals, and change requests count as durable provenance only after the wrapper reads the created provider record back and verifies that it was created by _this_ write.
`pr-review.sh` therefore fails closed when a Gitea comment cannot be written, its created comment ID cannot be identified, or provider read-back does not match. It reports comment success only after that read-back verification passes. **The write is a direct Gitea REST `POST` that returns the created record's id.** Neither wrapper writes through `tea` — tea 0.11.1 can silently no-op while exiting 0 and cannot emit the id of a record it creates, so its exit code is worthless as proof of a durable write (#865). Instead:
- Comments (`issue-comment.sh`, and the `comment` action of `pr-review.sh`) `POST /api/v1/repos/{owner}/{repo}/issues/{index}/comments`, requiring a `201` and parsing the created comment's `id` from the response body.
- Reviews (`approve` / `request-changes`) `POST /api/v1/repos/{owner}/{repo}/pulls/{index}/reviews` with the `event` (`APPROVED` / `REQUEST_CHANGES`), the review `body`, and `commit_id` pinned to the PR's current head, then parse the created review's `id`. The review body travels _in the review submit itself_ — there is no separate detached comment to reconcile (a Gitea `REQUEST_CHANGES` review requires a non-empty body, which the submit carries).
**Verification keys on that exact provider-returned id.** The wrapper then `GET`s that one record directly — `GET /issues/comments/{id}` or `GET /pulls/{n}/reviews/{id}` — and requires that its `id` equals the created id, its **author login equals the acting identity** (resolved via `GET /api/v1/user` for the token in use), and, for comments, its body exactly matches what was submitted **and its returned web URL belongs to this exact provider and repository** (the `issue_url` / `pull_request_url` origin — scheme, host, and effective port — and full path, i.e. deployment prefix + exact `owner/repo` + kind + number, must match; a suffix/`endsWith` test would accept a look-alike host or a decoy path prefix, so the whole normalized URL is compared). The `comment` action of `pr-review.sh` additionally requires the returned resource be a **pull request** (a populated `pull_request_url`); a bare `issue_url` is rejected, so if issue `#N` exists but PR `#N` does not, an issue comment cannot be reported as a verified PR comment. (`issue-comment.sh` legitimately keeps the broader issue-or-PR acceptance.) For reviews, its state matches the requested action, its reviewed `commit_id` equals the PR head, **and its persisted body equals the submitted body** — an exact, presence- and type-checked equality (a missing/`null` persisted body no longer counts as an empty match), because Gitea can finalize/reuse a pending review id whose stored content was authored elsewhere, so the body is bound too. The write, the `/user` identity lookup, and the read-back all use the **same** credential — the effective login's token, or the host credential when no login is named — so the write is verified against the identity that actually performed it.
**A review's pinned head is re-checked after verification (current-head TOCTOU).** The `commit_id` is pinned to the PR head read _before_ the submit; between that read and the read-back the branch could advance (a force-push or a new commit), leaving a verified review attached to a now-superseded commit while the live tip carries unreviewed code. After the exact-id read-back succeeds, the wrapper re-reads the live PR head (`GET …/pulls/{n}`) and requires it still equals the submitted SHA; if the head advanced it fails closed (non-zero, no success line) rather than reporting a review that no longer covers the PR's current commit.
**This closes the concurrency window rather than documenting it.** Because verification keys on the id the create returned, a no-op create yields no id and fails closed with no list-scan fallback, and a _concurrent_ record — even one written by the _same_ identity with an identical body/state — has a _different_ id and cannot be mistaken for this write. There is no residual same-identity window: the earlier boundary-and-author heuristic (accept any `id > pre-write-max` with a matching author) is replaced entirely by exact-id attribution.
**Exact-id read-back is the sole authority.** Verification is a direct `GET` of the one record the create returned; there is no follow-up list enumeration. An earlier redundant pass that re-listed the record's page (`?limit=&page=1,2,…`) was removed: server-capped page sizes and list-pagination quirks made it a false-failure source (a durable, exact-id-verified record could be missed by a non-exhaustive enumeration), and it added nothing over the authoritative exact-id `GET`.
## Credential handling
The Gitea API token is **never passed on a curl command line.** An `Authorization: token <value>` argument would be visible to any local process that can read the process table (`ps` / `/proc/<pid>/cmdline`) for the lifetime of the request. Instead, every authenticated curl call writes the header into a private, mode-`0600` config file under `$TMPDIR` and passes it with `curl --config <file>` (`gitea_write_auth_config`), so only the file _path_ — never the token — appears in argv. Each such file is unlinked on every exit path (success and failure) by the caller's `RETURN` trap.
## `tea` invocation notes (Gitea)
- tea v0.11.1 has **no `comment` subcommand under `tea pr` or `tea issue`** — the `tea pr comment` / `tea issue comment` forms don't error, they silently fall through to a no-op and still exit 0, producing a false-success write (#865). tea's write subcommands (`tea comment`, `tea pr approve`/`reject`) also cannot report the id of the record they create, so their exit code cannot prove a durable write. These wrappers therefore do **not** write reviews or comments through `tea` at all; they use direct Gitea REST `POST`s that return the created record's id (see "Durable review provenance" above). `tea` is consulted only to enumerate the login list for host→login resolution.
- Because the review body is carried in the `POST …/reviews` submit itself, there is no separate detached review comment, and the historical `tea pr approve`/`reject` trailing-positional-argument vs. nonexistent `--comment`/`-comment` flag hazard (#835) no longer applies to these wrappers — no review comment is ever passed to `tea`.
### `--login` override
Both `pr-review.sh` and `issue-comment.sh` accept an optional `--login <name>` flag that overrides the automatically detected Gitea login for that single invocation. The override selects **which credential the REST write, the `/user` identity lookup, and the read-back all use** — its token is resolved from the tea config for that login name (`get_gitea_token_for_login`), falling back to the repo host's credential when no login is named. The resolved login is **host- and port-bound**: the login's configured URL host **and effective port** (the scheme's default port — 80 for `http`, 443 for `https` — applies when a port is omitted, symmetrically on both sides) must match the repo remote's, so a login name shared across hosts (or an override configured for a different Gitea, including one on a different port of the same host) can never send one host's credential to another — a host or port mismatch fails closed rather than leaking a cross-host token. Resolving the acting identity and the read-back from the _same_ login that performs the write is essential: a write performed under an overridden login must be verified against that login's identity, not the host default's. Callers who need a different login than the host default should pass `--login <reviewer-login>`.
As a durable successor to this mechanism, consider giving each reviewer/approver slot its own dedicated Gitea login credential, so that author≠reviewer holds at the credential level rather than relying on wrapper-level `--login` bookkeeping. This is a recommendation for future hardening, not something implemented by this flag.
## Per-agent Gitea identity (Gate-16 author≠reviewer)
By default, git push/fetch (via `git-credential-mosaic`) and the API wrappers above (via
`detect-platform.sh`'s `get_gitea_token`) all authenticate as the single shared Gitea
account/token configured through `tools/_lib/credentials.sh`. That means every agent in a
fleet commits, pushes, and opens PRs under one identity — with no cryptographic
separation between an author and a reviewer.
Both `git-credential-mosaic` and `get_gitea_token()` resolve an optional **per-agent
identity** before falling back to the shared account:
1. `MOSAIC_GIT_IDENTITY` environment variable, or
2. `git config --get mosaic.gitIdentity` (set per-worktree; persists on disk across
non-persistent shells — `git config mosaic.gitIdentity <agent-id>`), or
3. (git-credential-mosaic only) the username git itself supplies for the credential
request.
If the resolved identity has a token file at
`~/.config/mosaic/secrets/gitea-tokens/gitea-{usc,mosaicstack}-<agent-id>.token`, that
identity + token is used. **Nothing configured → nothing changes**: with no per-slot
token file present, both tools fall through to the existing shared-account path
unchanged, so this feature is a no-op on any host that hasn't provisioned per-slot
tokens.
### Enabling it for a clone
The framework installer syncs `git-credential-mosaic` to
`~/.config/mosaic/tools/git/git-credential-mosaic` (executable) on every install/update,
but does **not** register it as git's credential helper automatically. Registration is a
one-time, explicit step:
```bash
# Per-repo (recommended — scopes the helper to this clone only):
git config credential.helper "$HOME/.config/mosaic/tools/git/git-credential-mosaic"
# Per-worktree identity pin (Gate-16 separation):
git config mosaic.gitIdentity <agent-id>
```
This is deliberately **not** auto-registered on install/update: `credential.helper` is
global, order-sensitive git config (`~/.gitconfig`) that can already hold an
operator-chosen credential manager (keychain, `store`, `manager-core`, …) for
repositories unrelated to Mosaic. Silently inserting an entry on every framework
install/upgrade risks reordering or shadowing that operator-owned surface across the
whole host — the same operator-owned config the installer's manifest system is
otherwise careful never to touch. Because identity is already resolved per-worktree
(`mosaic.gitIdentity`), the correct granularity for registering the helper is per-clone
too, so a documented manual step is the right shape here, not a global auto-write.
### PowerShell parity
`detect-platform.ps1`'s Gitea wrappers authenticate through `tea` CLI logins
(`Get-GiteaLoginForHost`), not a raw-token `get_gitea_token`-equivalent function — there
is nothing to prepend the identity-resolution block to on the PowerShell side. A native
PowerShell git-credential helper is also unnecessary: `git-credential-mosaic` is invoked
by git's credential-helper protocol (stdin/stdout), which works identically under Git for
Windows' bundled `bash`/`sh` when configured via `credential.helper`, without a `.ps1`
counterpart. A `tea`-login-based per-agent identity for the PowerShell wrappers is a
separate, larger design (mapping identities to `tea login` profiles) and is out of scope
here.

View File

@@ -185,6 +185,16 @@ switch ($platform) {
$headSha = ($branchPayload.commit.id | Out-String).Trim() $headSha = ($branchPayload.commit.id | Out-String).Trim()
} }
catch { catch {
# A not-yet-pushed feature branch has no in-flight pipeline, so the
# pre-push queue guard must treat 404 as "queue clear", not crash.
$statusCode = $null
if ($_.Exception.Response) {
$statusCode = [int]$_.Exception.Response.StatusCode
}
if ($statusCode -eq 404) {
Write-Host "[ci-queue-wait] branch $Branch not yet on remote — no in-flight pipeline; queue clear."
exit 0
}
Write-Error "Could not resolve $Branch head SHA from Gitea API." Write-Error "Could not resolve $Branch head SHA from Gitea API."
exit 1 exit 1
} }

View File

@@ -137,7 +137,21 @@ gitea_get_branch_head_sha() {
local branch="$3" local branch="$3"
local token="$4" local token="$4"
local url="https://${host}/api/v1/repos/${repo}/branches/${branch}" local url="https://${host}/api/v1/repos/${repo}/branches/${branch}"
curl -fsSL -H "User-Agent: curl/8" -H "Authorization: token ${token}" "$url" | python3 -c ' # Capture HTTP status so an absent branch (404) is distinguished from an API
# error. A not-yet-pushed feature branch has no in-flight pipeline, so the
# pre-push queue guard must treat 404 as "queue clear", not crash.
local resp code body
resp=$(curl -sS -H "User-Agent: curl/8" -H "Authorization: token ${token}" -w $'\n%{http_code}' "$url")
code="${resp##*$'\n'}"
body="${resp%$'\n'*}"
if [[ "$code" == "404" ]]; then
echo "__BRANCH_ABSENT__"
return 0
fi
if [[ "$code" != "200" ]]; then
return 1
fi
printf '%s' "$body" | python3 -c '
import json, sys import json, sys
data = json.load(sys.stdin) data = json.load(sys.stdin)
commit = data.get("commit") or {} commit = data.get("commit") or {}
@@ -219,6 +233,10 @@ elif [[ "$PLATFORM" == "gitea" ]]; then
exit 1 exit 1
} }
HEAD_SHA=$(gitea_get_branch_head_sha "$HOST" "$OWNER/$REPO" "$BRANCH" "$TOKEN") HEAD_SHA=$(gitea_get_branch_head_sha "$HOST" "$OWNER/$REPO" "$BRANCH" "$TOKEN")
if [[ "$HEAD_SHA" == "__BRANCH_ABSENT__" ]]; then
echo "[ci-queue-wait] branch ${BRANCH} not yet on remote — no in-flight pipeline; queue clear."
exit 0
fi
if [[ -z "$HEAD_SHA" ]]; then if [[ -z "$HEAD_SHA" ]]; then
echo "Error: Could not resolve ${BRANCH} head SHA." >&2 echo "Error: Could not resolve ${BRANCH} head SHA." >&2
exit 1 exit 1

View File

@@ -505,6 +505,28 @@ get_gitea_token() {
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
local cred_loader="$script_dir/../_lib/credentials.sh" local cred_loader="$script_dir/../_lib/credentials.sh"
# 0. Per-agent identity (Gate-16 author≠reviewer). If MOSAIC_GIT_IDENTITY, or the
# per-worktree `git config mosaic.gitIdentity`, resolves to an agent that has a
# stored per-slot token for this host, act AS that agent so API tooling
# (pr-create, issue-create, …) authors under the right identity — matching the
# git credential helper. Backward-compatible: nothing resolvable → shared logic below.
local _ident="${MOSAIC_GIT_IDENTITY:-}"
[[ -z "$_ident" ]] && _ident="$(git config --get mosaic.gitIdentity 2>/dev/null || true)"
if [[ -n "$_ident" ]]; then
local _idpfx=""
case "$host" in
git.uscllc.com) _idpfx=gitea-usc ;;
git.mosaicstack.dev) _idpfx=gitea-mosaicstack ;;
esac
if [[ -n "$_idpfx" ]]; then
local _idtok="$HOME/.config/mosaic/secrets/gitea-tokens/${_idpfx}-${_ident}.token"
if [[ -r "$_idtok" ]]; then
cat "$_idtok"
return 0
fi
fi
fi
# 1. Mosaic credential loader (host → service mapping, run in subshell to avoid polluting env) # 1. Mosaic credential loader (host → service mapping, run in subshell to avoid polluting env)
if [[ -f "$cred_loader" ]]; then if [[ -f "$cred_loader" ]]; then
local token local token
@@ -563,6 +585,873 @@ get_gitea_token() {
return 1 return 1
} }
# Stage the Gitea bearer credential for curl OUTSIDE the process argument vector.
# Passing "-H 'Authorization: token <value>'" on the curl command line exposes
# the token to anyone who can read the process table (ps / /proc/<pid>/cmdline)
# for the lifetime of the request. Instead, write the header into a private
# (mode 0600) curl config file and have callers pass it with `curl --config`, so
# only the FILE PATH — never the token — appears in argv. Prints the temp file
# path on success; the caller OWNS the file and MUST remove it on every exit
# path (success and failure). $1 = bearer token. Callers must not log the token.
gitea_write_auth_config() {
local token="$1" auth_file
auth_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-gitea-auth.XXXXXX") || return 1
# mktemp already creates the file with 0600; be explicit in case of an
# unusual umask so the credential is never briefly group/other readable.
chmod 600 "$auth_file" 2>/dev/null || true
# `header = "..."` is curl's config syntax for an extra request header. Only
# this filename reaches curl's argv; the token stays on disk, readable solely
# by this user, and is unlinked by the caller's trap after the request.
if ! printf 'header = "Authorization: token %s"\n' "$token" > "$auth_file"; then
rm -f "$auth_file"
return 1
fi
printf '%s' "$auth_file"
}
# Resolve the API token for a SPECIFIC tea login name from tea's own config
# (the same store tea itself writes/reads for `--login <name>`). This is what
# lets a REST write be performed AS the selected --login identity: tea keys its
# per-login tokens by `name` in $XDG_CONFIG_HOME/tea/config.yml (default
# ~/.config/tea/config.yml), exactly as the `tea` CLI resolves them, so a
# --login override and its REST read-back bind to the SAME credential/identity.
#
# $2 (repo host) binds the selected credential to the TARGET host: a tea login
# also records the `url` it authenticates against, and the matched login's URL
# host MUST equal the repo host. This fails closed when an override login is
# configured for a DIFFERENT host than the repo remote, so a login name shared
# across hosts (or a mistargeted override) can never send one host's credential
# to another host (cross-host credential leak). When $2 is empty the host bind
# is skipped (host-agnostic lookup) — callers that write should always pass it.
#
# Prints the token on success; returns non-zero (no output) if the config, a
# matching login token, or the host bind cannot be satisfied. Callers must not
# log the result.
get_gitea_token_for_login() {
local login_name="$1" repo_host="${2:-}" config_file
[[ -n "$login_name" ]] || return 1
config_file="${XDG_CONFIG_HOME:-$HOME/.config}/tea/config.yml"
[[ -f "$config_file" ]] || return 1
LOGIN_NAME="$login_name" REPO_HOST="$repo_host" python3 - "$config_file" <<'PY'
import datetime
import os
import re
import sys
from urllib.parse import urlparse
wanted = os.environ["LOGIN_NAME"]
repo_host = os.environ.get("REPO_HOST", "").strip().lower()
config_path = sys.argv[1]
# PyYAML 6.0.3 SafeLoader implicit resolver patterns (YAML 1.1). An UNQUOTED
# plain scalar matching any of these is resolved by PyYAML to a NON-string type
# (null->None, bool, int, float, timestamp->date/datetime); a quoted scalar is
# ALWAYS a string. These mirror yaml/resolver.py so the PyYAML-absent fallback
# types unquoted scalars exactly as PyYAML would (see _implicit_nonstring).
_IMPLICIT_NULL = re.compile(r"^(?:~|null|Null|NULL|)$")
_IMPLICIT_BOOL = re.compile(
r"^(?:yes|Yes|YES|no|No|NO|true|True|TRUE|false|False|FALSE"
r"|on|On|ON|off|Off|OFF)$"
)
_IMPLICIT_INT = re.compile(
r"^(?:[-+]?0b[0-1_]+"
r"|[-+]?0[0-7_]+"
r"|[-+]?(?:0|[1-9][0-9_]*)"
r"|[-+]?0x[0-9a-fA-F_]+"
r"|[-+]?[1-9][0-9_]*(?::[0-5]?[0-9])+)$"
)
# NOTE: the exponent group requires an EXPLICIT sign ([eE][-+][0-9]+), matching
# PyYAML 6.0.3's resolver.py float regex exactly. An unsigned exponent (e.g.
# "1.0e10", ".5e10", "4.e8") is NOT matched by PyYAML's implicit float resolver
# -- PyYAML resolves those as plain strings -- so this pattern must not match
# them either, or the fallback over-rejects a token PyYAML would emit verbatim.
_IMPLICIT_FLOAT = re.compile(
r"^(?:[-+]?(?:[0-9][0-9_]*)\.[0-9_]*(?:[eE][-+][0-9]+)?"
r"|\.[0-9][0-9_]*(?:[eE][-+][0-9]+)?"
r"|[-+]?[0-9][0-9_]*(?::[0-5]?[0-9])+\.[0-9_]*"
r"|[-+]?\.(?:inf|Inf|INF)"
r"|\.(?:nan|NaN|NAN))$"
)
_IMPLICIT_TIMESTAMP = re.compile(
r"^(?:[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]"
r"|[0-9][0-9][0-9][0-9]-[0-9][0-9]?-[0-9][0-9]?"
r"(?:[Tt]|[ \t]+)[0-9][0-9]?"
r":[0-9][0-9]:[0-9][0-9](?:\.[0-9]*)?"
r"(?:[ \t]*(?:Z|[-+][0-9][0-9]?(?::[0-9][0-9])?))?)$"
)
def _implicit_nonstring(text):
# True when an UNQUOTED plain scalar would be resolved by PyYAML's SafeLoader
# to a non-string type (null/bool/int/float/timestamp). Fuzzed against real
# PyYAML 6.0.3: it never returns False where PyYAML types the scalar as a
# non-string (i.e. never fail-open), and (post exponent-sign fix) no longer
# over-rejects unsigned-exponent spellings like "4.e8" or "1.0e10" -- those
# match PyYAML's own implicit float resolver exactly (explicit sign only),
# so PyYAML keeps them as strings and this function now agrees.
return bool(
_IMPLICIT_NULL.match(text)
or _IMPLICIT_BOOL.match(text)
or _IMPLICIT_INT.match(text)
or _IMPLICIT_FLOAT.match(text)
or _IMPLICIT_TIMESTAMP.match(text)
)
# PyYAML 6.0.3 SafeConstructor timestamp regexp (yaml/constructor.py). The
# constructor RE-parses a timestamp-tagged scalar with THIS pattern and then
# builds a datetime.date/datetime, which raises ValueError for an out-of-range
# calendar field (e.g. month 99, hour 25). _IMPLICIT_TIMESTAMP (the RESOLVER
# pattern) is byte-identical to PyYAML's resolver, so anything it tags is also
# tagged by PyYAML and re-matched here.
_TIMESTAMP_CONSTRUCT = re.compile(
r"""^(?P<year>[0-9][0-9][0-9][0-9])
-(?P<month>[0-9][0-9]?)
-(?P<day>[0-9][0-9]?)
(?:(?:[Tt]|[ \t]+)
(?P<hour>[0-9][0-9]?)
:(?P<minute>[0-9][0-9])
:(?P<second>[0-9][0-9])
(?:\.(?P<fraction>[0-9]*))?
(?:[ \t]*(?P<tz>Z|(?P<tz_sign>[-+])(?P<tz_hour>[0-9][0-9]?)
(?::(?P<tz_minute>[0-9][0-9]))?))?)?$""",
re.X,
)
def _int_constructible(text):
# Replicate PyYAML SafeConstructor.construct_yaml_int and report whether it
# would succeed. A resolver-tagged int whose radix body is empty after
# underscore removal (e.g. "0b_", "0x_", "0x__") makes int(base) raise, so
# PyYAML fails the WHOLE document -> the fallback must fail closed too.
value = text.replace("_", "")
if value[:1] in ("+", "-"):
value = value[1:]
if value == "0":
return True
try:
if value.startswith("0b"):
int(value[2:], 2)
elif value.startswith("0x"):
int(value[2:], 16)
elif value[:1] == "0":
int(value, 8)
elif ":" in value:
[int(part) for part in value.split(":")]
else:
int(value)
except ValueError:
return False
return True
def _float_constructible(text):
# Replicate PyYAML SafeConstructor.construct_yaml_float. Retained for the
# WHOLE-document invariant: _IMPLICIT_FLOAT now matches PyYAML's resolver
# pattern exactly (explicit-sign exponent only), and every scalar it tags
# is float()-constructible, so this never fails closed where PyYAML would
# emit a token.
value = text.replace("_", "").lower()
if value[:1] in ("+", "-"):
value = value[1:]
if value in (".inf", ".nan"):
return True
try:
if ":" in value:
[float(part) for part in value.split(":")]
else:
float(value)
except ValueError:
return False
return True
def _timestamp_constructible(text):
# Replicate PyYAML SafeConstructor.construct_yaml_timestamp: build the same
# datetime.date/datetime and report whether it raises. Returns False for an
# out-of-range calendar field (month/day/hour/...), matching PyYAML's
# whole-document ValueError.
match = _TIMESTAMP_CONSTRUCT.match(text)
if not match:
return False
values = match.groupdict()
try:
year = int(values["year"])
month = int(values["month"])
day = int(values["day"])
if not values["hour"]:
datetime.date(year, month, day)
return True
hour = int(values["hour"])
minute = int(values["minute"])
second = int(values["second"])
fraction = 0
if values["fraction"]:
frac = values["fraction"][:6]
frac += "0" * (6 - len(frac))
fraction = int(frac)
tzinfo = None
if values["tz_sign"]:
tz_hour = int(values["tz_hour"])
tz_minute = int(values["tz_minute"] or 0)
delta = datetime.timedelta(hours=tz_hour, minutes=tz_minute)
if values["tz_sign"] == "-":
delta = -delta
tzinfo = datetime.timezone(delta)
elif values["tz"]:
tzinfo = datetime.timezone.utc
datetime.datetime(
year, month, day, hour, minute, second, fraction, tzinfo=tzinfo
)
except (ValueError, OverflowError):
return False
return True
def _constructible(text):
# WHOLE-DOCUMENT INVARIANT: the fallback resolves the SAME login token as
# PyYAML safe_load or fails closed -- never less conservative -- INCLUDING
# when PyYAML raises a CONSTRUCTOR error anywhere in the document. A plain
# scalar can match a typed implicit resolver (int/float/timestamp) yet NOT be
# constructible (e.g. 2023-99-99, 0b_, 0x_); PyYAML then raises on the whole
# load and yields no token, so the fallback MUST fail closed for the whole
# document too. This returns True only when PyYAML's constructor would build
# the scalar (null/bool token sets are always constructible), else False so
# the caller fails closed. `text` is assumed to satisfy _implicit_nonstring.
if _IMPLICIT_NULL.match(text) or _IMPLICIT_BOOL.match(text):
return True
if _IMPLICIT_TIMESTAMP.match(text):
return _timestamp_constructible(text)
if _IMPLICIT_INT.match(text):
return _int_constructible(text)
if _IMPLICIT_FLOAT.match(text):
return _float_constructible(text)
return True
# PyYAML's Reader.check_printable scans the ENTIRE raw input stream (not just
# scalar contents, and NOT scoped by quoting) for code points outside its
# printable set and raises ReaderError -- a whole-document reject -- the instant
# one is found, no matter where it sits (an unrelated field, a comment, inside or
# outside quotes). To guarantee parity WITHOUT a hand-rolled subset (which missed
# the C1 block and BMP noncharacters), this is PyYAML 6.0.3's EXACT Reader
# printable definition, negated verbatim:
# Reader.NON_PRINTABLE =
# re.compile('[^\x09\x0A\x0D\x20-\x7E\x85\xA0-퟿-<2D>'
# '\U00010000-\U0010FFFF]')
# i.e. PRINTABLE = {TAB(0x09), LF(0x0A), CR(0x0D), 0x20-0x7E, NEL(0x85),
# 0xA0-0xD7FF, 0xE000-0xFFFD, 0x10000-0x10FFFF}; EVERYTHING else (all other C0
# controls, DEL 0x7F, the entire C1 block 0x80-0x84/0x86-0x9F, the surrogate
# range 0xD800-0xDFFF, and the BMP noncharacters 0xFFFE/0xFFFF) is non-printable
# -> whole-document fail closed. Verified empirically against real PyYAML 6.0.3
# in both directions: the C1 bytes and 0xFFFE/0xFFFF reject; NEL(0x85), 0xA0, and
# astral code points (e.g. U+1F600) are accepted and resolve the token. The prior
# C0/DEL/tab behavior is a strict subset of this. (A surrogate code point cannot
# occur in valid UTF-8, so it also trips the UTF-8 decode on read and fails
# closed; it is included here for exact definitional parity.)
_FORBIDDEN_CONTROL = re.compile(
"[^\x09\x0a\x0d\x20-\x7e\x85\xa0-퟿-<2D>\U00010000-\U0010ffff]"
)
# Sentinel: "outside the supported subset -> fail closed". Distinct from a
# genuine null (None), which is a valid resolved value.
_FAIL = object()
# Separators are SPACE-only: PyYAML rejects a tab used as key/value whitespace
# (before or after the ':') with a scanner error, so a tab there must NOT be
# treated as a benign separator. Space before the colon and one space after it
# stay valid (PyYAML strips a plain key's trailing spaces); a tab in either
# position makes the whole line fail to match -> the caller fails closed.
_KEY_RE = re.compile(r"^([A-Za-z0-9_][A-Za-z0-9_.\-]*) *:(?:[ ](.*)|)$")
_FLOW = set("[]{}*&!")
class _Bail(Exception):
# Raised the instant the document leaves the narrow tea-config subset this
# recognizer can prove it resolves IDENTICALLY to PyYAML. Caught by
# _safe_parse, which then fails closed (returns _FAIL) rather than guess.
pass
def _strip_properties(value):
# Consume a leading YAML node non-specific tag ('!' + space) and return the
# remaining node text for normal resolution. PyYAML's SafeLoader applies
# this TRANSPARENT property and then resolves the following node with its
# ordinary implicit typing, so (verified against real PyYAML 6.0.3)
# '! x' -> 'x', '! ' -> null, exactly as the bare node would resolve.
# Returns _FAIL for every form we do NOT reproduce identically, so the
# caller fails closed: a tag shorthand ('!x' -> ConstructorError), an
# explicit or verbatim tag ('!!str' / '!!int' / '!<...>' / '!foo') whose
# type coercion we do not emulate, a tab separator (PyYAML scanner error),
# a duplicate tag ('! ! x' -> ParserError).
#
# ANCHORS ('&name'): deliberately rejected in FULL, not just duplicates.
# Real PyYAML tracks anchor NAMES in a document-scoped composer registry
# and raises ComposerError ("found duplicate anchor") the instant the
# SAME anchor name is declared on a SECOND node anywhere in the document
# -- including on a totally unrelated node far from the token field. This
# line-oriented fallback has no such document-wide registry (and building
# one reliably, across every node shape this recognizer does not even
# parse, risks missing a scope and re-opening the fail-open). So instead
# of emulating the registry, every '&'-anchor property fails closed here,
# unconditionally. This is a conservative OVER-reject relative to PyYAML:
# a single, non-duplicated '&a x' is valid YAML that real PyYAML resolves
# to 'x', but we refuse it too. That is intentional and safe -- fail
# closed can only cost an emitted token PyYAML would have allowed, never
# emit one PyYAML would reject.
seen_tag = False
while value[:1] in ("!", "&"):
if value[0] == "!":
# Non-specific tag: '!' then AT LEAST ONE SPACE. A tab is a PyYAML
# scanner error; '!x' / '!!type' / '!<verbatim>' are typed/short tags.
if seen_tag or len(value) < 2 or value[1] != " ":
return _FAIL
seen_tag = True
value = value[1:].lstrip(" ")
else: # anchor '&name' -- always fail closed, see comment above
return _FAIL
return value
def _scalar(raw):
# Resolve a single flow scalar (quoted or plain) the way PyYAML would for
# tea's simple values, or _FAIL when it is outside the supported subset so
# the caller fails closed instead of guessing.
#
# A quoted scalar is ALWAYS a string (its contents returned verbatim); a '#'
# inside quotes is data. An UNQUOTED scalar ends at the first whitespace
# -preceded '#' (a YAML comment must be preceded by whitespace or line start,
# so "abc#def" stays literal while "abc # note" becomes "abc"), and is then
# subject to PyYAML's implicit typing: forms like 12345 / null / ~ / yes /
# 3.14 / a timestamp resolve to a NON-string (int/None/bool/float/date), so
# they return None here (PyYAML's path rejects a non-str token via _accept).
# Strip SPACES only, never tabs: PyYAML raises a scanner error on a tab in a
# plain/leading/trailing scalar position, so a tab must be PRESERVED here to
# trip the fail-closed guard below rather than be silently normalized away.
value = raw.strip(" ")
if not value:
return None # empty plain scalar -> null
if value[0] in ("'", '"'):
quote = value[0]
end = value.find(quote, 1)
if end == -1:
return _FAIL # unterminated quote: PyYAML would error / continue
rest = value[end + 1:].strip(" ")
if rest and not rest.startswith("#"):
return _FAIL # trailing junk after a quoted scalar
inner = value[1:end]
# Single-quote '' escaping and double-quote backslash escapes are NOT
# interpreted here; reject any scalar that uses them so we never diverge
# from PyYAML on escape handling.
if quote == "'" and "'" in inner:
return _FAIL
if quote == '"' and "\\" in inner:
return _FAIL
return inner
for i, ch in enumerate(value):
if ch == "#" and (i == 0 or value[i - 1] in (" ", "\t")):
value = value[:i]
break
value = value.strip(" ") # spaces only; a tab must survive to fail closed
if not value:
return None
if value[0] in ("!", "&"):
# Leading node property (non-specific tag '! ' / plain anchor '&name '):
# PyYAML applies it to the FOLLOWING node and resolves THAT, so strip the
# transparent forms and re-resolve the remainder EXACTLY (implicit typing
# and all). _strip_properties returns _FAIL for the tag/anchor forms
# PyYAML rejects or type-coerces, which fail closed here.
remainder = _strip_properties(value)
if remainder is _FAIL:
return _FAIL
return _scalar(remainder)
if value[0] in _FLOW or value[0] in ("|", ">", "@", "`", '"', "'", "%", ","):
return _FAIL # flow / block-scalar / reserved / directive / quote
# NOTE: "?" is deliberately NOT in the blanket-reject set above. Empirically
# verified against real PyYAML 6.0.3: "?x" (indicator immediately followed by
# a non-space) is a plain scalar string "?x" like "-x" and ":x" are -- only
# "?" alone or "? " (followed by space/EOL) opens a complex mapping key and
# is illegal in a value position. Blanket-rejecting every leading "?" would
# over-reject a token PyYAML accepts verbatim; the narrower check below
# (mirroring "-" and ":") handles exactly the illegal bare-indicator forms.
if value[0] in ("-", "?", ":") and (len(value) == 1 or value[1] in (" ", "\t")):
# A bare block indicator, not a plain scalar: '-'/'- ' opens a sequence
# entry, '?'/'? ' a complex mapping key, ':'/': ' a mapping value -- all
# illegal in a value position, where PyYAML raises a scanner error on the
# whole document. "-x"/"-1"/"?x"/":x" (indicator NOT followed by space)
# remain valid plain scalars and fall through. Fail closed on the bare
# indicator so the fallback never emits a token PyYAML would refuse.
return _FAIL
# NO blanket internal-indicator reject here. In BLOCK context (where a tea
# config value always sits) PyYAML treats ',[]{}' as ordinary plain-scalar
# content -- they are flow indicators ONLY inside a flow collection -- and
# treats '!&*|>#%@`' plus quotes as significant ONLY at the FIRST non-space
# character of a node (a leading one starts a tag/anchor/alias/block-scalar/
# comment/directive/reserved/quote), which the leading-char guard above
# (value[0] ...) already rejects. Verified empirically against real PyYAML
# 6.0.3: an INTERNAL '!' ',' '[' ']' '{' '}' '&' '*' '#'(not space-preceded)
# ':'(not space-followed) all resolve as a plain-string token. The only
# internal positions PyYAML treats as significant in block context are ' #'
# (whitespace-preceded '#' -> comment; stripped by the loop above) and ': '
# or a trailing ':' (-> mapping; rejected by the ": "/endswith(":") guard
# below). A blanket any(ch in _FLOW ...) scan over-rejected those internal
# indicators, dropping a token PyYAML emits verbatim; removing it restores
# parity while the position-specific guards keep the fail-open direction shut.
if "\t" in value:
# A tab anywhere in a plain scalar (leading, trailing, or embedded) is a
# PyYAML scanner error on the whole document -- it accepts tabs ONLY
# inside quoted scalars (handled above, returned verbatim). Fail closed
# so the fallback never emits a token PyYAML would refuse over a tab.
return _FAIL
if ": " in value or value.endswith(":"):
return _FAIL # nested-mapping-in-scalar / ambiguous
if _implicit_nonstring(value):
# A plain scalar PyYAML would tag as a non-string (null/bool/int/float/
# timestamp). If PyYAML's CONSTRUCTOR would build it, the value is a
# non-string -> null-equivalent for a token field: return None and keep
# parsing (as before). But if it matches a typed implicit resolver yet is
# NOT constructible (e.g. 2023-99-99, 0b_, 0x_), PyYAML raises on the
# WHOLE document and yields no token, so the fallback MUST fail closed
# for the whole document too -> _FAIL (which the caller turns into _Bail).
if not _constructible(value):
return _FAIL
return None
return value
class _Parser:
# A deliberately NARROW, conservative recognizer for the block-style YAML
# subset tea writes (mappings of scalar fields; a `logins:` block SEQUENCE of
# such mappings; optional shallow nested mappings for e.g. preferences). It
# reconstructs the SAME Python object PyYAML's SafeLoader would, but the
# instant it meets anything it cannot prove it handles identically -- a
# document marker (--- / ...), a block scalar (| / >), a flow collection, a
# duplicate mapping key, inconsistent indentation, an escape, or any line
# outside the grammar -- it raises _Bail so the whole resolution fails
# closed. This guarantees the module invariant (only ever MORE conservative
# than PyYAML, never less) at the DOCUMENT level, closing the structural
# fail-open classes (nested-logins shadow, block-scalar shadow, duplicate
# root key, malformed-after-valid, and extra-document) that a line scan that
# does not validate whole-document structure would miss.
def __init__(self, lines):
self.toks = []
for raw in lines:
if not raw.strip():
continue
lead = raw[: len(raw) - len(raw.lstrip(" \t"))]
if "\t" in lead:
raise _Bail() # tab in indentation: PyYAML scanner error
indent = len(lead)
body = raw[indent:]
if body.lstrip().startswith("#"):
continue
if re.match(r"^(---|\.\.\.)(\s|$)", body) or body in ("---", "..."):
raise _Bail() # document / end marker -> multi-doc -> fail closed
# rstrip SPACES only: a trailing tab is a PyYAML scanner error, so it
# must be kept on the token body to reach the fail-closed guards
# (rstrip() would swallow it and let a bad line resolve a token).
self.toks.append((indent, body.rstrip(" ")))
self.i = 0
def peek(self):
return self.toks[self.i] if self.i < len(self.toks) else None
def parse_document(self):
if not self.toks:
return None
node = self.parse_node(0)
if self.i != len(self.toks):
raise _Bail() # trailing unconsumed content -> malformed
return node
def parse_node(self, min_indent):
tok = self.peek()
if tok is None:
return None
indent, body = tok
if indent < min_indent:
return None
if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")):
return self.parse_seq(indent)
return self.parse_map(indent)
def parse_map(self, indent):
result = {}
while True:
tok = self.peek()
if tok is None:
break
cur_indent, body = tok
if cur_indent < indent:
break
if cur_indent > indent:
raise _Bail() # unexpected deeper line (bad indentation)
if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")):
raise _Bail() # sequence item where a mapping entry was expected
m = _KEY_RE.match(body)
if not m:
raise _Bail()
key = m.group(1)
inline = m.group(2)
self.i += 1
if key in result:
raise _Bail() # duplicate mapping key (PyYAML last-wins; we bail)
if inline is not None and inline.strip(" ") != "":
val = _scalar(inline)
if val is _FAIL:
raise _Bail()
result[key] = val
else:
result[key] = self.parse_block_value(indent)
return result
def parse_block_value(self, key_indent):
# The value after a "key:" with no inline scalar. A block SEQUENCE may sit
# at the same indent as the key (YAML permits `- ` aligned with the key --
# tea's own on-disk shape) or deeper; a block MAPPING must be strictly
# deeper; otherwise the value is null.
nxt = self.peek()
if nxt is None:
return None
ni, nb = nxt
is_item = nb.startswith("-") and (len(nb) == 1 or nb[1] in (" ", "\t"))
if is_item and ni >= key_indent:
return self.parse_seq(ni)
if ni > key_indent:
return self.parse_node(ni)
return None
def parse_seq(self, indent):
result = []
while True:
tok = self.peek()
if tok is None:
break
cur_indent, body = tok
if cur_indent < indent:
break
if cur_indent > indent:
raise _Bail()
if not (body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t"))):
break # a mapping entry at this indent ends the sequence
rest = body[1:].strip(" ") # spaces only; a tab must survive to bail
self.i += 1
if rest == "":
nxt = self.peek()
if nxt is not None and nxt[0] > indent:
result.append(self.parse_node(indent + 1))
else:
result.append(None)
continue
km = _KEY_RE.match(rest)
if km:
# "- key: value" opens a mapping whose fields continue at the
# column where the content after the dash began.
field_indent = indent + (len(body) - len(body[1:].lstrip()))
result.append(self.parse_inline_map(field_indent, km))
else:
val = _scalar(rest)
if val is _FAIL:
raise _Bail()
result.append(val)
return result
def parse_inline_map(self, field_indent, first_match):
result = {}
key = first_match.group(1)
inline = first_match.group(2)
if inline is not None and inline.strip(" ") != "":
val = _scalar(inline)
if val is _FAIL:
raise _Bail()
result[key] = val
else:
result[key] = self.parse_block_value(field_indent)
while True:
tok = self.peek()
if tok is None:
break
cur_indent, body = tok
if cur_indent != field_indent:
if cur_indent > field_indent:
raise _Bail()
break
if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")):
raise _Bail()
m = _KEY_RE.match(body)
if not m:
raise _Bail()
k = m.group(1)
iv = m.group(2)
self.i += 1
if k in result:
raise _Bail()
if iv is not None and iv.strip(" ") != "":
v = _scalar(iv)
if v is _FAIL:
raise _Bail()
result[k] = v
else:
result[k] = self.parse_block_value(field_indent)
return result
def _safe_parse(lines):
# Return the parsed root object (dict/list/scalar/None) when the WHOLE
# document is inside the supported subset, else _FAIL (fail closed).
try:
return _Parser(lines).parse_document()
except _Bail:
return _FAIL
except Exception:
return _FAIL
def _url_matches_repo_host(url):
# Mirror gitea_url_matches_host (detect-platform.sh): the login's recorded
# URL must name the SAME host AND the SAME effective port as the repo remote
# host, not merely the same hostname. A login configured for an explicit,
# non-default provider port (e.g. :9443) must NOT satisfy a portless (default
# -port) repo host, and a login on the matching port (e.g. :8443) must NOT be
# rejected. Ports are normalized by applying the login URL's scheme default
# (80 for http, else 443) to whichever side omits the port, symmetrically, so
# an implicit port and its explicit default-port form compare equal.
if not isinstance(url, str) or not url:
return False
configured = urlparse(url if "//" in url else f"//{url}")
remote = urlparse(f"//{repo_host}")
configured_host = configured.hostname
if not configured_host or configured_host.lower() != (remote.hostname or "").lower():
return False
default_port = 80 if configured.scheme == "http" else 443
configured_port = configured.port if configured.port is not None else default_port
remote_port = remote.port if remote.port is not None else default_port
return configured_port == remote_port
def _accept(token, url):
# Enforce the host bind before surfacing a token. When a repo host is given,
# the login's recorded URL host AND port must match it; a login with no
# usable URL (or a mismatched host/port) is rejected (fail closed) so a
# cross-host (or cross-port) credential is never emitted.
if not isinstance(token, str) or not token:
return None
if repo_host:
if not _url_matches_repo_host(url):
return None
return token
def _token_via_pyyaml():
# Preferred, fully general path when PyYAML is installed. Raises ImportError
# (caught by the caller) when the module is unavailable so the environment
# -robust fallback can take over instead of failing closed on every host
# that lacks PyYAML.
import yaml
with open(config_path, encoding="utf-8") as handle:
config = yaml.safe_load(handle)
logins = config.get("logins") if isinstance(config, dict) else None
if not isinstance(logins, list):
return None
for login in logins:
if isinstance(login, dict) and str(login.get("name") or "") == wanted:
return _accept(login.get("token"), login.get("url"))
return None
_BREAKS = "\r\n\x85"
_VALUE_OPEN_RE = re.compile(r"^\s*(?:-\s+)*[A-Za-z0-9_][A-Za-z0-9_.\-]*:$")
_SEQ_OPEN_RE = re.compile(r"^\s*(?:-\s+)*-$")
def _value_open_before(prefix):
# True when a quote appearing at the END of `prefix` (the current logical
# line so far) begins a flow SCALAR value -- the only position where a quote
# opens a quoted scalar whose embedded line breaks PyYAML folds instead of
# splitting. That is: the prefix is empty/all-indent (root or block scalar
# start), or ends with a mapping ': ' or a sequence '- ' indicator that is
# SPACE-separated from the quote. A quote glued to the previous char (e.g.
# 'key:"x') is NOT a value opener (PyYAML needs the space), and a quote mid
# -content is literal -- both fail this test, so their breaks split the line
# exactly as before (fail closed / plain-scalar behavior preserved).
stripped = prefix.rstrip(" ")
if stripped == "":
return True # start of line (after any indentation) -> node value start
if prefix == stripped:
return False # no space before the quote -> not a value opener
return bool(_VALUE_OPEN_RE.match(stripped) or _SEQ_OPEN_RE.match(stripped))
def _fold_quoted_break_run(run):
# Reproduce PyYAML's flow-scalar folding (scan_flow_scalar_spaces +
# scan_flow_scalar_breaks) for the maximal ' \t' + line-break run inside a
# quoted scalar. A run with NO break is literal whitespace (verbatim). With a
# break: surrounding spaces/tabs are dropped; a single \n-class break
# (\n / \r / \r\n / \x85 NEL) folds to ONE space; (LS) / (PS)
# are kept verbatim; each ADDITIONAL break (a blank line) contributes its own
# break char (\n for the \n-class). Verified against real PyYAML 6.0.3 for
# both double- and single-quoted scalars (they fold identically here).
def _lb(s, j):
if s[j] == "\r" and j + 1 < len(s) and s[j + 1] == "\n":
return "\n", 2
if s[j] in "\r\n\x85":
return "\n", 1
return s[j], 1 # / preserved verbatim
n = len(run)
j = 0
while j < n and run[j] in " \t":
j += 1
if j >= n:
return run # pure whitespace, no break -> literal content
line_break, adv = _lb(run, j)
j += adv
breaks = []
while True:
while j < n and run[j] in " \t":
j += 1
if j < n and run[j] in _BREAKS:
b, adv = _lb(run, j)
j += adv
breaks.append(b)
else:
break
out = ""
if line_break != "\n":
out += line_break
elif not breaks:
out += " "
return out + "".join(breaks)
def _split_logical_lines(raw_text):
# Split like str.splitlines() EXCEPT a line break INSIDE a value-opening
# quoted scalar does NOT end the line: PyYAML keeps the quoted scalar together
# and folds the break (flow folding above), so we fold it and continue the
# same logical line. Only the break chars that survive the _FORBIDDEN_CONTROL
# gate reach here: \n, \r, \x85 (NEL), (LS), (PS). OUTSIDE
# quotes every one ends the line (matching PyYAML's block-level line breaks
# and str.splitlines), so a NEL/LS/PS used as the document's line-terminator
# style, and an UNQUOTED plain scalar carrying an embedded break, split and
# fail closed exactly as before. A quoted scalar left unterminated at EOF
# stays on the final line and fails closed in _scalar (PyYAML errors too).
out = []
cur = []
i = 0
n = len(raw_text)
quote = None
while i < n:
ch = raw_text[i]
if quote is None:
if ch in _BREAKS:
out.append("".join(cur))
cur = []
i += 2 if (ch == "\r" and i + 1 < n and raw_text[i + 1] == "\n") else 1
continue
if ch in ("'", '"') and _value_open_before("".join(cur)):
quote = ch
cur.append(ch)
i += 1
continue
cur.append(ch)
i += 1
continue
# inside a value-opening quoted scalar
if ch == quote:
cur.append(ch)
quote = None
i += 1
continue
if ch in " \t" or ch in _BREAKS:
j = i
has_break = False
while j < n and (raw_text[j] in " \t" or raw_text[j] in _BREAKS):
if raw_text[j] in _BREAKS:
has_break = True
j += 1
run = raw_text[i:j]
if has_break:
# A continuation line beginning (at column 0, no indent) with a
# document marker is a PyYAML scanner error even inside a quoted
# scalar; when the run ends on a break the continuation content at
# j is at column 0, so guard it and fail closed. (An INDENTED
# '---' is literal content to PyYAML and folds normally.)
if run[-1] in _BREAKS and raw_text[j:j + 3] in ("---", "...") and (
j + 3 >= n or raw_text[j + 3] in "\0 \t" + _BREAKS
):
raise _Bail()
cur.append(_fold_quoted_break_run(run))
else:
cur.append(run)
i = j
continue
cur.append(ch)
i += 1
if cur or not out:
out.append("".join(cur))
return out
def _token_via_lines():
# Conservative fallback for hosts without PyYAML. It parses config.yml with a
# strict recognizer (_safe_parse) of the narrow block-style subset tea writes,
# which reconstructs the SAME object PyYAML would OR fails closed (_FAIL) on
# ANYTHING it cannot prove it resolves identically -- document markers, block
# scalars, flow collections, duplicate keys, inconsistent indentation, or any
# line outside the grammar. On a recognized document it then resolves the
# login EXACTLY as the PyYAML path does (root `logins` list -> first entry
# whose `name` equals the request -> host/port-bound token), so the fallback
# can only ever be MORE conservative than PyYAML, never less.
with open(config_path, encoding="utf-8") as handle:
raw_text = handle.read()
# Whole-document, position-independent reject: PyYAML's Reader rejects the
# ENTIRE document (ReaderError) if a NON-printable code point (per its exact
# printable definition -- see _FORBIDDEN_CONTROL: C0 controls, DEL, the C1
# block, surrogates, and 0xFFFE/0xFFFF) appears ANYWHERE in the raw stream --
# in a plain scalar, inside single- or double-quoted scalars, in a comment,
# or in a field this recognizer never even looks at. Checking the raw text
# (before line splitting, which would otherwise split on some of these same
# code points -- e.g. \x0b, \x0c, \x1c-\x1e -- and obscure them) mirrors that
# exactly: fail closed for the whole document. (NEL 0x85, LS 0x2028 and PS
# 0x2029 are PRINTABLE to PyYAML's Reader, so they are NOT in
# _FORBIDDEN_CONTROL; _split_logical_lines below reproduces PyYAML's line-break
# semantics for them -- a structural break outside quotes, a FOLDED break
# inside a quoted scalar -- rather than str.splitlines(), which would wrongly
# split them mid-quote and fail-close a token PyYAML resolves.)
if _FORBIDDEN_CONTROL.search(raw_text):
return None
lines = _split_logical_lines(raw_text)
config = _safe_parse(lines)
if config is _FAIL:
return None
logins = config.get("logins") if isinstance(config, dict) else None
if not isinstance(logins, list):
return None
for login in logins:
if isinstance(login, dict) and str(login.get("name") or "") == wanted:
return _accept(login.get("token"), login.get("url"))
return None
try:
try:
token = _token_via_pyyaml()
except ImportError:
token = _token_via_lines()
except Exception:
raise SystemExit(1)
if isinstance(token, str) and token:
print(token)
raise SystemExit(0)
raise SystemExit(1)
PY
}
# Resolve HTTPS basic auth credentials for a Gitea host from ~/.git-credentials. # Resolve HTTPS basic auth credentials for a Gitea host from ~/.git-credentials.
# Prints "username:password" for direct curl -u consumption. Callers must not log it. # Prints "username:password" for direct curl -u consumption. Callers must not log it.
get_gitea_basic_auth() { get_gitea_basic_auth() {

View File

@@ -0,0 +1,69 @@
#!/bin/bash
# git-credential-mosaic — git credential helper — resolves Gitea tokens from
# the Mosaic credential store at runtime so remote URLs never embed secrets.
#
# Install (one-time, per clone or globally):
# git config credential.helper "$HOME/.config/mosaic/tools/git/git-credential-mosaic"
# # or, fleet-wide: git config --global credential.helper "$HOME/.config/mosaic/tools/git/git-credential-mosaic"
#
# Per-agent Gate-16 identity (author != reviewer separation):
# git config mosaic.gitIdentity <agent-id> # per-worktree, persists on disk
# # or: export MOSAIC_GIT_IDENTITY=<agent-id>
#
# Resolution priority: MOSAIC_GIT_IDENTITY env > git config mosaic.gitIdentity
# (per-worktree, survives across non-persistent shells) > git-supplied username
# (credential.username / URL). When the resolved identity has a matching
# per-agent token file, use it instead of the shared account. Backward
# compatible: nothing resolvable -> shared token (unchanged behavior).
[ "$1" = "get" ] || exit 0
host=""; username_in=""
while IFS= read -r line; do
[ -z "$line" ] && break
case "$line" in
host=*) host=${line#host=};;
username=*) username_in=${line#username=};;
esac
done
# Per-agent identity resolution (Gate-16 author≠reviewer separation).
# Priority: MOSAIC_GIT_IDENTITY env > git config mosaic.gitIdentity (per-worktree,
# survives across non-persistent shells) > git-supplied username (credential.username
# / URL). When the resolved identity has a matching per-agent token, use it instead of
# the shared account. Backward-compatible: nothing resolvable → shared token.
ident="$MOSAIC_GIT_IDENTITY"
[ -z "$ident" ] && ident=$(git config --get mosaic.gitIdentity 2>/dev/null)
[ -z "$ident" ] && ident="$username_in"
if [ -n "$ident" ]; then
case "$host" in
git.uscllc.com) idpfx=gitea-usc;;
git.mosaicstack.dev) idpfx=gitea-mosaicstack;;
*) idpfx="";;
esac
if [ -n "$idpfx" ]; then
idtok="$HOME/.config/mosaic/secrets/gitea-tokens/${idpfx}-${ident}.token"
if [ -r "$idtok" ]; then
echo "username=${ident}"
echo "password=$(cat "$idtok")"
exit 0
fi
fi
fi
case "$host" in
git.uscllc.com) svc=gitea-usc;;
git.mosaicstack.dev) svc=gitea-mosaicstack;;
*) exit 0;;
esac
# Script-relative (not $HOME-absolute) so this resolves correctly regardless
# of where the framework installer places tools/ under $HOME — mirrors
# detect-platform.sh's own cred_loader resolution in this same directory.
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=../_lib/credentials.sh
source "$script_dir/../_lib/credentials.sh"
load_credentials "$svc" >/dev/null 2>&1 || exit 0
# GITEA_USER is not populated by load_credentials (it only exports
# GITEA_URL/GITEA_TOKEN for gitea-*), so this fallback is normally taken. Gitea's
# git-over-HTTP auth authenticates from the token itself (the password field),
# not from the username string, so any non-empty placeholder works here — this
# is deliberately NOT a real account name (framework files must stay
# operator-agnostic; see tools/quality/scripts/verify-sanitized.sh).
echo "username=${GITEA_USER:-git}"
echo "password=$GITEA_TOKEN"

View File

@@ -1,6 +1,26 @@
#!/bin/bash #!/bin/bash
# issue-comment.sh - Add a comment to an issue on GitHub or Gitea # issue-comment.sh - Add a comment to an issue on GitHub or Gitea
# Usage: issue-comment.sh -i <issue_number> -c <comment> # Usage: issue-comment.sh -i <issue_number> -c <comment> [--login <name>]
#
# tea v0.11.1 defines no `comment` subcommand under `tea issue` (or `tea pr`);
# the non-existent `tea issue comment ...` form does not error — tea silently
# no-ops and still exits 0, so a caller trusting the exit code believes a
# comment was posted when it was not (#865). tea 0.11.1 also cannot reliably
# emit the id of a record it created, so an exit code is the ONLY signal it
# offers — and that signal is untrustworthy. This script therefore does not
# write via tea at all: it POSTs the comment through the Gitea REST API (which
# returns the created comment object, including its id), then GETs that exact
# id back and fails closed unless it matches. Keying verification to the
# provider-returned created id means a concurrent comment cannot masquerade as
# this write and a no-op create simply yields no id to verify.
#
# --login override: the default login is resolved from the local `tea` login
# list for this repo's host (get_gitea_login). Pass --login <name> to override
# it for this invocation only. The REST write, the /user identity read, and the
# read-back are ALL performed with the token of the EFFECTIVE login (the
# override when given), so the write and its verification bind to the same
# identity — a --login override is never written under one credential and
# verified under a different default one.
set -e set -e
@@ -10,6 +30,7 @@ source "$SCRIPT_DIR/detect-platform.sh"
# Parse arguments # Parse arguments
ISSUE_NUMBER="" ISSUE_NUMBER=""
COMMENT="" COMMENT=""
LOGIN_OVERRIDE=""
while [[ $# -gt 0 ]]; do while [[ $# -gt 0 ]]; do
case $1 in case $1 in
@@ -21,12 +42,17 @@ while [[ $# -gt 0 ]]; do
COMMENT="$2" COMMENT="$2"
shift 2 shift 2
;; ;;
-l|--login)
LOGIN_OVERRIDE="$2"
shift 2
;;
-h|--help) -h|--help)
echo "Usage: issue-comment.sh -i <issue_number> -c <comment>" echo "Usage: issue-comment.sh -i <issue_number> -c <comment> [--login <name>]"
echo "" echo ""
echo "Options:" echo "Options:"
echo " -i, --issue Issue number (required)" echo " -i, --issue Issue number (required)"
echo " -c, --comment Comment text (required)" echo " -c, --comment Comment text (required)"
echo " -l, --login Override the detected Gitea tea login for this call"
echo " -h, --help Show this help" echo " -h, --help Show this help"
exit 0 exit 0
;; ;;
@@ -49,20 +75,273 @@ fi
detect_platform >/dev/null detect_platform >/dev/null
# Resolve and cache the Gitea REST endpoint + token for the current remote,
# bound to a SPECIFIC login identity ($1). Populates GITEA_API_ROOT (…/api/v1),
# GITEA_API_BASE (…/api/v1/repos/<slug>), and GITEA_API_TOKEN.
#
# The token is resolved for the EFFECTIVE login (the --login override when
# given, otherwise the detected default) so that the single credential used for
# the write ALSO drives the /user identity read and the read-back — write token
# and read-back token are the same identity by construction (this is the
# credential-ordering fix: a --login override is no longer written under one
# credential and verified under a different default one). Falls back to the
# host-scoped credential ONLY when NO --login override was supplied (the
# best-effort default path). When $2 is "explicit" the login came from a
# caller-supplied --login: that exact login's token MUST resolve, and we FAIL
# CLOSED rather than silently downgrading the write to the host default
# identity — otherwise a caller relying on a dedicated per-role credential would
# be told the write succeeded as requested while it was attributed to the shared
# default. Returns non-zero (clear stderr) on any resolution failure.
gitea_resolve_api_for_login() {
local effective_login="$1" override_explicit="${2:-}" host configured_url repo
host=$(get_remote_host)
if [[ -n "$override_explicit" ]]; then
GITEA_API_TOKEN=$(get_gitea_token_for_login "$effective_login" "$host") || {
echo "Error: could not resolve a host-matched Gitea token for --login '$effective_login' on host '$host'; refusing to fall back to the host default identity or a cross-host credential (comment write/read-back)" >&2
return 1
}
else
GITEA_API_TOKEN=$(get_gitea_token_for_login "$effective_login" "$host") \
|| GITEA_API_TOKEN=$(get_gitea_token "$host") || {
echo "Error: Gitea token not found for login '$effective_login' (comment write/read-back)" >&2
return 1
}
fi
configured_url=$(get_gitea_url_for_host "$host") || {
echo "Error: Configured Gitea URL not found for comment read-back verification" >&2
return 1
}
repo=$(get_gitea_repo_slug_for_url "$configured_url") || {
echo "Error: Could not resolve Gitea owner/repository relative to configured URL" >&2
return 1
}
GITEA_API_ROOT="${configured_url%/}/api/v1"
GITEA_API_BASE="$GITEA_API_ROOT/repos/$repo"
# The provider WEB base (scheme + host + effective port + any deployment path
# prefix) that Gitea uses to build a comment's html issue_url/pull_request_url.
# Read-back verification pins the returned URL's origin + path prefix to THIS,
# not just a repo/issue suffix.
GITEA_WEB_BASE="${configured_url%/}"
return 0
}
# Resolve the login of the identity the API token authenticates as (GET
# /user). Used to attribute a read-back record to THIS invocation's writer so
# a concurrent write from a DIFFERENT identity cannot satisfy verification.
# Prints the login on success.
gitea_authenticated_login() {
local response_file auth_config status
response_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-issue-comment-whoami.XXXXXX")
auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$response_file"
echo "Error: could not stage Gitea credential for identity read" >&2
return 1
}
trap 'rm -f "$response_file" "$auth_config"' RETURN
if ! status=$(curl -sS -o "$response_file" -w '%{http_code}' \
--config "$auth_config" \
"$GITEA_API_ROOT/user"); then
echo "Error: Gitea authenticated-identity read transport failed" >&2
return 1
fi
if [[ "$status" != "200" ]]; then
echo "Error: Gitea authenticated-identity read failed with HTTP $status" >&2
return 1
fi
python3 - "$response_file" <<'PY'
import json
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
user = json.load(response)
login = user.get("login") if isinstance(user, dict) else None
if not isinstance(login, str) or not login:
raise ValueError("missing authenticated login")
except (OSError, json.JSONDecodeError, TypeError, ValueError) as error:
print(f"Error: could not resolve authenticated Gitea identity: {error}", file=sys.stderr)
raise SystemExit(1)
print(login)
PY
}
# Post a comment to a Gitea issue via the supported REST API and verify it
# durably against a PROVIDER-RETURNED created id — never trust an exit code
# (#865 defect class: tea's non-existent `tea issue comment` no-ops yet exits
# 0). The write is a direct POST that returns the created comment object, so we
# learn the exact id of THIS write; we then GET that exact id and require
# id == created id AND author == acting identity AND exact body AND that it
# belongs to this issue. Because verification is keyed to the id the create
# returned, a concurrent comment (even same identity, same body) CANNOT
# masquerade as this write, and a suppressed/no-op write yields no created id
# and fails closed — there is no fallback list scan that a concurrent record
# could satisfy. Prints the created comment id on success.
#
# Args: $1 = issue number, $2 = comment body, $3 = acting identity login.
gitea_create_comment_verified() {
local issue_number="$1" comment_body="$2" acting_login="$3"
local payload write_file readback_file auth_config write_status readback_status created_id
payload=$(COMMENT_BODY="$comment_body" python3 -c '
import json
import os
print(json.dumps({"body": os.environ["COMMENT_BODY"]}))
')
write_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-issue-comment-write.XXXXXX")
readback_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-issue-comment-getid.XXXXXX")
auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$write_file" "$readback_file"
echo "Error: could not stage Gitea credential for comment write" >&2
return 1
}
trap 'rm -f "$write_file" "$readback_file" "$auth_config"' RETURN
if ! write_status=$(curl -sS -o "$write_file" -w '%{http_code}' \
-X POST \
--config "$auth_config" \
-H 'Content-Type: application/json' \
-d "$payload" \
"$GITEA_API_BASE/issues/$issue_number/comments"); then
echo "Error: Gitea comment write transport failed" >&2
return 1
fi
if [[ "$write_status" != "201" ]]; then
echo "Error: Gitea comment write failed with HTTP $write_status (#865: no durable comment created)" >&2
return 1
fi
created_id=$(python3 - "$write_file" <<'PY'
import json
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
comment = json.load(response)
created_id = comment.get("id") if isinstance(comment, dict) else None
if not isinstance(created_id, int) or created_id <= 0:
raise ValueError("create response carried no positive comment id")
except (OSError, json.JSONDecodeError, ValueError) as error:
print(f"Error: could not identify created Gitea comment: {error}", file=sys.stderr)
raise SystemExit(1)
print(created_id)
PY
) || return 1
if ! readback_status=$(curl -sS -o "$readback_file" -w '%{http_code}' \
--config "$auth_config" \
"$GITEA_API_BASE/issues/comments/$created_id"); then
echo "Error: Gitea comment read-back transport failed" >&2
return 1
fi
if [[ "$readback_status" != "200" ]]; then
echo "Error: Gitea comment read-back failed with HTTP $readback_status" >&2
return 1
fi
EXPECTED_COMMENT_ID="$created_id" EXPECTED_COMMENT_BODY="$comment_body" \
ACTING_LOGIN="$acting_login" EXPECTED_REPO_SLUG="${GITEA_API_BASE##*/repos/}" \
EXPECTED_NUMBER="$issue_number" EXPECTED_WEB_BASE="$GITEA_WEB_BASE" \
python3 - "$readback_file" <<'PY' || return 1
import json
import os
import sys
from urllib.parse import urlparse
def _origin_and_path(url):
# Normalize a URL to (scheme, host, effective-port) + comment path. The port
# defaults to the scheme's default (80 http / 443 otherwise) so an implicit
# port and its explicit default form compare equal.
parsed = urlparse(url or "")
scheme = (parsed.scheme or "").lower()
host = (parsed.hostname or "").lower()
default_port = 80 if scheme == "http" else 443
port = parsed.port if parsed.port is not None else default_port
return (scheme, host, port), parsed.path.rstrip("/")
try:
with open(sys.argv[1], encoding="utf-8") as response:
comment = json.load(response)
if not isinstance(comment, dict):
raise ValueError("response is not a comment object")
expected_id = int(os.environ["EXPECTED_COMMENT_ID"])
expected_body = os.environ["EXPECTED_COMMENT_BODY"]
acting_login = os.environ["ACTING_LOGIN"]
slug = os.environ["EXPECTED_REPO_SLUG"]
number = os.environ["EXPECTED_NUMBER"]
web_base = os.environ["EXPECTED_WEB_BASE"]
# Gitea populates WEB (html) URLs here, not API paths. A plain issue comment
# carries issue_url = <web_base>/<owner>/<repo>/issues/<n> (pull_request_url
# empty); a comment posted to a PR's conversation carries
# pull_request_url = <web_base>/<owner>/<repo>/pulls/<n> (issue_url empty).
# Pin the returned URL's ORIGIN (scheme+host+port) and its FULL path to this
# provider + repo + kind + number — an endswith/suffix test would accept a
# look-alike host (evil.example/deceptive/<slug>/issues/N) or a same-host
# decoy prefix (/other/<slug>/issues/N), so compare the whole thing.
base_origin, base_path = _origin_and_path(web_base)
expected_issue_path = f"{base_path}/{slug}/issues/{number}"
expected_pr_path = f"{base_path}/{slug}/pulls/{number}"
def _belongs(url, expected_path):
if not url:
return False
origin, path = _origin_and_path(url)
return origin == base_origin and path == expected_path
if comment.get("id") != expected_id:
raise ValueError("read-back id does not match the created id")
if (comment.get("user") or {}).get("login") != acting_login:
raise ValueError("created comment is not authored by the acting identity")
if comment.get("body") != expected_body:
raise ValueError("created comment body does not match")
if not (
_belongs(comment.get("issue_url"), expected_issue_path)
or _belongs(comment.get("pull_request_url"), expected_pr_path)
):
raise ValueError("created comment does not belong to this issue on this provider/repo")
except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
print(f"Error: Gitea comment persistence verification failed: {error}", file=sys.stderr)
raise SystemExit(1)
PY
echo "$created_id"
return 0
}
if [[ "$PLATFORM" == "github" ]]; then if [[ "$PLATFORM" == "github" ]]; then
gh issue comment "$ISSUE_NUMBER" --body "$COMMENT" gh issue comment "$ISSUE_NUMBER" --body "$COMMENT"
echo "Added comment to GitHub issue #$ISSUE_NUMBER" echo "Added comment to GitHub issue #$ISSUE_NUMBER"
elif [[ "$PLATFORM" == "gitea" ]]; then elif [[ "$PLATFORM" == "gitea" ]]; then
# Build the invocation as an argv array (not unquoted $(get_gitea_repo_args) # Resolve the login this comment should be attributed to: the --login
# word-splitting) so the comment body — including Markdown backticks, $(...), # override when given, otherwise the detected default for this repo's host.
# and quotes — is passed verbatim and never re-split or shell-evaluated. # A --login override always wins. Otherwise name this repo host's login only
REPO_SLUG=$(get_repo_slug) # as a best effort: the login name merely selects a per-login token, and
GITEA_LOGIN_NAME=$(get_gitea_login) || { # gitea_resolve_api_for_login falls back to the host credential
echo "Error: could not resolve a Gitea login for this repo; cannot comment on issue #$ISSUE_NUMBER." >&2 # (get_gitea_token) when no tea login is named, so the default credential
# still resolves even when the host tea has no matching login entry.
EFFECTIVE_LOGIN="$LOGIN_OVERRIDE"
[[ -n "$EFFECTIVE_LOGIN" ]] || EFFECTIVE_LOGIN=$(get_gitea_login 2>/dev/null || true)
# Bind the REST endpoint + token to the effective login, then derive the
# acting identity from that SAME credential (GET /user). The write below and
# its read-back both use this credential, so the write is verified against
# the identity that actually performed it. Passing "explicit" when --login
# was supplied forbids the host-default fallback: an unresolvable explicit
# override fails closed instead of writing under the default identity.
gitea_resolve_api_for_login "$EFFECTIVE_LOGIN" "${LOGIN_OVERRIDE:+explicit}" || exit 1
ACTING_LOGIN=$(gitea_authenticated_login) || exit 1
comment_id=$(gitea_create_comment_verified "$ISSUE_NUMBER" "$COMMENT" "$ACTING_LOGIN") || {
echo "Error: could not create and verify a comment on Gitea issue #$ISSUE_NUMBER via a provider-returned created id (#865)." >&2
exit 1 exit 1
} }
tea issue comment "$ISSUE_NUMBER" "$COMMENT" --repo "$REPO_SLUG" --login "$GITEA_LOGIN_NAME" echo "Added and verified comment on Gitea issue #$ISSUE_NUMBER (comment ID $comment_id)"
echo "Added comment to Gitea issue #$ISSUE_NUMBER"
else else
echo "Error: Unknown platform" echo "Error: Unknown platform"
exit 1 exit 1

View File

@@ -1,6 +1,21 @@
#!/bin/bash #!/bin/bash
# pr-review.sh - Review a pull request on GitHub or Gitea # pr-review.sh - Review a pull request on GitHub or Gitea
# Usage: pr-review.sh -n <pr_number> -a <action> [-c <comment>] # Usage: pr-review.sh -n <pr_number> -a <action> [-c <comment>] [--login <name>]
#
# Gitea reviews and comments are written through the supported REST API, not
# `tea`: tea 0.11.1 cannot emit the id of a record it creates and can silently
# no-op while exiting 0 (#865 defect class), so an exit code is the only — and
# untrustworthy — signal it offers. approve/request-changes POST to
# /pulls/{n}/reviews (returns the created review with its id); the `comment`
# action POSTs to /issues/{n}/comments (returns the created comment with its
# id). Each write is then verified by GETting that exact returned id, so a
# concurrent record cannot masquerade as this write and a no-op fails closed.
#
# --login override: the default login is resolved from the local tea login list
# for this repo's host (get_gitea_login_for_host). Pass --login <name> to
# override it for this invocation only. The REST write, the /user identity read,
# and every read-back are ALL performed with the token of the EFFECTIVE login,
# so the write and its verification bind to the same identity.
set -e set -e
@@ -12,6 +27,7 @@ source "$SCRIPT_DIR/detect-platform.sh"
PR_NUMBER="" PR_NUMBER=""
ACTION="" ACTION=""
COMMENT="" COMMENT=""
LOGIN_OVERRIDE=""
while [[ $# -gt 0 ]]; do while [[ $# -gt 0 ]]; do
case $1 in case $1 in
@@ -27,13 +43,18 @@ while [[ $# -gt 0 ]]; do
COMMENT="$2" COMMENT="$2"
shift 2 shift 2
;; ;;
-l|--login)
LOGIN_OVERRIDE="$2"
shift 2
;;
-h|--help) -h|--help)
echo "Usage: pr-review.sh -n <pr_number> -a <action> [-c <comment>]" echo "Usage: pr-review.sh -n <pr_number> -a <action> [-c <comment>] [--login <name>]"
echo "" echo ""
echo "Options:" echo "Options:"
echo " -n, --number PR number (required)" echo " -n, --number PR number (required)"
echo " -a, --action Review action: approve, request-changes, comment (required)" echo " -a, --action Review action: approve, request-changes, comment (required)"
echo " -c, --comment Review comment (required for request-changes)" echo " -c, --comment Review comment (required for request-changes)"
echo " -l, --login Override the detected Gitea tea login (approve/request-changes only)"
echo " -h, --help Show this help" echo " -h, --help Show this help"
exit 0 exit 0
;; ;;
@@ -56,51 +77,42 @@ fi
detect_platform >/dev/null detect_platform >/dev/null
# Post a review comment body to a Gitea PR via the supported comments REST API # Post a comment to a Gitea PR (PR comments ARE issue comments) via the
# and verify it durably via provider read-back (see docs on durable review # supported REST API and verify it against a PROVIDER-RETURNED created id. The
# provenance in README.md). Used by the `comment` action and, since `tea` # write is a direct POST that returns the created comment object, so we learn
# v0.11.1 defines no `--comment`/`-comment` flag on `pr approve`/`pr reject`, # the exact id of THIS write; we GET that exact id and require id == created id
# also by the `approve` and `request-changes` actions to carry an optional # AND author == acting identity AND exact body AND that it belongs to this PR.
# review body that `tea` itself cannot attach. # Keying to the returned id means no concurrent comment (even same identity /
# body) can masquerade as this write, and a no-op create yields no id and fails
# closed. Requires GITEA_API_BASE / GITEA_API_TOKEN to be resolved first (via
# gitea_resolve_api_for_login). Prints the created comment id on success.
# #
# Args: $1 = PR number, $2 = comment body # Args: $1 = PR number, $2 = comment body, $3 = acting identity login.
# On success: prints only the created comment ID to stdout, returns 0. gitea_create_comment_verified() {
# On failure: prints an error to stderr, returns 1. local pr_number="$1" comment_body="$2" acting_login="$3"
gitea_post_verified_comment() { local payload write_file readback_file auth_config write_status readback_status created_id
local pr_number="$1" comment_body="$2"
local host token configured_url repo api_base payload
local write_response_file readback_response_file comment_id
host=$(get_remote_host)
token=$(get_gitea_token "$host") || {
echo "Error: Gitea token not found for comment persistence" >&2
return 1
}
configured_url=$(get_gitea_url_for_host "$host") || {
echo "Error: Configured Gitea URL not found for comment persistence" >&2
return 1
}
repo=$(get_gitea_repo_slug_for_url "$configured_url") || {
echo "Error: Could not resolve Gitea owner/repository relative to configured URL" >&2
return 1
}
api_base="${configured_url%/}/api/v1/repos/$repo"
payload=$(COMMENT_BODY="$comment_body" python3 -c ' payload=$(COMMENT_BODY="$comment_body" python3 -c '
import json import json
import os import os
print(json.dumps({"body": os.environ["COMMENT_BODY"]})) print(json.dumps({"body": os.environ["COMMENT_BODY"]}))
') ')
write_response_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-write.XXXXXX") write_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-write.XXXXXX")
readback_response_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-readback.XXXXXX") readback_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-getid.XXXXXX")
trap 'rm -f "$write_response_file" "$readback_response_file"' RETURN auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$write_file" "$readback_file"
echo "Error: could not stage Gitea credential for comment write" >&2
return 1
}
trap 'rm -f "$write_file" "$readback_file" "$auth_config"' RETURN
if ! write_status=$(curl -sS -o "$write_response_file" -w '%{http_code}' \ if ! write_status=$(curl -sS -o "$write_file" -w '%{http_code}' \
-X POST \ -X POST \
-H "Authorization: token $token" \ --config "$auth_config" \
-H 'Content-Type: application/json' \ -H 'Content-Type: application/json' \
-d "$payload" \ -d "$payload" \
"$api_base/issues/$pr_number/comments"); then "$GITEA_API_BASE/issues/$pr_number/comments"); then
echo "Error: Gitea comment write transport failed" >&2 echo "Error: Gitea comment write transport failed" >&2
return 1 return 1
fi fi
@@ -109,26 +121,26 @@ print(json.dumps({"body": os.environ["COMMENT_BODY"]}))
return 1 return 1
fi fi
comment_id=$(python3 - "$write_response_file" <<'PY' created_id=$(python3 - "$write_file" <<'PY'
import json import json
import sys import sys
try: try:
with open(sys.argv[1], encoding="utf-8") as response: with open(sys.argv[1], encoding="utf-8") as response:
comment = json.load(response) comment = json.load(response)
comment_id = comment.get("id") if isinstance(comment, dict) else None created_id = comment.get("id") if isinstance(comment, dict) else None
if not isinstance(comment_id, int) or comment_id <= 0: if not isinstance(created_id, int) or created_id <= 0:
raise ValueError("missing positive comment id") raise ValueError("create response carried no positive comment id")
except (OSError, json.JSONDecodeError, ValueError) as error: except (OSError, json.JSONDecodeError, ValueError) as error:
print(f"Error: could not identify created Gitea comment: {error}", file=sys.stderr) print(f"Error: could not identify created Gitea comment: {error}", file=sys.stderr)
raise SystemExit(1) raise SystemExit(1)
print(comment_id) print(created_id)
PY PY
) || return 1 ) || return 1
if ! readback_status=$(curl -sS -o "$readback_response_file" -w '%{http_code}' \ if ! readback_status=$(curl -sS -o "$readback_file" -w '%{http_code}' \
-H "Authorization: token $token" \ --config "$auth_config" \
"$api_base/issues/comments/$comment_id"); then "$GITEA_API_BASE/issues/comments/$created_id"); then
echo "Error: Gitea comment read-back transport failed" >&2 echo "Error: Gitea comment read-back transport failed" >&2
return 1 return 1
fi fi
@@ -137,13 +149,28 @@ PY
return 1 return 1
fi fi
if EXPECTED_COMMENT_ID="$comment_id" EXPECTED_COMMENT_BODY="$comment_body" EXPECTED_REPO="$repo" EXPECTED_PR_NUMBER="$pr_number" \ EXPECTED_COMMENT_ID="$created_id" EXPECTED_COMMENT_BODY="$comment_body" \
python3 - "$readback_response_file" <<'PY' ACTING_LOGIN="$acting_login" EXPECTED_REPO_SLUG="${GITEA_API_BASE##*/repos/}" \
EXPECTED_NUMBER="$pr_number" EXPECTED_WEB_BASE="$GITEA_WEB_BASE" \
python3 - "$readback_file" <<'PY' || return 1
import json import json
import os import os
import sys import sys
from urllib.parse import urlparse from urllib.parse import urlparse
def _origin_and_path(url):
# Normalize a URL to (scheme, host, effective-port) + comment path. The port
# defaults to the scheme's default (80 http / 443 otherwise) so an implicit
# port and its explicit default form compare equal.
parsed = urlparse(url or "")
scheme = (parsed.scheme or "").lower()
host = (parsed.hostname or "").lower()
default_port = 80 if scheme == "http" else 443
port = parsed.port if parsed.port is not None else default_port
return (scheme, host, port), parsed.path.rstrip("/")
try: try:
with open(sys.argv[1], encoding="utf-8") as response: with open(sys.argv[1], encoding="utf-8") as response:
comment = json.load(response) comment = json.load(response)
@@ -151,27 +178,352 @@ try:
raise ValueError("response is not a comment object") raise ValueError("response is not a comment object")
expected_id = int(os.environ["EXPECTED_COMMENT_ID"]) expected_id = int(os.environ["EXPECTED_COMMENT_ID"])
expected_body = os.environ["EXPECTED_COMMENT_BODY"] expected_body = os.environ["EXPECTED_COMMENT_BODY"]
expected_repo = os.environ["EXPECTED_REPO"] acting_login = os.environ["ACTING_LOGIN"]
expected_pr = os.environ["EXPECTED_PR_NUMBER"] slug = os.environ["EXPECTED_REPO_SLUG"]
issue_path = urlparse(comment.get("issue_url", "")).path.rstrip("/") number = os.environ["EXPECTED_NUMBER"]
expected_suffix = f"/repos/{expected_repo}/issues/{expected_pr}" web_base = os.environ["EXPECTED_WEB_BASE"]
# Gitea populates WEB (html) URLs here, not API paths. A PR-conversation
# comment carries pull_request_url = <web_base>/<owner>/<repo>/pulls/<n> (with
# issue_url empty), while a plain issue comment carries
# issue_url = <web_base>/<owner>/<repo>/issues/<n> (with pull_request_url empty).
# This is the pr-review `comment` action, so the comment MUST land on a pull
# request: require pull_request_url. A plain issue_url is REJECTED — if issue
# #N exists but PR #N does not, POST /issues/N/comments creates an issue
# comment, and accepting that issue_url would let the wrapper falsely report a
# verified PR comment (issue-comment.sh legitimately keeps the broader
# issue-or-PR acceptance; a PR review does not).
# Pin the returned URL's ORIGIN (scheme+host+port) and its FULL path to this
# provider + repo + kind + number — an endswith/suffix test would accept a
# look-alike host (evil.example/deceptive/<slug>/pulls/N) or a same-host
# decoy prefix (/other/<slug>/pulls/N), so compare the whole thing.
base_origin, base_path = _origin_and_path(web_base)
expected_pr_path = f"{base_path}/{slug}/pulls/{number}"
def _belongs(url, expected_path):
if not url:
return False
origin, path = _origin_and_path(url)
# Repo owner/repo slugs are case-insensitive (Gitea canonicalizes the
# pull_request_url slug to lowercase on return), while EXPECTED_REPO_SLUG
# is taken verbatim from GITEA_API_BASE and may be mixed-case. The
# remainder of the path (".../pulls/<number>") is numeric, so lowercasing
# the whole path for this comparison only relaxes case, not identity: the
# origin tuple (scheme+host+port) above still pins the provider host, and
# the path is still compared in FULL (no endswith/suffix match), so the
# look-alike-host and same-host decoy-prefix protections are unchanged.
return origin == base_origin and path.lower() == expected_path.lower()
if comment.get("id") != expected_id: if comment.get("id") != expected_id:
raise ValueError("comment id mismatch") raise ValueError("read-back id does not match the created id")
if (comment.get("user") or {}).get("login") != acting_login:
raise ValueError("created comment is not authored by the acting identity")
if comment.get("body") != expected_body: if comment.get("body") != expected_body:
raise ValueError("comment body mismatch") raise ValueError("created comment body does not match")
if not issue_path.endswith(expected_suffix): if not _belongs(comment.get("pull_request_url"), expected_pr_path):
raise ValueError("repository or PR mismatch") raise ValueError("claimed PR comment did not land on a pull request (kind=pulls) on this provider/repo")
except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error: except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
print(f"Error: Gitea comment persistence verification failed: {error}", file=sys.stderr) print(f"Error: Gitea comment persistence verification failed: {error}", file=sys.stderr)
raise SystemExit(1) raise SystemExit(1)
PY PY
then
true echo "$created_id"
return 0
}
# Resolve and cache the Gitea REST endpoint + token for the current remote,
# bound to a SPECIFIC login identity ($1). Populates GITEA_API_ROOT (…/api/v1),
# GITEA_API_BASE (…/api/v1/repos/<slug>), and GITEA_API_TOKEN.
#
# The token is resolved for the EFFECTIVE login (the --login override when
# given, otherwise the detected default), so the one credential used to submit
# the review/comment ALSO drives the /user identity read and every read-back —
# write token and read-back token are the same identity by construction. This
# is the credential-ordering fix: a --login override is no longer submitted
# under one credential and verified under a different default one. Falls back to
# the host-scoped credential ONLY when NO --login override was supplied (the
# best-effort default path). When $2 is "explicit" the login came from a
# caller-supplied --login: that exact login's token MUST resolve, and we FAIL
# CLOSED rather than silently downgrading the review/comment to the host default
# identity. Returns non-zero (clear stderr) on any resolution failure.
gitea_resolve_api_for_login() {
local effective_login="$1" override_explicit="${2:-}" host configured_url repo
host=$(get_remote_host)
if [[ -n "$override_explicit" ]]; then
GITEA_API_TOKEN=$(get_gitea_token_for_login "$effective_login" "$host") || {
echo "Error: could not resolve a host-matched Gitea token for --login '$effective_login' on host '$host'; refusing to fall back to the host default identity or a cross-host credential (review write/read-back)" >&2
return 1
}
else else
GITEA_API_TOKEN=$(get_gitea_token_for_login "$effective_login" "$host") \
|| GITEA_API_TOKEN=$(get_gitea_token "$host") || {
echo "Error: Gitea token not found for login '$effective_login' (review write/read-back)" >&2
return 1
}
fi
configured_url=$(get_gitea_url_for_host "$host") || {
echo "Error: Configured Gitea URL not found for review read-back verification" >&2
return 1
}
repo=$(get_gitea_repo_slug_for_url "$configured_url") || {
echo "Error: Could not resolve Gitea owner/repository relative to configured URL" >&2
return 1
}
GITEA_API_ROOT="${configured_url%/}/api/v1"
GITEA_API_BASE="$GITEA_API_ROOT/repos/$repo"
# The provider WEB base (scheme + host + effective port + any deployment path
# prefix) that Gitea uses to build a comment's html issue_url/pull_request_url.
# Read-back verification pins the returned URL's origin + path prefix to THIS,
# not just a repo/PR suffix.
GITEA_WEB_BASE="${configured_url%/}"
return 0
}
# Resolve the login of the identity the API token authenticates as (GET
# /user). Used to attribute a read-back review to THIS action's reviewer so a
# concurrent review from a DIFFERENT identity cannot satisfy verification.
# Prints the login on success.
gitea_authenticated_login() {
local response_file auth_config status
response_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-whoami.XXXXXX")
auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$response_file"
echo "Error: could not stage Gitea credential for identity read" >&2
return 1
}
trap 'rm -f "$response_file" "$auth_config"' RETURN
if ! status=$(curl -sS -o "$response_file" -w '%{http_code}' \
--config "$auth_config" \
"$GITEA_API_ROOT/user"); then
echo "Error: Gitea authenticated-identity read transport failed" >&2
return 1
fi
if [[ "$status" != "200" ]]; then
echo "Error: Gitea authenticated-identity read failed with HTTP $status" >&2
return 1 return 1
fi fi
echo "$comment_id" python3 - "$response_file" <<'PY'
import json
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
user = json.load(response)
login = user.get("login") if isinstance(user, dict) else None
if not isinstance(login, str) or not login:
raise ValueError("missing authenticated login")
except (OSError, json.JSONDecodeError, TypeError, ValueError) as error:
print(f"Error: could not resolve authenticated Gitea identity: {error}", file=sys.stderr)
raise SystemExit(1)
print(login)
PY
}
# GET /pulls/{n} into a caller-owned response file and print its head commit
# SHA. This core sets NO RETURN trap and reuses a caller-provided auth config +
# response file, so it is safe to call from INSIDE another trapped function
# (the post-verify re-read below) without clobbering that function's cleanup
# trap. $1 = PR number, $2 = response file, $3 = curl auth config file.
gitea_read_pr_head_into() {
local pr_number="$1" pr_file="$2" auth_config="$3" status
if ! status=$(curl -sS -o "$pr_file" -w '%{http_code}' \
--config "$auth_config" \
"$GITEA_API_BASE/pulls/$pr_number"); then
echo "Error: Gitea PR head read transport failed" >&2
return 1
fi
if [[ "$status" != "200" ]]; then
echo "Error: Gitea PR head read failed with HTTP $status" >&2
return 1
fi
python3 - "$pr_file" <<'PY'
import json
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
pr = json.load(response)
head_sha = pr.get("head", {}).get("sha") if isinstance(pr, dict) else None
if not isinstance(head_sha, str) or not head_sha:
raise ValueError("missing PR head sha")
except (OSError, json.JSONDecodeError, AttributeError, TypeError, ValueError) as error:
print(f"Error: could not resolve PR head commit: {error}", file=sys.stderr)
raise SystemExit(1)
print(head_sha)
PY
}
# Resolve the PR's current head commit SHA (GET /pulls/{n}). The review is
# submitted against — and later verified as pinned to — this exact commit, so a
# stale review left over from an earlier push cannot be mistaken for this one.
# Prints the head SHA on success.
gitea_pr_head_sha() {
local pr_number="$1" pr_file auth_config
pr_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-head.XXXXXX")
auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$pr_file"
echo "Error: could not stage Gitea credential for PR head read" >&2
return 1
}
trap 'rm -f "$pr_file" "$auth_config"' RETURN
gitea_read_pr_head_into "$pr_number" "$pr_file" "$auth_config"
}
# Submit a review to a Gitea PR via the supported REST API and verify it against
# a PROVIDER-RETURNED created id. tea 0.11.1's `pr approve`/`reject` cannot emit
# the id of the review it created and can silently no-op while exiting 0 (#865
# defect class), so this does NOT shell out to tea: it POSTs to
# /pulls/{n}/reviews with the event (APPROVED / REQUEST_CHANGES), the PR head
# commit_id, and the review body, which returns the created review object
# including its id. It then GETs that exact review id and requires
# id == created id AND author == acting identity AND state == expected AND
# commit_id == PR head. Keying to the returned id means no concurrent review
# (even same identity/state/head) can masquerade as this one, and a no-op
# submit yields no id and fails closed. Prints the created review id on success.
#
# Args: $1 = PR number, $2 = event (APPROVED|REQUEST_CHANGES),
# $3 = review body (may be empty for APPROVED), $4 = acting login,
# $5 = PR head sha.
gitea_submit_review_verified() {
local pr_number="$1" event="$2" review_body="$3" acting_login="$4" head_sha="$5"
local payload write_file readback_file recheck_file auth_config
local write_status readback_status created_id live_head
payload=$(REVIEW_EVENT="$event" REVIEW_BODY="$review_body" REVIEW_COMMIT="$head_sha" python3 -c '
import json
import os
print(json.dumps({
"event": os.environ["REVIEW_EVENT"],
"body": os.environ["REVIEW_BODY"],
"commit_id": os.environ["REVIEW_COMMIT"],
}))
')
write_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-submit.XXXXXX")
readback_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-getid.XXXXXX")
recheck_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-pr-review-recheck.XXXXXX")
auth_config=$(gitea_write_auth_config "$GITEA_API_TOKEN") || {
rm -f "$write_file" "$readback_file" "$recheck_file"
echo "Error: could not stage Gitea credential for review submit" >&2
return 1
}
trap 'rm -f "$write_file" "$readback_file" "$recheck_file" "$auth_config"' RETURN
if ! write_status=$(curl -sS -o "$write_file" -w '%{http_code}' \
-X POST \
--config "$auth_config" \
-H 'Content-Type: application/json' \
-d "$payload" \
"$GITEA_API_BASE/pulls/$pr_number/reviews"); then
echo "Error: Gitea review submit transport failed" >&2
return 1
fi
# Gitea returns 200 (occasionally 201) with the created review object.
if [[ "$write_status" != "200" && "$write_status" != "201" ]]; then
echo "Error: Gitea review submit failed with HTTP $write_status (#865: no durable review created)" >&2
return 1
fi
created_id=$(python3 - "$write_file" <<'PY'
import json
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
review = json.load(response)
created_id = review.get("id") if isinstance(review, dict) else None
if not isinstance(created_id, int) or created_id <= 0:
raise ValueError("submit response carried no positive review id")
except (OSError, json.JSONDecodeError, ValueError) as error:
print(f"Error: could not identify created Gitea review: {error}", file=sys.stderr)
raise SystemExit(1)
print(created_id)
PY
) || return 1
if ! readback_status=$(curl -sS -o "$readback_file" -w '%{http_code}' \
--config "$auth_config" \
"$GITEA_API_BASE/pulls/$pr_number/reviews/$created_id"); then
echo "Error: Gitea review read-back transport failed" >&2
return 1
fi
if [[ "$readback_status" != "200" ]]; then
echo "Error: Gitea review read-back failed with HTTP $readback_status" >&2
return 1
fi
EXPECTED_REVIEW_ID="$created_id" EXPECTED_STATE="$event" ACTING_LOGIN="$acting_login" \
EXPECTED_HEAD_SHA="$head_sha" EXPECTED_REVIEW_BODY="$review_body" \
python3 - "$readback_file" <<'PY' || return 1
import json
import os
import sys
try:
with open(sys.argv[1], encoding="utf-8") as response:
review = json.load(response)
if not isinstance(review, dict):
raise ValueError("response is not a review object")
expected_id = int(os.environ["EXPECTED_REVIEW_ID"])
expected_state = os.environ["EXPECTED_STATE"]
acting_login = os.environ["ACTING_LOGIN"]
expected_head = os.environ["EXPECTED_HEAD_SHA"]
expected_body = os.environ["EXPECTED_REVIEW_BODY"]
if review.get("id") != expected_id:
raise ValueError("read-back id does not match the created id")
if (review.get("user") or {}).get("login") != acting_login:
raise ValueError("created review is not authored by the acting identity")
if review.get("state") != expected_state:
raise ValueError("created review is not in the expected state")
if review.get("commit_id") != expected_head:
raise ValueError("created review is not pinned to the PR head commit")
# Bind to the exact submitted body. On Gitea v1.25.4 SubmitReview may
# finalize/reuse a pending review id whose Content was authored elsewhere;
# the exact GET exposes the persisted body, so a mismatch (a reused/foreign
# review carrying different Content) fails closed even when id/author/state/
# head all line up. Require presence + string TYPE + exact equality rather
# than `(body or "")`: the old coalesce treated a missing/null persisted body
# as equal to an empty submitted one, so a non-empty submitted body that
# persisted as null (a suppressed/lost body) would have passed. When a
# non-empty body was submitted the persisted value MUST be that exact string;
# when an empty body was submitted the persisted value must be empty or
# absent (a non-empty persisted body is likewise a divergence — vice-versa).
persisted_body = review.get("body")
if expected_body == "":
if persisted_body not in (None, ""):
raise ValueError("created review carries a body but none was submitted")
elif not isinstance(persisted_body, str) or persisted_body != expected_body:
raise ValueError("created review body does not match the submitted body")
except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
print(f"Error: Gitea review persistence verification failed: {error}", file=sys.stderr)
raise SystemExit(1)
PY
# Current-head TOCTOU close-out: the review verified above is pinned to
# head_sha, but that head was read BEFORE the submit. Between then and now
# the PR branch may have advanced (a force-push or a new commit), which would
# leave this verified review attached to a now-superseded commit while the
# live tip carries unreviewed code — yet the wrapper would still report
# success. Re-read the LIVE PR head and require it STILL equals the submitted
# SHA; if it advanced, fail closed (nonzero, no created id emitted, no
# success line). This reuses the submit-scoped auth config + recheck file so
# it neither leaks the token to argv nor clobbers this function's cleanup.
live_head=$(gitea_read_pr_head_into "$pr_number" "$recheck_file" "$auth_config") || {
echo "Error: could not re-read Gitea PR head after review verification" >&2
return 1
}
if [[ "$live_head" != "$head_sha" ]]; then
echo "Error: Gitea PR head advanced from $head_sha to $live_head between review submit and verification; refusing to report a review pinned to a superseded commit (#865 current-head TOCTOU)" >&2
return 1
fi
echo "$created_id"
return 0 return 0
} }
@@ -205,40 +557,76 @@ if [[ "$PLATFORM" == "github" ]]; then
elif [[ "$PLATFORM" == "gitea" ]]; then elif [[ "$PLATFORM" == "gitea" ]]; then
case $ACTION in case $ACTION in
approve) approve)
repo=$(get_repo_slug)
host=$(get_remote_host) host=$(get_remote_host)
login=$(get_gitea_login_for_host "$host") # A --login override always wins. Otherwise name this host's login
# tea v0.11.1 defines no --comment/-comment flag on `pr approve`; # only as a best effort: the login name merely selects a per-login
# route any review body via the durable comment API instead (#835). # token, and gitea_resolve_api_for_login falls back to the host
tea pr approve "$PR_NUMBER" --repo "$repo" --login "$login" # credential (get_gitea_token) when no tea login is named — so a host
echo "Approved Gitea PR #$PR_NUMBER" # tea's login list need not enumerate exotic (e.g. ported) hosts for
if [[ -n "$COMMENT" ]]; then # the default credential to resolve. The single resolved token is
comment_id=$(gitea_post_verified_comment "$PR_NUMBER" "$COMMENT") || exit 1 # then used for the write, the /user identity, and the read-back.
echo "Added and verified review comment on Gitea PR #$PR_NUMBER (comment ID $comment_id)" EFFECTIVE_LOGIN="$LOGIN_OVERRIDE"
fi [[ -n "$EFFECTIVE_LOGIN" ]] || EFFECTIVE_LOGIN=$(get_gitea_login_for_host "$host" 2>/dev/null || true)
# Bind the REST endpoint + token to the effective login, then derive
# the acting identity from that SAME credential so the review submit
# and its read-back verify against the identity that performed them.
gitea_resolve_api_for_login "$EFFECTIVE_LOGIN" "${LOGIN_OVERRIDE:+explicit}" || exit 1
ACTING_LOGIN=$(gitea_authenticated_login) || exit 1
head_sha=$(gitea_pr_head_sha "$PR_NUMBER") || exit 1
# The review body (if any) travels with the review itself in the REST
# submit — the created review record carries it — so there is no
# separate detached comment to reconcile.
review_id=$(gitea_submit_review_verified "$PR_NUMBER" "APPROVED" "$COMMENT" "$ACTING_LOGIN" "$head_sha") || {
echo "Error: could not submit and verify an APPROVED review on Gitea PR #$PR_NUMBER via a provider-returned created id (#865)." >&2
exit 1
}
echo "Approved and verified Gitea PR #$PR_NUMBER (review ID $review_id)"
;; ;;
request-changes) request-changes)
if [[ -z "$COMMENT" ]]; then if [[ -z "$COMMENT" ]]; then
echo "Error: Comment required for request-changes" echo "Error: Comment required for request-changes"
exit 1 exit 1
fi fi
repo=$(get_repo_slug)
host=$(get_remote_host) host=$(get_remote_host)
login=$(get_gitea_login_for_host "$host") # A --login override always wins. Otherwise name this host's login
# tea v0.11.1 defines no --comment/-comment flag on `pr reject`; # only as a best effort: the login name merely selects a per-login
# route the review body via the durable comment API instead (#835). # token, and gitea_resolve_api_for_login falls back to the host
tea pr reject "$PR_NUMBER" --repo "$repo" --login "$login" # credential (get_gitea_token) when no tea login is named — so a host
echo "Requested changes on Gitea PR #$PR_NUMBER" # tea's login list need not enumerate exotic (e.g. ported) hosts for
comment_id=$(gitea_post_verified_comment "$PR_NUMBER" "$COMMENT") || exit 1 # the default credential to resolve. The single resolved token is
echo "Added and verified review comment on Gitea PR #$PR_NUMBER (comment ID $comment_id)" # then used for the write, the /user identity, and the read-back.
EFFECTIVE_LOGIN="$LOGIN_OVERRIDE"
[[ -n "$EFFECTIVE_LOGIN" ]] || EFFECTIVE_LOGIN=$(get_gitea_login_for_host "$host" 2>/dev/null || true)
gitea_resolve_api_for_login "$EFFECTIVE_LOGIN" "${LOGIN_OVERRIDE:+explicit}" || exit 1
ACTING_LOGIN=$(gitea_authenticated_login) || exit 1
head_sha=$(gitea_pr_head_sha "$PR_NUMBER") || exit 1
review_id=$(gitea_submit_review_verified "$PR_NUMBER" "REQUEST_CHANGES" "$COMMENT" "$ACTING_LOGIN" "$head_sha") || {
echo "Error: could not submit and verify a REQUEST_CHANGES review on Gitea PR #$PR_NUMBER via a provider-returned created id (#865)." >&2
exit 1
}
echo "Requested changes and verified on Gitea PR #$PR_NUMBER (review ID $review_id)"
;; ;;
comment) comment)
if [[ -z "$COMMENT" ]]; then if [[ -z "$COMMENT" ]]; then
echo "Error: Comment required" echo "Error: Comment required"
exit 1 exit 1
fi fi
host=$(get_remote_host)
comment_id=$(gitea_post_verified_comment "$PR_NUMBER" "$COMMENT") || exit 1 # A --login override always wins. Otherwise name this host's login
# only as a best effort: the login name merely selects a per-login
# token, and gitea_resolve_api_for_login falls back to the host
# credential (get_gitea_token) when no tea login is named — so a host
# tea's login list need not enumerate exotic (e.g. ported) hosts for
# the default credential to resolve. The single resolved token is
# then used for the write, the /user identity, and the read-back.
EFFECTIVE_LOGIN="$LOGIN_OVERRIDE"
[[ -n "$EFFECTIVE_LOGIN" ]] || EFFECTIVE_LOGIN=$(get_gitea_login_for_host "$host" 2>/dev/null || true)
gitea_resolve_api_for_login "$EFFECTIVE_LOGIN" "${LOGIN_OVERRIDE:+explicit}" || exit 1
ACTING_LOGIN=$(gitea_authenticated_login) || exit 1
comment_id=$(gitea_create_comment_verified "$PR_NUMBER" "$COMMENT" "$ACTING_LOGIN") || {
echo "Error: could not create and verify a comment on Gitea PR #$PR_NUMBER via a provider-returned created id (#865)." >&2
exit 1
}
echo "Added and verified comment on Gitea PR #$PR_NUMBER (comment ID $comment_id)" echo "Added and verified comment on Gitea PR #$PR_NUMBER (comment ID $comment_id)"
;; ;;
*) *)

View File

@@ -0,0 +1,153 @@
#!/usr/bin/env bash
# Regression harness for ci-queue-wait.sh's 404-branch-absent handling.
#
# gitea_get_branch_head_sha() resolves a branch's head SHA before the
# pre-push queue guard runs. A branch that has never been pushed doesn't
# exist on the remote yet, so Gitea's branches/<branch> endpoint 404s.
# Before the fix, `curl -fsSL` failed on the 404, its empty stdout was piped
# into `python3 -c 'json.load(sys.stdin)'`, and the resulting
# JSONDecodeError crashed the guard -- blocking every new feature branch's
# first push. The fix must treat 404 as "no in-flight pipeline" (queue
# clear) while still failing closed on a genuine API error.
#
# Covers:
# (a) 404 branch-absent -> exit 0, "queue clear" message.
# (b) 200 existing branch + a terminal CI state -> unchanged behavior.
# (c) genuine API error (500) -> still fail-closed (nonzero exit).
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
WORK_DIR="${MOSAIC_TEST_WORK_DIR:-$PWD/.mosaic-test-work/ci-queue-wait-branch-absent}"
REPO_DIR="$WORK_DIR/repo"
STUB_DIR="$WORK_DIR/stubs"
rm -rf "$WORK_DIR"
mkdir -p "$REPO_DIR" "$STUB_DIR"
git -C "$REPO_DIR" init -q
git -C "$REPO_DIR" remote add origin https://git.example.test/acme/widgets.git
# Minimal curl stub. Selects a canned response by inspecting which Gitea
# endpoint is being hit (branches/<branch> vs commits/<sha>/status) and
# whether -w '%{http_code}' was requested. Only the patched branch-lookup
# call passes -w; the unpatched call and the (unchanged) status call both
# use plain `curl -fsSL` semantics -- exit nonzero and print nothing on a
# non-2xx response. This lets the same stub exercise both the pre-fix and
# post-fix branch-lookup code paths faithfully.
cat > "$STUB_DIR/curl" <<'SH'
#!/usr/bin/env bash
set -euo pipefail
has_w=0
url=""
for arg in "$@"; do
case "$arg" in
-w) has_w=1 ;;
http://*|https://*) url="$arg" ;;
esac
done
case "$url" in
*/branches/*) mode="${MOSAIC_STUB_BRANCH_MODE:?MOSAIC_STUB_BRANCH_MODE not set}" ;;
*/status) mode="${MOSAIC_STUB_STATUS_MODE:-terminal-success}" ;;
*)
echo "curl stub: unrecognized URL: $url" >&2
exit 2
;;
esac
case "$mode" in
404) code=404; body="" ;;
200) code=200; body='{"commit":{"id":"deadbeefcafef00d0123456789abcdef01234567"}}' ;;
500) code=500; body='{"message":"internal server error"}' ;;
no-status) code=200; body='{}' ;;
terminal-success) code=200; body='{"state":"success"}' ;;
*)
echo "curl stub: unknown mode=$mode" >&2
exit 2
;;
esac
if [[ "$has_w" == 1 ]]; then
printf '%s\n%s' "$body" "$code"
exit 0
fi
# Unpatched branch-lookup call / status-endpoint call: real curl -fsSL
# exits nonzero and emits nothing on stdout for a non-2xx response.
if [[ "$code" != "200" ]]; then
exit 22
fi
printf '%s' "$body"
SH
chmod +x "$STUB_DIR/curl"
run_ci_queue_wait() {
local branch="$1"
(
cd "$REPO_DIR"
export PATH="$STUB_DIR:$PATH"
export MOSAIC_CREDENTIALS_FILE="$WORK_DIR/no-credentials.json"
export GITEA_TOKEN="stub-token"
export GITEA_URL="https://git.example.test"
"$SCRIPT_DIR/ci-queue-wait.sh" -B "$branch" --purpose push -t 5 -i 1
)
}
fail=0
# (a) 404 branch-absent -> queue clear, exit 0.
set +e
out_a=$(MOSAIC_STUB_BRANCH_MODE=404 run_ci_queue_wait "feat/not-pushed-yet" 2>&1)
status_a=$?
set -e
if [[ "$status_a" -ne 0 ]]; then
echo "FAIL(a): expected exit 0 for 404 branch-absent, got $status_a" >&2
echo "$out_a" >&2
fail=1
elif [[ "$out_a" != *"queue clear"* ]]; then
echo "FAIL(a): expected a queue-clear message, got:" >&2
echo "$out_a" >&2
fail=1
fi
# (b) 200 existing branch + terminal CI state -> unchanged behavior, exit 0.
set +e
out_b=$(MOSAIC_STUB_BRANCH_MODE=200 MOSAIC_STUB_STATUS_MODE=terminal-success run_ci_queue_wait "main" 2>&1)
status_b=$?
set -e
if [[ "$status_b" -ne 0 ]]; then
echo "FAIL(b): expected exit 0 for existing branch with terminal status, got $status_b" >&2
echo "$out_b" >&2
fail=1
elif [[ "$out_b" != *"sha=deadbeefcafef00d0123456789abcdef01234567"* ]]; then
echo "FAIL(b): expected the resolved HEAD SHA to be logged, got:" >&2
echo "$out_b" >&2
fail=1
elif [[ "$out_b" == *"queue clear"* ]]; then
echo "FAIL(b): an existing branch must not take the branch-absent path" >&2
echo "$out_b" >&2
fail=1
fi
# (c) genuine API error (500) -> still fail-closed, exit nonzero.
set +e
out_c=$(MOSAIC_STUB_BRANCH_MODE=500 run_ci_queue_wait "feat/some-branch" 2>&1)
status_c=$?
set -e
if [[ "$status_c" -eq 0 ]]; then
echo "FAIL(c): expected a nonzero exit for a genuine 500 API error, got 0" >&2
echo "$out_c" >&2
fail=1
elif [[ "$out_c" == *"queue clear"* ]]; then
echo "FAIL(c): a genuine API error must not be reported as queue-clear" >&2
echo "$out_c" >&2
fail=1
fi
if [[ "$fail" -eq 0 ]]; then
echo "ci-queue-wait branch-absent regression passed (3/3 cases)"
fi
exit "$fail"

View File

@@ -0,0 +1,161 @@
#!/usr/bin/env bash
# Regression harness for `git-credential-mosaic` — per-agent Gitea identity
# resolution (Gate-16 author≠reviewer separation).
#
# Covers:
# 1. Identity resolution priority: MOSAIC_GIT_IDENTITY env > git config
# mosaic.gitIdentity (per-worktree) > git-supplied username.
# 2. Correct per-slot token file path chosen per host
# (gitea-usc-<id>.token vs gitea-mosaicstack-<id>.token).
# 3. Per-slot token present -> emits that identity + token.
# 4. Per-slot token absent -> falls back to the shared account
# (backward-compat / no-op for hosts without per-slot tokens).
# 5. Unknown/unrelated host -> exits 0 with no output (passthrough).
#
# Uses stubbed token files under a fake HOME + a real (throwaway) git repo.
# NEVER reads real secrets or touches the real ~/.config/mosaic/secrets.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
WORK_DIR="${MOSAIC_TEST_WORK_DIR:-$PWD/.mosaic-test-work/git-credential-mosaic}"
FAKE_HOME="$WORK_DIR/home"
REPO_DIR="$WORK_DIR/repo"
# Mirror the real deployed layout (~/.config/mosaic/tools/{git,_lib}/) under the
# fake HOME: git-credential-mosaic resolves its credentials.sh sibling via a
# script-relative path (BASH_SOURCE), so the copy must live next to a stubbed
# _lib/credentials.sh, not the real one, to keep this test hermetic.
HELPER="$FAKE_HOME/.config/mosaic/tools/git/git-credential-mosaic"
rm -rf "$WORK_DIR"
mkdir -p "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens" \
"$FAKE_HOME/.config/mosaic/tools/git" \
"$FAKE_HOME/.config/mosaic/tools/_lib" \
"$REPO_DIR"
cp "$SCRIPT_DIR/git-credential-mosaic" "$HELPER"
chmod +x "$HELPER"
git -C "$REPO_DIR" init -q
git -C "$REPO_DIR" config user.email "test@example.invalid"
git -C "$REPO_DIR" config user.name "Test"
# Fake shared-account credential loader — stands in for
# tools/_lib/credentials.sh's load_credentials(), scoped to this test only.
cat > "$FAKE_HOME/.config/mosaic/tools/_lib/credentials.sh" <<'SH'
load_credentials() {
case "$1" in
gitea-mosaicstack) GITEA_URL="https://git.mosaicstack.dev"; GITEA_TOKEN="shared-mosaicstack-token"; export GITEA_URL GITEA_TOKEN; return 0 ;;
gitea-usc) GITEA_URL="https://git.uscllc.com"; GITEA_TOKEN="shared-usc-token"; export GITEA_URL GITEA_TOKEN; return 0 ;;
*) return 1 ;;
esac
}
SH
fail=0
assert_eq() {
local desc="$1" expected="$2" actual="$3"
if [[ "$expected" != "$actual" ]]; then
echo "FAIL: $desc — expected '$expected', got '$actual'" >&2
fail=1
fi
}
# Feed "host=<h>\nusername=<u>\n\n" on stdin (mirrors git's credential protocol)
# and run the helper with the fake HOME, inside REPO_DIR (so `git config
# mosaic.gitIdentity` resolves per-worktree), plus any extra env passed in $@.
run_helper() {
local host="$1" username_in="$2"; shift 2
(
cd "$REPO_DIR"
env -i HOME="$FAKE_HOME" PATH="$PATH" "$@" bash "$HELPER" get <<EOF
host=$host
username=$username_in
EOF
)
}
# ---------------------------------------------------------------------------
# 1. No identity resolvable anywhere, no per-slot token -> shared fallback
# (backward-compat: unchanged behavior when nothing is configured).
# ---------------------------------------------------------------------------
git -C "$REPO_DIR" config --unset mosaic.gitIdentity 2>/dev/null || true
out=$(run_helper "git.mosaicstack.dev" "")
assert_eq "shared fallback: username" "username=git" "$(echo "$out" | grep '^username=')"
assert_eq "shared fallback: password" "password=shared-mosaicstack-token" "$(echo "$out" | grep '^password=')"
# ---------------------------------------------------------------------------
# 2. git-supplied username resolves to an identity WITH a per-slot token ->
# that identity + token wins over the shared account.
# ---------------------------------------------------------------------------
echo -n "agentA-mosaicstack-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-mosaicstack-agentA.token"
out=$(run_helper "git.mosaicstack.dev" "agentA")
assert_eq "username-resolved identity: username" "username=agentA" "$(echo "$out" | grep '^username=')"
assert_eq "username-resolved identity: password" "password=agentA-mosaicstack-token" "$(echo "$out" | grep '^password=')"
# ---------------------------------------------------------------------------
# 3. git config mosaic.gitIdentity (per-worktree) beats git-supplied username.
# ---------------------------------------------------------------------------
echo -n "agentB-mosaicstack-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-mosaicstack-agentB.token"
git -C "$REPO_DIR" config mosaic.gitIdentity agentB
out=$(run_helper "git.mosaicstack.dev" "agentA")
assert_eq "git-config beats username: username" "username=agentB" "$(echo "$out" | grep '^username=')"
assert_eq "git-config beats username: password" "password=agentB-mosaicstack-token" "$(echo "$out" | grep '^password=')"
# ---------------------------------------------------------------------------
# 4. MOSAIC_GIT_IDENTITY env beats git config mosaic.gitIdentity.
# ---------------------------------------------------------------------------
echo -n "agentC-mosaicstack-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-mosaicstack-agentC.token"
out=$(run_helper "git.mosaicstack.dev" "agentA" MOSAIC_GIT_IDENTITY=agentC)
assert_eq "env beats git-config: username" "username=agentC" "$(echo "$out" | grep '^username=')"
assert_eq "env beats git-config: password" "password=agentC-mosaicstack-token" "$(echo "$out" | grep '^password=')"
git -C "$REPO_DIR" config --unset mosaic.gitIdentity
# ---------------------------------------------------------------------------
# 5. Identity resolves, but no matching per-slot token file -> falls back to
# the shared account (per-agent identity is opt-in, not a hard requirement).
# ---------------------------------------------------------------------------
out=$(run_helper "git.mosaicstack.dev" "no-such-agent")
assert_eq "no per-slot token: username" "username=git" "$(echo "$out" | grep '^username=')"
assert_eq "no per-slot token: password" "password=shared-mosaicstack-token" "$(echo "$out" | grep '^password=')"
# ---------------------------------------------------------------------------
# 6. Correct per-slot token PATH is chosen per host: same agent id, different
# host prefix (gitea-usc- vs gitea-mosaicstack-).
# ---------------------------------------------------------------------------
echo -n "agentD-usc-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-usc-agentD.token"
out=$(run_helper "git.uscllc.com" "agentD")
assert_eq "host-scoped token path (usc): username" "username=agentD" "$(echo "$out" | grep '^username=')"
assert_eq "host-scoped token path (usc): password" "password=agentD-usc-token" "$(echo "$out" | grep '^password=')"
# agentD has NO mosaicstack token -> must fall back to shared mosaicstack, not
# leak the usc token across hosts.
out=$(run_helper "git.mosaicstack.dev" "agentD")
assert_eq "host-scoped token path (cross-host must not leak): username" "username=git" "$(echo "$out" | grep '^username=')"
assert_eq "host-scoped token path (cross-host must not leak): password" "password=shared-mosaicstack-token" "$(echo "$out" | grep '^password=')"
# ---------------------------------------------------------------------------
# 7. Unrelated/unknown host -> exit 0, no output (passthrough for non-Gitea
# remotes, e.g. github.com via a different credential helper).
# ---------------------------------------------------------------------------
out=$(run_helper "github.com" "agentA")
assert_eq "unknown host: no output" "" "$out"
# ---------------------------------------------------------------------------
# 8. Non-"get" verb (store/erase) -> exit 0, no output (git-credential
# protocol: this helper only implements get).
# ---------------------------------------------------------------------------
store_out=$(cd "$REPO_DIR" && env -i HOME="$FAKE_HOME" PATH="$PATH" bash "$HELPER" store <<EOF
host=git.mosaicstack.dev
username=agentA
password=whatever
EOF
)
assert_eq "store verb: no output" "" "$store_out"
if [[ "$fail" -eq 0 ]]; then
echo "git-credential-mosaic identity resolution regression passed"
fi
exit "$fail"

View File

@@ -312,4 +312,901 @@ if [[ "$override_wins" != "mosaicstack" ]]; then
fi fi
git -C "$REPO_DIR" remote set-url origin https://git.uscllc.com/USC/uconnect.git git -C "$REPO_DIR" remote set-url origin https://git.uscllc.com/USC/uconnect.git
# ---------------------------------------------------------------------------
# #865 Blocker 1 & 2: get_gitea_token_for_login must resolve the SAME token as
# PyYAML would (or fail closed identically) even when PyYAML is ABSENT, and must
# bind the credential to the repo host's scheme + host + EFFECTIVE PORT — not the
# hostname alone. These fixtures probe the ImportError-dispatched line-parser
# fallback under FORCED PyYAML absence with adversarial YAML shapes, asserting it
# NEVER misattributes a token from a nested sub-map or a mis-indented line, strips
# inline comments like PyYAML, fails closed where PyYAML errors, and rejects a
# port mismatch while accepting an exact / default-port match. When PyYAML is
# available the same fixtures also assert the PyYAML path agrees (equivalence).
# ---------------------------------------------------------------------------
FIXTURE_XDG="$WORK_DIR/tokenfix"
NOYAML_DIR="$WORK_DIR/noyaml"
mkdir -p "$FIXTURE_XDG/tea" "$NOYAML_DIR"
# A shadow `yaml` module that raises ImportError, forcing the fallback path.
printf 'raise ImportError("forced-absent for #865 fallback regression")\n' > "$NOYAML_DIR/yaml.py"
if python3 -c 'import yaml' >/dev/null 2>&1; then HAVE_PYYAML=true; else HAVE_PYYAML=false; fi
# Confirm the shim really does force ImportError, so the fallback is exercised.
if python3 -c 'import yaml' >/dev/null 2>&1; then
if PYTHONPATH="$NOYAML_DIR" python3 -c 'import yaml' >/dev/null 2>&1; then
echo "FAIL: PyYAML-absence shim did not force ImportError (fallback not exercised)" >&2
exit 1
fi
fi
write_fixture() { printf '%s' "$1" > "$FIXTURE_XDG/tea/config.yml"; }
# Resolve a token via the FORCED-fallback path (PyYAML shimmed to ImportError).
token_fallback() {
(
cd "$REPO_DIR"
XDG_CONFIG_HOME="$FIXTURE_XDG" PYTHONPATH="$NOYAML_DIR" bash -c '
source "'"$SCRIPT_DIR"'/detect-platform.sh"
get_gitea_token_for_login "$1" "$2"
' _ "$1" "$2"
) 2>/dev/null || true
}
# Resolve a token via the normal path (uses PyYAML when installed).
token_pyyaml() {
(
cd "$REPO_DIR"
XDG_CONFIG_HOME="$FIXTURE_XDG" bash -c '
source "'"$SCRIPT_DIR"'/detect-platform.sh"
get_gitea_token_for_login "$1" "$2"
' _ "$1" "$2"
) 2>/dev/null || true
}
assert_token() {
local desc="$1" expected="$2" login="$3" host="$4" got
got=$(token_fallback "$login" "$host")
if [[ "$got" != "$expected" ]]; then
echo "FAIL fallback [$desc]: expected [$expected] got [$got]" >&2
exit 1
fi
if [[ "$HAVE_PYYAML" == true ]]; then
got=$(token_pyyaml "$login" "$host")
if [[ "$got" != "$expected" ]]; then
echo "FAIL pyyaml [$desc]: expected [$expected] got [$got]" >&2
exit 1
fi
fi
}
# 1. Plain, well-formed entry resolves its token.
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
'
assert_token "plain scalar" "TOK_PLAIN" primary git.example
# 2. A token nested inside a deeper SUB-MAP must NOT attach to the entry — PyYAML
# resolves the entry's own token to None here, so the fallback must too.
write_fixture 'logins:
- name: primary
url: https://git.example
extra:
token: TOK_NESTED_ATTACKER
- name: other
url: https://git.example
token: TOK_OTHER
'
assert_token "nested sub-map token is not attributed" "" primary git.example
assert_token "sibling entry still resolves its own token" "TOK_OTHER" other git.example
# 3. A MIS-INDENTED token line (deeper than the entry's fields) must not attach;
# PyYAML errors on this shape, so both fail closed.
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_MISINDENT
'
assert_token "mis-indented token fails closed" "" primary git.example
# 4. A trailing inline comment on a scalar is stripped, exactly as PyYAML does.
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_INLINE # trailing note
'
assert_token "inline comment stripped" "TOK_INLINE" primary git.example
# 5. A PyYAML-fail-closed case: tab indentation. PyYAML raises a scanner error;
# the fallback resolves no token. Both fail closed identically.
write_fixture "$(printf 'logins:\n - name: primary\n url: https://git.example\n\ttoken: TOK_TAB\n')"
assert_token "tab-indent fails closed like PyYAML" "" primary git.example
# 6. Host binding is scheme + host + EFFECTIVE PORT, not hostname alone.
write_fixture 'logins:
- name: ported
url: https://git.example:8443
token: TOK_PORTED
'
assert_token "explicit port exact match accepted" "TOK_PORTED" ported git.example:8443
assert_token "portless repo host rejects :8443 login" "" ported git.example
assert_token "wrong explicit port rejected" "" ported git.example:9443
# 7. An implicit (portless) login URL equals the scheme's explicit default port.
write_fixture 'logins:
- name: defported
url: https://git.example
token: TOK_DEFPORT
'
assert_token "implicit https vs explicit :443 match" "TOK_DEFPORT" defported git.example:443
assert_token "implicit https vs :8443 rejected" "" defported git.example:8443
# 8. An UNQUOTED token whose raw text PyYAML's implicit resolver types as a
# NON-string (int / null / bool / float) must fail closed: PyYAML yields a
# non-str value that _accept rejects, so the fallback must NOT surface the
# stringified scalar as a credential. Each raw form fails closed IDENTICALLY
# to PyYAML (a prior residual emitted "12345"/"null"/"true"/etc. here).
assert_nonstring_token_fails_closed() {
local desc="$1" raw="$2"
write_fixture "logins:
- name: primary
url: https://git.example
token: ${raw}
"
assert_token "$desc" "" primary git.example
}
assert_nonstring_token_fails_closed "unquoted int token fails closed" "12345"
assert_nonstring_token_fails_closed "unquoted null token fails closed" "null"
assert_nonstring_token_fails_closed "unquoted tilde-null token fails closed" "~"
assert_nonstring_token_fails_closed "unquoted yes(bool) token fails closed" "yes"
assert_nonstring_token_fails_closed "unquoted true(bool) token fails closed" "true"
assert_nonstring_token_fails_closed "unquoted float token fails closed" "3.14"
# 9. A QUOTED scalar is ALWAYS a string, even when its contents look like a
# non-string implicit form. The quotes force str typing in PyYAML, so the
# fallback must accept the literal (quote-stripped) contents as the token.
write_fixture 'logins:
- name: primary
url: https://git.example
token: "12345"
'
assert_token "double-quoted digit token is a literal string" "12345" primary git.example
write_fixture "logins:
- name: primary
url: https://git.example
token: 'abc'
"
assert_token "single-quoted token is a literal string" "abc" primary git.example
# assert_fallback_fails_closed: the forced-fallback path MUST resolve no token
# (fail closed). Used for STRUCTURAL cases where PyYAML would resolve a DIFFERENT
# token (e.g. duplicate-key last-wins) — the fallback must never surface the
# wrong/stale token, so it fails closed instead; when PyYAML is present we also
# confirm it really does resolve a (divergent) token, proving the fallback is the
# strictly-more-conservative side and the case is a genuine fail-open guard.
assert_fallback_fails_closed() {
local desc="$1" login="$2" host="$3" got
got=$(token_fallback "$login" "$host")
if [[ -n "$got" ]]; then
echo "FAIL fallback [$desc]: expected fail-closed, got a token" >&2
exit 1
fi
if [[ "$HAVE_PYYAML" == true ]]; then
got=$(token_pyyaml "$login" "$host")
if [[ -z "$got" ]]; then
echo "FAIL [$desc]: expected PyYAML to resolve a divergent token" >&2
exit 1
fi
fi
}
# 10. tea's REAL on-disk shape: the `logins:` block SEQUENCE items sit at the
# SAME indentation as the key (dash at column 0), with extra scalar fields.
# The recognizer must resolve this exactly like PyYAML (regression guard so
# the stricter whole-document recognizer does not fail closed on real input).
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_REAL
default: false
ssh_host: ""
- name: other
url: https://other.example
token: TOK_REAL_OTHER
preferences:
editor: false
flags: null
'
assert_token "tea dash-at-column-0 real shape resolves" "TOK_REAL" primary git.example
assert_token "tea real shape sibling resolves own token" "TOK_REAL_OTHER" other other.example
# 11. NESTED-SHADOW: a nested `logins:` (NOT at root scope) must not be mistaken
# for the real root logins. The recognizer parses whole-document structure,
# so it selects the ROOT logins token exactly as PyYAML does — never the
# nested attacker token. (A prior line scan matched the FIRST logins at ANY
# indent and returned ATTACKER.)
write_fixture 'outer:
logins:
- name: primary
url: https://git.example
token: ATTACKER_NESTED
logins:
- name: primary
url: https://git.example
token: ROOT_TOK
'
assert_token "nested logins shadow selects ROOT token" "ROOT_TOK" primary git.example
# 12. BLOCK-SCALAR-SHADOW: text inside a YAML literal/folded block ( | or > ) is
# an OPAQUE scalar to PyYAML (so `logins` is a string, not a list) and must
# not be scanned as live logins entries. Both fail closed.
write_fixture 'logins: |
- name: primary
url: https://git.example
token: ATTACKER_BLOCK
'
assert_token "block-scalar logins value fails closed" "" primary git.example
# A folded/literal block scalar anywhere is outside the recognizer's subset, so
# the fallback fails closed (conservative) even though PyYAML can still resolve
# the real root token past the opaque scalar. Fail-closed is the safe side.
write_fixture 'note: >
logins:
- name: primary
token: ATTACKER_FOLDED
logins:
- name: primary
url: https://git.example
token: ROOT_OK
'
assert_fallback_fails_closed "folded block scalar present fails closed" primary git.example
# 13. DUPLICATE-ROOT / DUPLICATE-FIELD: a duplicated `logins:` root key (PyYAML
# last-wins) or a duplicated field within a login must fail closed rather
# than take the FIRST (stale) value. PyYAML resolves the LAST; the fallback
# refuses to guess.
write_fixture 'logins:
- name: primary
url: https://git.example
token: FIRST_DUP
logins:
- name: primary
url: https://git.example
token: LAST_DUP
'
assert_fallback_fails_closed "duplicate root logins key fails closed" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: FIRST_FIELD
token: SECOND_FIELD
'
assert_fallback_fails_closed "duplicate token field fails closed" primary git.example
# 14. MALFORMED-AFTER-VALID: a syntax error LATER in the file makes PyYAML reject
# the WHOLE document; the recognizer must too (not emit the earlier token).
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
broken: a: b: c
'
assert_token "malformed line after valid login fails closed" "" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
broken: [unclosed
'
assert_token "unclosed flow after valid login fails closed" "" primary git.example
# 15. EXTRA-DOCUMENT: a multi-document file (--- separator, or ... end marker)
# makes PyYAML safe_load reject multi-document input; the recognizer fails
# closed on ANY document marker rather than emit the first doc's token.
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
---
logins:
- name: primary
url: https://git.example
token: SECOND_DOC
'
assert_token "second document (--- separator) fails closed" "" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
...
trailing: 1
'
assert_token "end marker then more content fails closed" "" primary git.example
# 16. CONSTRUCTOR-VALIDITY / INVALID-INDICATOR: a plain scalar can match a typed
# implicit resolver (int/float/timestamp) yet be NON-constructible, or begin
# with an indicator a plain scalar may not start with. PyYAML then RAISES on
# the WHOLE document (constructor error / scanner error) and yields NO token,
# so the fallback must ALSO fail closed for the whole document -- even though
# the (unrelated) malformed key sits alongside an otherwise-valid logins
# block whose token is itself well-formed. A prior residual proved STRUCTURE
# and implicit TYPE but not constructor validity, so it ignored the malformed
# key and still emitted the valid login token (fail-open in the dangerous
# direction). assert_both_fail_closed asserts fallback == PyYAML == no token.
assert_both_fail_closed() {
local desc="$1" login="$2" host="$3" got
got=$(token_fallback "$login" "$host")
if [[ -n "$got" ]]; then
echo "FAIL fallback [$desc]: expected fail-closed, got a token" >&2
exit 1
fi
if [[ "$HAVE_PYYAML" == true ]]; then
got=$(token_pyyaml "$login" "$host")
if [[ -n "$got" ]]; then
echo "FAIL pyyaml [$desc]: expected PyYAML to also fail closed (raise/no token), got a token" >&2
exit 1
fi
fi
}
# write_bad_key_fixture: an unrelated root key carrying $1 as its plain scalar,
# followed by an otherwise-valid logins block whose token is well-formed.
write_bad_key_fixture() {
write_fixture "bad: $1
logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
"
}
# Non-constructible TIMESTAMP-tagged scalars: match the resolver, but the
# calendar field is out of range so PyYAML's datetime construction raises.
write_bad_key_fixture '2023-99-99' # month 99 / day 99 invalid
assert_both_fail_closed "bad-date 2023-99-99 fails closed like PyYAML" primary git.example
write_bad_key_fixture '2023-13-01' # month 13 invalid
assert_both_fail_closed "bad-month 2023-13-01 fails closed like PyYAML" primary git.example
write_bad_key_fixture '2023-01-15T25:00:00' # hour 25 invalid
assert_both_fail_closed "bad-hour timestamp fails closed like PyYAML" primary git.example
# Non-constructible INT-tagged scalars: match the int resolver, but the radix
# body is empty after underscore removal so int(base) raises.
write_bad_key_fixture '0b_'
assert_both_fail_closed "empty-binary 0b_ fails closed like PyYAML" primary git.example
write_bad_key_fixture '0x_'
assert_both_fail_closed "empty-hex 0x_ fails closed like PyYAML" primary git.example
write_bad_key_fixture '0x__'
assert_both_fail_closed "empty-hex 0x__ (multi-underscore) fails closed" primary git.example
# Invalid plain-scalar INDICATOR forms: a plain scalar may not begin with '%'
# (directive) or ',' (flow) -- PyYAML raises a scanner/parser error on the whole
# document, so the fallback fails closed on the leading indicator.
write_bad_key_fixture '%broken'
assert_both_fail_closed "leading-%% directive indicator fails closed" primary git.example
write_bad_key_fixture ',bad'
assert_both_fail_closed "leading-comma flow indicator fails closed" primary git.example
# Bare block indicators in a value position ('-'/'- ', '?'/'? ', ':'/': '):
# PyYAML raises a scanner error on the whole document, so the fallback must fail
# closed rather than accept the indicator as a plain-scalar string.
write_bad_key_fixture '-'
assert_both_fail_closed "bare dash (seq indicator) fails closed" primary git.example
write_bad_key_fixture '- x'
assert_both_fail_closed "dash-space (seq entry) fails closed" primary git.example
write_bad_key_fixture '? key'
assert_both_fail_closed "question-space (complex key) fails closed" primary git.example
# ...but an indicator NOT followed by whitespace is a valid plain scalar string,
# so the token still resolves (no over-broad fail-close).
write_bad_key_fixture '-x'
assert_token "dash-not-space is a plain string, token resolves" "TOK_PLAIN" primary git.example
write_bad_key_fixture ':x'
assert_token "colon-not-space is a plain string, token resolves" "TOK_PLAIN" primary git.example
# NOT over-broad: a genuinely CONSTRUCTIBLE typed scalar (or a look-alike PyYAML
# keeps as a plain string) leaves the document valid, so BOTH still resolve the
# login token -- the fix must not fail closed on these.
write_bad_key_fixture '2023-01-15'
assert_token "valid date unrelated key still resolves token" "TOK_PLAIN" primary git.example
write_bad_key_fixture '2023-01-15 10:00:00'
assert_token "valid datetime unrelated key still resolves token" "TOK_PLAIN" primary git.example
# '0o_' is NOT matched by PyYAML's int resolver (YAML 1.1 octal is 0[0-7]+, not
# 0o...), so PyYAML keeps it a STRING and resolves the token; the fallback must
# agree (no spurious fail-close).
write_bad_key_fixture '0o_'
assert_token "0o_ is a plain string in PyYAML, token still resolves" "TOK_PLAIN" primary git.example
# '4.e8' matches the fallback's (superset) float pattern but PyYAML keeps it a
# string; either way it is constructible, so the token still resolves in both.
write_bad_key_fixture '4.e8'
assert_token "4.e8 float look-alike still resolves token" "TOK_PLAIN" primary git.example
# A valid radix int as an unrelated key must not fail closed.
write_bad_key_fixture '0x1f'
assert_token "valid hex int unrelated key still resolves token" "TOK_PLAIN" primary git.example
# 17. TAB / SCANNER PARITY: PyYAML raises a ScannerError on a tab used anywhere
# outside a quoted scalar -- leading, trailing, or embedded in a plain value,
# immediately after a key colon, before a key colon, or as indentation -- and
# yields NO token, accepting tabs ONLY inside single/double-quoted scalars
# (where the tab is preserved as string content). A prior fallback swallowed
# those tabs (via .strip()/.rstrip() normalization and [ \t] key separators)
# and still emitted the login token -- a fail-open in the dangerous direction.
# The recognizer now fails CLOSED for the whole document on any tab PyYAML
# rejects, while preserving the tabs PyYAML keeps (inside quotes). All tab
# positions were verified empirically against PyYAML 6.0.3 (ScannerError for
# each rejected position; string-preserved for quoted inner tabs).
TAB=$'\t'
# Fail-close: a tab in a plain value position (trailing / leading / embedded).
write_bad_key_fixture "l4o${TAB}"
assert_both_fail_closed "trailing tab in plain value fails closed" primary git.example
write_bad_key_fixture "${TAB}9"
assert_both_fail_closed "leading tab in plain value fails closed" primary git.example
write_bad_key_fixture "a${TAB}b"
assert_both_fail_closed "embedded tab in plain value fails closed" primary git.example
# Fail-close: a tab immediately after the key colon (no separating space).
write_fixture "bad:${TAB}9
logins:
- name: primary
url: https://git.example
token: TOK_PLAIN
"
assert_both_fail_closed "tab immediately after key colon fails closed" primary git.example
# Fail-close: a tab used as indentation (before a sequence dash).
write_fixture "logins:
${TAB}- name: primary
url: https://git.example
token: TOK_PLAIN
"
assert_both_fail_closed "tab used as indentation fails closed" primary git.example
# Fail-close: a tab trailing a sequence-mapping field value.
write_fixture "logins:
- name: primary
url: https://git.example
token: TOK_PLAIN${TAB}
"
assert_both_fail_closed "tab trailing a seq field value fails closed" primary git.example
# NOT over-broad: a tab strictly INSIDE a quoted scalar is valid YAML (PyYAML
# keeps it as string content), so the document parses and the login token still
# resolves in BOTH paths -- double-quoted and single-quoted.
write_bad_key_fixture "\"a${TAB}b\""
assert_token "tab inside a double-quoted value still resolves token" "TOK_PLAIN" primary git.example
write_bad_key_fixture "'a${TAB}b'"
assert_token "tab inside a single-quoted value still resolves token" "TOK_PLAIN" primary git.example
# ...and a quoted token value carrying an inner tab resolves to the exact string
# (tab preserved), identical to PyYAML's construction.
write_fixture "logins:
- name: primary
url: https://git.example
token: \"T${TAB}OK\"
"
assert_token "quoted token with inner tab resolves verbatim" "T${TAB}OK" primary git.example
# 18. CONTROL-CHARACTER FAIL-CLOSE (#865 round-9 blocker 1): PyYAML's Reader
# scans the ENTIRE raw document stream (not merely scalar contents, and NOT
# scoped by quoting) for bytes outside its printable set and raises
# ReaderError -- a WHOLE-DOCUMENT reject -- the instant one is found,
# regardless of where it sits: an unrelated field's plain scalar, inside a
# double- or single-quoted scalar, or a comment. Verified empirically against
# real installed PyYAML 6.0.3 (see detect-platform.sh's _FORBIDDEN_CONTROL
# comment): every C0 control byte {0x00-0x08, 0x0B, 0x0C, 0x0E-0x1F} plus DEL
# (0x7F) rejects in ALL THREE contexts (plain / double-quoted / single-quoted);
# only TAB(0x09), LF(0x0A), CR(0x0D) are accepted among the low byte range
# (TAB has its own narrower, position-aware coverage in section 17 above; LF/CR
# are line separators). A prior fallback ONLY guarded tabs and emitted the
# login token from documents PyYAML rejects over an UNRELATED field's control
# byte -- a dangerous fail-open (credential emission from a document PyYAML
# refuses). write_control_char_fixture writes the raw byte directly via
# printf's octal escape (never through a bash string/variable, which cannot
# hold an embedded NUL) so 0x00 is exercised faithfully alongside the rest.
write_control_char_fixture() {
local octal="$1" quote="${2:-}"
{
if [[ -n "$quote" ]]; then
printf 'bad: %sx' "$quote"
# shellcheck disable=SC2059 # deliberate: $octal supplies printf's
# own \NNN octal escape so the raw control byte reaches the file
# directly, never passing through a bash string (which truncates
# at an embedded NUL and so cannot represent byte 0x00 otherwise).
printf "\\${octal}"
printf 'y%s\n' "$quote"
else
printf 'bad: x'
# shellcheck disable=SC2059 # deliberate: $octal supplies printf's
# own \NNN octal escape so the raw control byte reaches the file
# directly, never passing through a bash string (which truncates
# at an embedded NUL and so cannot represent byte 0x00 otherwise).
printf "\\${octal}"
printf 'y\n'
fi
printf 'logins:\n - name: primary\n url: https://git.example\n token: TOK_PLAIN\n'
} > "$FIXTURE_XDG/tea/config.yml"
}
# Plain (unquoted) unrelated-field placement: the full empirically-confirmed
# forbidden C0/DEL set.
for octal in 000 001 002 003 004 005 006 007 010 013 014 \
016 017 020 021 022 023 024 025 026 027 \
030 031 032 033 034 035 036 037 177; do
write_control_char_fixture "$octal"
assert_both_fail_closed "control byte \\$octal in unrelated plain field fails closed" primary git.example
done
# Inside-quote variants (double and single) for the six bytes called out
# explicitly in the round-9 blocker report: 0x00,0x01,0x07,0x0e,0x1f,0x7f.
for octal in 000 001 007 016 037 177; do
write_control_char_fixture "$octal" '"'
assert_both_fail_closed "control byte \\$octal inside double-quoted unrelated field fails closed" primary git.example
write_control_char_fixture "$octal" "'"
assert_both_fail_closed "control byte \\$octal inside single-quoted unrelated field fails closed" primary git.example
done
# Same forbidden byte inside a comment line -- PyYAML's Reader check is
# stream-wide, so it rejects here too, not merely inside live scalar content.
{
printf '# note'
printf '\007'
printf 'here\nlogins:\n - name: primary\n url: https://git.example\n token: TOK_PLAIN\n'
} > "$FIXTURE_XDG/tea/config.yml"
assert_both_fail_closed "control byte in a comment line fails closed" primary git.example
# NOT over-broad: TAB/LF/CR remain accepted where PyYAML already accepts them
# (covered by section 17's tab fixtures and the ordinary newline-delimited
# fixtures used throughout this file), so no additional assertion is needed
# here beyond confirming the forbidden-control guard does not fire on them.
# 19. UNSIGNED-EXPONENT FLOAT OVER-REJECTION (#865 round-9 blocker 2): PyYAML
# 6.0.3's implicit float resolver requires an EXPLICIT SIGN on the exponent
# ([eE][-+][0-9]+); an unsigned exponent is NOT matched, so PyYAML resolves
# the scalar as a plain STRING, not a float. A prior fallback's float
# recognizer accepted an OPTIONAL sign ([eE][-+]?[0-9]+), over-matching these
# spellings as floats and dropping the token PyYAML would emit verbatim
# (over-rejection). Verified empirically against real PyYAML 6.0.3.
for form in '1.0e10' '+1.0e10' '-1.0e10' '1.0E10' '.5e10' '4.e8'; do
write_fixture "logins:
- name: primary
url: https://git.example
token: ${form}
"
assert_token "unsigned-exponent form '$form' is a PyYAML string, token resolves" "$form" primary git.example
done
# Parity guard: a genuine SIGNED-exponent float is still typed as a non-string
# float by PyYAML and must still fail closed (not regress into over-acceptance).
write_fixture 'logins:
- name: primary
url: https://git.example
token: 1.0e+10
'
assert_token "signed-exponent genuine float still fails closed" "" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: 1.0e-10
'
assert_token "signed-exponent (negative) genuine float still fails closed" "" primary git.example
# 20. RESIDUAL OVER-REJECTION found via round-9 differential fuzzing (folded into
# this round, not split off): a plain scalar starting with "?" NOT followed by
# whitespace (e.g. "?x") is a valid PyYAML string -- only a bare "?" or "? "
# (question mark followed by space/EOL) opens a complex mapping key and is
# illegal in a value position. A prior blanket-reject set treated EVERY
# leading "?" as illegal, over-rejecting a token PyYAML accepts verbatim.
write_fixture 'logins:
- name: primary
url: https://git.example
token: ?x
'
assert_token "question-mark-not-space is a plain string, token resolves" "?x" primary git.example
# 21. PRINTABLE-BOUNDARY FAIL-CLOSE (#865 round-10 blocker 1): the round-9 guard
# used a hand-rolled C0/DEL subset that MISSED code points PyYAML's Reader
# also rejects -- the C1 block (U+0080-0084, U+0086-009F) and the BMP
# noncharacters U+FFFE/U+FFFF -- so the fallback still emitted the token from
# documents PyYAML rejects whole (fail-open). The guard now uses PyYAML
# 6.0.3's EXACT Reader.NON_PRINTABLE character class (see detect-platform.sh
# _FORBIDDEN_CONTROL). Verified empirically against real PyYAML 6.0.3:
# PRINTABLE = {0x09,0x0A,0x0D, 0x20-0x7E, 0x85(NEL), 0xA0-0xD7FF,
# 0xE000-0xFFFD, 0x10000-0x10FFFF}; everything else fails the whole document
# closed. write_codepoint_fixture emits a chosen Unicode code point's real
# UTF-8 bytes (via python3, since bash strings cannot faithfully carry many
# of these) into a selectable position, then the login block follows.
write_codepoint_fixture() {
# $1 = hex code point (e.g. 0x80); $2 = position: field|comment|token|nelterm
CP_HEX="$1" CP_POS="$2" python3 - "$FIXTURE_XDG/tea/config.yml" <<'PY'
import sys
cp = int(__import__("os").environ["CP_HEX"], 16)
pos = __import__("os").environ["CP_POS"]
ch = chr(cp)
head = "logins:\n - name: primary\n url: https://git.example\n token: TOK_PLAIN\n"
if pos == "field":
doc = head + "other: x" + ch + "y\n"
elif pos == "comment":
doc = head + "# note x" + ch + "y here\n"
elif pos == "token":
doc = "logins:\n - name: primary\n url: https://git.example\n token: T" + ch + "K\n"
elif pos == "nelterm":
# NEL (U+0085) used as the line terminator throughout: PyYAML treats it as a
# line break (printable, NOT a ReaderError) and resolves the token; the
# fallback's _split_logical_lines splits on NEL identically OUTSIDE a quote
# -> parity, token resolves. (Round 23 covers NEL/LS/PS INSIDE a quote, where
# PyYAML folds rather than breaks and a naive splitlines() would over-split.)
doc = ("logins:" + ch + " - name: primary" + ch
+ " url: https://git.example" + ch + " token: TOK_NEL" + ch)
else:
raise SystemExit("bad pos")
with open(sys.argv[1], "w", encoding="utf-8") as f:
f.write(doc)
PY
}
# C1-block + BMP-noncharacter code points fail the WHOLE document closed in an
# unrelated field and in a comment, exactly as PyYAML's ReaderError does.
for cphex in 0x80 0x81 0x84 0x86 0x9f 0xfffe 0xffff; do
write_codepoint_fixture "$cphex" field
assert_both_fail_closed "code point $cphex in unrelated field fails closed" primary git.example
write_codepoint_fixture "$cphex" comment
assert_both_fail_closed "code point $cphex in a comment fails closed" primary git.example
done
# NOT over-broad: printable code points PyYAML ACCEPTS must still resolve the
# token in BOTH paths -- NEL(0x85) as a line separator, U+00A0 (NBSP) inside a
# value, and an astral code point (U+1F600) inside the token value.
write_codepoint_fixture 0x85 nelterm
assert_token "NEL (U+0085) line-terminator resolves token" "TOK_NEL" primary git.example
write_codepoint_fixture 0xa0 field
assert_token "U+00A0 in unrelated value still resolves token" "TOK_PLAIN" primary git.example
write_codepoint_fixture 0x1f600 token
assert_token "astral U+1F600 inside token resolves verbatim" "$(printf 'T\360\237\230\200K')" primary git.example
# 22. INTERNAL-INDICATOR OVER-REJECTION (#865 round-10 blocker 2): the round-9
# recognizer blanket-rejected any plain scalar CONTAINING a flow indicator
# ([]{}*&!), but in BLOCK context PyYAML treats ',[]{}' as ordinary content
# and treats '!&*#...' as significant ONLY at the FIRST non-space char. So an
# INTERNAL indicator is legal plain-string content and PyYAML emits the token
# verbatim; the fallback dropped it (over-rejection). Verified empirically
# against real PyYAML 6.0.3. The fix removes the blanket internal scan while
# the leading-char guard and the ' #'/': '/trailing-':' guards keep the
# fail-OPEN direction shut.
for tv in 'a!b' 'a,b' 'a[b' 'a]b' 'a{b' 'a}b' 'a&b' 'a*b' 'a[b]c' 'a{b}c' 'a,b,c' 'a#b' 'a:b'; do
write_fixture "logins:
- name: primary
url: https://git.example
token: ${tv}
"
assert_token "internal-indicator token '$tv' resolves verbatim" "$tv" primary git.example
done
# Fail-OPEN direction stays shut: a LEADING indicator, a ' #' comment tail, an
# internal ': ' (colon-space) inline map, and a trailing ':' each make PyYAML
# resolve NO usable string token (tag/flow/anchor reject or None, comment strip,
# or a mapping), so BOTH must fail closed. (Leading tag/anchor/flow are the
# documented, round-9-approved structural fail-closed class; kept intact here.)
write_fixture 'logins:
- name: primary
url: https://git.example
token: !x
'
assert_both_fail_closed "leading '!' tag token fails closed" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: [a]
'
assert_both_fail_closed "leading '[' flow-seq token fails closed" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: a # trailing comment
'
# ' #' comment tail: PyYAML strips the comment -> token is the string 'a', which
# still resolves. This is the NOT-over-broad boundary partner of the guard.
assert_token "space-hash comment tail strips to plain token" "a" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: a: b
'
assert_both_fail_closed "internal colon-space (inline map) token fails closed" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: ab:
'
assert_both_fail_closed "trailing colon (map indicator) token fails closed" primary git.example
# 23. EMBEDDED LINE-BREAK INSIDE A QUOTED SCALAR (#865 round-11 blocker A): the
# round-10 fallback split the raw document with str.splitlines(), which breaks
# at NEL(U+0085), LS(U+2028) and PS(U+2029) -- code points that are PRINTABLE
# to PyYAML's Reader. Inside a flow (quoted) scalar PyYAML does NOT break at
# these: it LINE-FOLDS a double/single-quoted scalar (NEL/LF/CR -> a single
# space; LS/PS -> the char verbatim), so it resolves ONE token, while
# splitlines() cut the value mid-quote and failed the whole document closed
# (over-rejection). The fallback now uses _split_logical_lines, which
# reproduces PyYAML's flow-folding. Verified empirically vs real PyYAML 6.0.3.
# write_break_fixture emits a chosen break code point in a selectable context.
write_break_fixture() {
# $1 = hex code point of the break; $2 = context: dq|sq|plain|comment|dq2
CP_HEX="$1" Q_STYLE="$2" python3 - "$FIXTURE_XDG/tea/config.yml" <<'PY'
import sys, os
cp = int(os.environ["CP_HEX"], 16)
q = os.environ["Q_STYLE"]
ch = chr(cp)
head = "logins:\n - name: primary\n url: https://git.example\n token: "
if q == "dq":
doc = head + '"tok' + ch + 'en"'
elif q == "sq":
doc = head + "'tok" + ch + "en'"
elif q == "plain":
doc = head + "tok" + ch + "en"
elif q == "dq2":
# blank line inside a quoted scalar: PyYAML folds a two-break run to a literal
# newline, which the recognizer's key regex cannot carry -> endorsed
# fail-closed over-reject (see assert_fallback_fails_closed below).
doc = head + '"tok' + ch + ch + 'en"'
elif q == "comment":
doc = ("logins:\n - name: primary\n url: https://git.example\n"
" token: TOK_PLAIN\n# c" + ch + "x")
else:
raise SystemExit("bad q")
with open(sys.argv[1], "w", encoding="utf-8") as f:
f.write(doc + "\n")
PY
}
# NEL folds to a single space inside double- AND single-quoted scalars: the token
# resolves identically in both paths (fallback no longer over-splits).
write_break_fixture 0x85 dq
assert_token "NEL inside double-quote folds to space, token resolves" "tok en" primary git.example
write_break_fixture 0x85 sq
assert_token "NEL inside single-quote folds to space, token resolves" "tok en" primary git.example
# LS(U+2028)/PS(U+2029) are preserved VERBATIM by PyYAML's flow fold (they are
# not \n-class breaks); the fallback must surface them byte-for-byte.
write_break_fixture 0x2028 dq
assert_token "LS inside double-quote is verbatim" "$(printf 'tok\342\200\250en')" primary git.example
write_break_fixture 0x2028 sq
assert_token "LS inside single-quote is verbatim" "$(printf 'tok\342\200\250en')" primary git.example
write_break_fixture 0x2029 dq
assert_token "PS inside double-quote is verbatim" "$(printf 'tok\342\200\251en')" primary git.example
# Direction-sensitivity: the SAME code points UNQUOTED (a plain scalar) or in a
# COMMENT make PyYAML raise a scanner error, so both paths must fail closed. The
# fold rule applies ONLY inside a quoted scalar.
write_break_fixture 0x85 plain
assert_token "NEL in an unquoted plain scalar fails closed" "" primary git.example
write_break_fixture 0x2028 plain
assert_token "LS in an unquoted plain scalar fails closed" "" primary git.example
write_break_fixture 0x85 comment
assert_token "NEL in a comment fails closed" "" primary git.example
# Endorsed fail-closed over-reject: a blank line inside a quoted scalar folds to a
# literal newline that the recognizer cannot carry -- PyYAML resolves a
# (newline-bearing) token, the fallback fails closed (strictly safer).
write_break_fixture 0x85 dq2
assert_fallback_fails_closed "blank-line-in-quote (NEL run) fails closed" primary git.example
# 24. LEADING NON-SPECIFIC TAG / ANCHOR PROPERTY (#865 round-11 blocker B, revised
# in round 12): the round-10 recognizer blanket-rejected any scalar beginning
# with '!' or '&'. Round 11 taught it to strip a transparent NON-SPECIFIC tag
# ('! ' bang + SPACE) and a transparent plain ANCHOR ('&name ') so '! x' /
# '&a x' resolve the STRING 'x', matching PyYAML. Round 12 discovered that the
# anchor half of that was a HIGH fail-open: PyYAML's Composer tracks anchor
# NAMES in a document-scoped registry and raises ComposerError ("found
# duplicate anchor") the instant the SAME name is declared on a SECOND node
# ANYWHERE in the document (even an unrelated one) -- the fallback's
# per-scalar-only view has no such registry and would emit the later token.
# Round 12's fix: reject EVERY '&'-anchor property, unconditionally. The
# transparent NON-SPECIFIC TAG behavior ('! x' -> 'x') is unchanged and still
# verified below; only the anchor half now fails closed (deliberate
# conservative over-reject, verified safe both ways against real PyYAML 6.0.3).
for pair in '! x=x' '! x y=x y' '! "q"=q' "! 'q'=q"; do
tv="${pair%%=*}"; want="${pair#*=}"
write_fixture "logins:
- name: primary
url: https://git.example
token: ${tv}
"
assert_token "tag property token '$tv' resolves node string" "$want" primary git.example
done
# Fail-OPEN direction stays shut. '!x' (bang + NON-space) is a tag HANDLE ->
# ConstructorError; '!foo x'/'* a'/'! !x' raise; a property over a NON-string node
# ('! 123'/'! true'/'! null') types non-str -> no usable token. All fail closed in
# BOTH paths.
for tv in '!x' '!foo x' '* a' '! !x' '! 123' '! true' '! null' '&a &b x'; do
write_fixture "logins:
- name: primary
url: https://git.example
token: ${tv}
"
assert_token "non-resolving property token '$tv' fails closed" "" primary git.example
done
# Round 12: a SINGLE, non-duplicated '&a x' is valid YAML that real PyYAML
# resolves to the string 'x' (round-11 behavior, and still true of the oracle).
# The fallback now rejects it anyway -- a deliberate, endorsed CONSERVATIVE
# over-reject (see the round-12 comment block above): fail-closed can only cost
# an emitted token PyYAML would have allowed, never emit one PyYAML rejects, and
# a per-scalar recognizer cannot safely prove document-wide anchor-name
# uniqueness. assert_fallback_fails_closed also confirms PyYAML really does
# resolve a token here, proving this is a genuine (safe-direction) divergence
# and not an accidental parity loss.
write_fixture 'logins:
- name: primary
url: https://git.example
token: &a x
'
assert_fallback_fails_closed "round-12: single non-duplicated anchor '&a x' now fails closed (conservative over-reject; PyYAML resolves x)" primary git.example
# Same over-reject for the combined tag+anchor forms round 11 used to resolve.
write_fixture 'logins:
- name: primary
url: https://git.example
token: ! &b x
'
assert_fallback_fails_closed "round-12: '! &b x' (tag+anchor) now fails closed (conservative over-reject)" primary git.example
write_fixture 'logins:
- name: primary
url: https://git.example
token: &b ! x
'
assert_fallback_fails_closed "round-12: '&b ! x' (anchor+tag) now fails closed (conservative over-reject)" primary git.example
# Endorsed fail-closed over-reject: an EXPLICIT tag ('!!str x', verbose
# '!<tag:yaml.org,2002:str> x') forces a string PyYAML resolves, but the fallback
# recognizes only the transparent non-specific tag and fails closed (safer).
for tv in '!!str x' '!<tag:yaml.org,2002:str> x'; do
write_fixture "logins:
- name: primary
url: https://git.example
token: ${tv}
"
assert_fallback_fails_closed "explicit-tag token '$tv' fails closed" primary git.example
done
# 25. #865 round 12 HIGH fail-open closure: DUPLICATE ANCHOR NAME across separate
# nodes. Real PyYAML's Composer tracks anchor names in a DOCUMENT-SCOPED
# registry and raises ComposerError ("found duplicate anchor ... first
# occurrence") the instant the SAME anchor name is declared a second time
# ANYWHERE in the document -- failing the WHOLE document closed, no token,
# regardless of how far the duplicate sits from the logins block. The round-11
# fallback tracked anchors only WITHIN a single scalar's `_strip_properties`
# call, so it had no visibility into a duplicate declared on an unrelated
# node and would still emit the (later) token: fail-open. The round-12 fix
# (reject every '&'-anchor property, unconditionally -- see section 24 above)
# closes this as a strict superset: since NO anchor is ever accepted, a
# duplicate anchor can never slip through. These cases exercise that
# document-wide duplicate-anchor invariant specifically (as opposed to
# section 24's single-anchor-on-the-token-field cases) and pair each fallback
# assertion with confirmation that real PyYAML also fails closed here (via
# ComposerError), proving this was a genuine fail-open, not a hypothetical.
write_fixture 'first: &same one
second: &same two
logins:
- name: primary
url: https://git.example
token: TOK_DUP_ROOT
'
assert_both_fail_closed "round-12: duplicate anchor name on two unrelated root nodes fails closed" primary git.example
write_fixture 'outer:
nested: &dup x
dup_root: &dup y
logins:
- name: primary
url: https://git.example
token: TOK_DUP_NESTED
'
assert_both_fail_closed "round-12: duplicate anchor name across a nested node and a root node fails closed" primary git.example
write_fixture 'first: &dup one
logins:
- name: primary
url: https://git.example
token: &dup TOK_DUP_TOKEN_NODE
'
assert_both_fail_closed "round-12: anchor name declared earlier and repeated on the token-bearing node fails closed" primary git.example
# Regression guard: the non-specific TAG half of section 24 ('! x' -> 'x') is
# UNCHANGED by the round-12 anchor fix and must still resolve.
write_fixture 'logins:
- name: primary
url: https://git.example
token: ! x
'
assert_token "round-12 regression: '! x' (tag, no anchor) still resolves 'x'" "x" primary git.example
echo "Gitea login resolution regression harness passed" echo "Gitea login resolution regression harness passed"

View File

@@ -0,0 +1,122 @@
#!/usr/bin/env bash
# Regression harness for detect-platform.sh's get_gitea_token() per-agent
# identity resolution (Gate-16 author≠reviewer separation) — the API-tooling
# counterpart to git-credential-mosaic, so pr-create.sh/issue-create.sh/etc.
# open records under the resolved agent identity, not the shared account.
#
# Covers:
# 1. Identity resolution priority: MOSAIC_GIT_IDENTITY env > git config
# mosaic.gitIdentity (per-worktree).
# 2. Correct per-slot token file path chosen per host
# (gitea-usc-<id>.token vs gitea-mosaicstack-<id>.token).
# 3. Per-slot token present -> that token is returned (agent-authored calls).
# 4. Per-slot token absent -> falls back to the shared credential-loader
# token (backward-compat / no-op for hosts without per-slot tokens).
# 5. Unrelated host with no shared credentials configured -> failure
# (unchanged, existing behavior).
#
# Uses a stubbed credentials.json + stubbed per-slot token files under a fake
# HOME. NEVER reads real secrets or touches the real ~/.config/mosaic/secrets.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
WORK_DIR="${MOSAIC_TEST_WORK_DIR:-$PWD/.mosaic-test-work/gitea-token-identity}"
FAKE_HOME="$WORK_DIR/home"
REPO_DIR="$WORK_DIR/repo"
CREDENTIALS_FILE="$FAKE_HOME/.config/mosaic/credentials.json"
rm -rf "$WORK_DIR"
mkdir -p "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens" "$REPO_DIR"
git -C "$REPO_DIR" init -q
git -C "$REPO_DIR" remote add origin https://git.mosaicstack.dev/mosaicstack/stack.git
cat > "$CREDENTIALS_FILE" <<'JSON'
{
"gitea": {
"mosaicstack": {
"url": "https://git.mosaicstack.dev",
"token": "shared-mosaicstack-token"
},
"usc": {
"url": "https://git.uscllc.com",
"token": "shared-usc-token"
}
}
}
JSON
fail=0
assert_eq() {
local desc="$1" expected="$2" actual="$3"
if [[ "$expected" != "$actual" ]]; then
echo "FAIL: $desc — expected '$expected', got '$actual'" >&2
fail=1
fi
}
# Runs get_gitea_token for $1=host inside REPO_DIR (per-worktree git config
# resolves there) with a fake HOME + the stub credentials.json, plus any
# extra env passed in $@.
call_get_gitea_token() {
local host="$1"; shift
(
cd "$REPO_DIR"
# shellcheck disable=SC2016 # deliberately deferred: $DETECT_PLATFORM_SH is
# expanded by the INNER bash -c (via the exported env var below), not here.
env -i HOME="$FAKE_HOME" PATH="$PATH" MOSAIC_CREDENTIALS_FILE="$CREDENTIALS_FILE" \
DETECT_PLATFORM_SH="$SCRIPT_DIR/detect-platform.sh" "$@" \
bash -c 'source "$DETECT_PLATFORM_SH"; get_gitea_token "$1"' _ "$host"
)
}
# ---------------------------------------------------------------------------
# 1. No identity resolvable -> shared credential-loader token (unchanged).
# ---------------------------------------------------------------------------
git -C "$REPO_DIR" config --unset mosaic.gitIdentity 2>/dev/null || true
out=$(call_get_gitea_token "git.mosaicstack.dev")
assert_eq "shared fallback (no identity)" "shared-mosaicstack-token" "$out"
# ---------------------------------------------------------------------------
# 2. git config mosaic.gitIdentity resolves to an agent WITH a per-slot
# token -> that token wins over the shared account.
# ---------------------------------------------------------------------------
echo -n "agentA-mosaicstack-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-mosaicstack-agentA.token"
git -C "$REPO_DIR" config mosaic.gitIdentity agentA
out=$(call_get_gitea_token "git.mosaicstack.dev")
assert_eq "git-config identity token" "agentA-mosaicstack-token" "$out"
# ---------------------------------------------------------------------------
# 3. MOSAIC_GIT_IDENTITY env beats git config mosaic.gitIdentity.
# ---------------------------------------------------------------------------
echo -n "agentB-mosaicstack-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-mosaicstack-agentB.token"
out=$(call_get_gitea_token "git.mosaicstack.dev" MOSAIC_GIT_IDENTITY=agentB)
assert_eq "env beats git-config identity token" "agentB-mosaicstack-token" "$out"
# ---------------------------------------------------------------------------
# 4. Identity resolves but has no per-slot token for THIS host -> falls back
# to the shared token (per-agent identity is opt-in per host).
# ---------------------------------------------------------------------------
git -C "$REPO_DIR" config mosaic.gitIdentity no-such-agent
out=$(call_get_gitea_token "git.mosaicstack.dev")
assert_eq "no per-slot token falls back to shared" "shared-mosaicstack-token" "$out"
# ---------------------------------------------------------------------------
# 5. Correct per-slot token PATH per host: same agent id, only a usc token
# exists -> usc host returns it, mosaicstack host must NOT leak it and
# instead falls back to the shared mosaicstack token.
# ---------------------------------------------------------------------------
echo -n "agentD-usc-token" > "$FAKE_HOME/.config/mosaic/secrets/gitea-tokens/gitea-usc-agentD.token"
git -C "$REPO_DIR" config mosaic.gitIdentity agentD
out=$(call_get_gitea_token "git.uscllc.com")
assert_eq "host-scoped token path (usc)" "agentD-usc-token" "$out"
out=$(call_get_gitea_token "git.mosaicstack.dev")
assert_eq "host-scoped token path (no cross-host leak)" "shared-mosaicstack-token" "$out"
git -C "$REPO_DIR" config --unset mosaic.gitIdentity
if [[ "$fail" -eq 0 ]]; then
echo "get_gitea_token identity resolution regression passed"
fi
exit "$fail"

View File

@@ -0,0 +1,571 @@
#!/usr/bin/env bash
# Regression harness for issue-comment.sh's Gitea comment write + verification
# (#865).
#
# The #865 defect class: tea 0.11.1's `tea issue comment ...` (a nonexistent
# subcommand) silently no-ops yet exits 0, and tea cannot emit the id of a
# record it created — so an exit code is worthless as proof of a durable write.
# The wrapper therefore does NOT write via tea at all. It POSTs the comment to
# the Gitea REST API (which returns the created comment object, including its
# id), then GETs THAT EXACT id back and requires it to match on id, author
# (acting identity), body, and issue. Because verification is keyed to the id
# the create returned, no concurrent comment can masquerade as this write, and a
# suppressed/no-op create yields no id and fails closed.
#
# This harness models a REAL server: the curl stub keeps persistent comment
# state on disk, the POST actually CREATES and PERSISTS a record and returns its
# id, and the read-back GET reads that same state. There is no independently
# fabricated record for the wrapper to "find" — the only way verification
# passes is if the POST genuinely created the record the read-back retrieves.
# It proves the wrapper:
# 1. never shells out to tea to write (no `tea comment` / `tea issue comment`);
# 2. creates the comment via REST POST and learns the provider-returned id;
# 3. verifies THAT EXACT id by direct GET, attributed to the acting identity;
# 4. fails closed when the write is a no-op even though a concurrent
# SAME-IDENTITY comment with the same body already exists (the closed
# concurrency window — no fallback list scan can rescue a no-op);
# 5. fails closed when the created record is not authored by the acting
# identity;
# 6. treats the exact-id GET as the SOLE authority — it performs NO follow-up
# list enumeration (the stub exposes no comment-list endpoint, so any
# residual enumeration attempt would fail the run);
# 7. with a RESOLVABLE --login override, performs the write, the /user identity
# lookup, and the read-back ALL under THAT login's token/identity — never
# the host default;
# 8. with an UNRESOLVABLE --login override, FAILS CLOSED (nonzero, no write, no
# success line) instead of silently downgrading to the host default
# identity — the token seam maps each bearer token to the identity it
# authenticates as, so a misattributed write is caught;
# 9. with a --login override whose tea config URL is a DIFFERENT host than the
# repo remote, FAILS CLOSED (host-bound token selection) rather than sending
# that other host's credential cross-host;
# 10. leaves NO temp files behind (POST/GET bodies + metadata) on either the
# success or the failure path — nested function-scoped RETURN traps do not
# clobber each other and every scratch file is removed on all exit paths.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
WORK_DIR="${MOSAIC_TEST_WORK_DIR:-$PWD/.mosaic-test-work/issue-comment-readback}"
REPO_DIR="$WORK_DIR/repo"
BIN_DIR="$WORK_DIR/bin"
XDG_DIR="$WORK_DIR/xdg"
TEA_LOG="$WORK_DIR/tea.log"
CURL_LOG="$WORK_DIR/curl.log"
# Full curl argv per invocation — proves the bearer token never rides in argv.
CURL_ARGV_LOG="$WORK_DIR/curl-argv.log"
AUTH_LOG="$WORK_DIR/auth.log"
OUTPUT_FILE="$WORK_DIR/output.log"
CREDENTIALS_FILE="$WORK_DIR/credentials.json"
STATE_FILE="$WORK_DIR/comments.json"
# A dedicated scratch dir the wrapper is pointed at via TMPDIR, so the leak
# check can assert every POST/GET body + metadata temp file is cleaned up.
TMP_SCRATCH="$WORK_DIR/scratch"
cleanup() {
rm -rf "$WORK_DIR"
}
trap cleanup EXIT
mkdir -p "$REPO_DIR" "$BIN_DIR" "$XDG_DIR" "$TMP_SCRATCH"
git -C "$REPO_DIR" init -q
git -C "$REPO_DIR" remote add origin https://git.mosaicstack.dev/mosaicstack/stack.git
ISSUE_NUMBER=7
REPO_SLUG="mosaicstack/stack"
API_BASE="https://git.mosaicstack.dev/api/v1/repos/mosaicstack/stack"
API_ROOT="https://git.mosaicstack.dev/api/v1"
BODY='durable "note" -- marker'
ACTING_LOGIN="primary-reviewer"
FOREIGN_LOGIN="other-writer"
# A dedicated per-role --login override identity, with its own token stored in
# tea's config (exactly the author-not-equal-reviewer hardening path).
OVERRIDE_LOGIN="delegated-reviewer"
DEFAULT_TOKEN="test-only-placeholder"
OVERRIDE_TOKEN="override-token-placeholder"
# A --login override whose tea config URL points at a DIFFERENT Gitea host than
# the repo remote (git.mosaicstack.dev). Its token must NEVER be sent to the
# repo host: host-bound selection must fail closed on the host mismatch.
CROSS_HOST_LOGIN="foreign-host-reviewer"
CROSS_HOST_TOKEN="cross-host-token-placeholder"
# tea config: the override login has its own token here (as tea itself stores
# per-login tokens). The default login name ("mosaicstack") is deliberately NOT
# present, so the no-override default path resolves via the host credential
# fallback while an explicit --login must resolve from this file or fail closed.
# A second login is configured for a DIFFERENT host to exercise host-bound
# rejection.
mkdir -p "$XDG_DIR/tea"
OVERRIDE_LOGIN="$OVERRIDE_LOGIN" OVERRIDE_TOKEN="$OVERRIDE_TOKEN" \
CROSS_HOST_LOGIN="$CROSS_HOST_LOGIN" CROSS_HOST_TOKEN="$CROSS_HOST_TOKEN" \
python3 - "$XDG_DIR/tea/config.yml" <<'PY'
import os
import sys
with open(sys.argv[1], "w", encoding="utf-8") as handle:
handle.write("logins:\n")
handle.write(f" - name: {os.environ['OVERRIDE_LOGIN']}\n")
handle.write(" url: https://git.mosaicstack.dev\n")
handle.write(f" token: {os.environ['OVERRIDE_TOKEN']}\n")
handle.write(f" - name: {os.environ['CROSS_HOST_LOGIN']}\n")
handle.write(" url: https://git.uscllc.com\n")
handle.write(f" token: {os.environ['CROSS_HOST_TOKEN']}\n")
PY
CONFIGURED_GITEA_URL="https://git.mosaicstack.dev" python3 - "$CREDENTIALS_FILE" <<'PY'
import json
import os
import sys
with open(sys.argv[1], "w", encoding="utf-8") as credentials:
json.dump({
"gitea": {
"mosaicstack": {
"url": os.environ["CONFIGURED_GITEA_URL"],
"token": "test-only-placeholder",
}
}
}, credentials)
PY
# tea stub: only ever answers the login list (used to resolve the default login
# name). It must NEVER be asked to write a comment — the wrapper writes via REST.
cat > "$BIN_DIR/tea" <<'SH'
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' "$*" >> "$ISSUE_COMMENT_TEA_LOG"
if [[ "$*" == "login list --output json" ]]; then
printf '%s\n' '[{"name":"mosaicstack","url":"https://git.mosaicstack.dev"}]'
exit 0
fi
echo "Unexpected tea command (wrapper must not write via tea): $*" >&2
exit 92
SH
chmod +x "$BIN_DIR/tea"
# curl stub: a small REST server backed by persistent on-disk comment state.
# GET /user -> acting identity
# POST /issues/7/comments -> CREATE + PERSIST, return created object
# GET /issues/comments/{id} -> read the persisted record by exact id
# There is deliberately NO comment-LIST endpoint: exact-id read-back is the sole
# authority, so any residual list enumeration attempt hits the unexpected-request
# guard and fails the test.
cat > "$BIN_DIR/curl" <<'SH'
#!/usr/bin/env bash
set -euo pipefail
# Record the FULL argv exactly as spawned, before consumption. The bearer token
# must NOT appear here — it is delivered via a curl --config file (#865 ITEM 3a),
# so only the config file PATH may show up.
printf '%s\n' "$*" >> "$ISSUE_COMMENT_CURL_ARGV_LOG"
output_file=""
method="GET"
url=""
data=""
auth_token=""
config_file=""
while [[ $# -gt 0 ]]; do
case "$1" in
-o) output_file="$2"; shift 2 ;;
-H)
[[ "$2" == Authorization:* ]] && auth_token="${2##* }"
shift 2 ;;
-K|--config) config_file="$2"; shift 2 ;;
-w) shift 2 ;;
-X) method="$2"; shift 2 ;;
-d|--data) data="$2"; shift 2 ;;
-s|-S|-sS) shift ;;
http://*|https://*) url="$1"; shift ;;
*) shift ;;
esac
done
# Resolve the bearer token from the curl --config file (its real, secure source);
# fall back to an -H header only for defense in depth. The config line is
# `header = "Authorization: token <value>"`.
if [[ -z "$auth_token" && -n "$config_file" && -f "$config_file" ]]; then
config_hdr="$(grep -i 'Authorization' "$config_file" 2>/dev/null || true)"
if [[ "$config_hdr" == *"token "* ]]; then
auth_token="${config_hdr##*token }"
auth_token="${auth_token%\"}"
fi
fi
path="${url%%\?*}"
query="${url#*\?}"
[[ "$query" == "$url" ]] && query=""
printf '%s %s\n' "$method" "$url" >> "$ISSUE_COMMENT_CURL_LOG"
# Map the presented bearer token to the identity it authenticates as — the same
# derivation Gitea's own /user does. The wrapper's write, /user lookup, and
# read-back must all carry the SAME token, so the acting identity recorded here
# reveals which credential actually performed the request.
acting_identity=""
case "$auth_token" in
"$ISSUE_COMMENT_DEFAULT_TOKEN") acting_identity="$ISSUE_COMMENT_ACTING_LOGIN" ;;
"$ISSUE_COMMENT_OVERRIDE_TOKEN") acting_identity="$ISSUE_COMMENT_OVERRIDE_LOGIN" ;;
"$ISSUE_COMMENT_CROSS_HOST_TOKEN") acting_identity="$ISSUE_COMMENT_CROSS_HOST_LOGIN" ;;
esac
printf '%s %s %s\n' "$method" "$path" "${acting_identity:-<unauthenticated>}" >> "$ISSUE_COMMENT_AUTH_LOG"
write_response() {
local status="$1" body="$2"
[[ -n "$output_file" ]] || exit 96
printf '%s' "$body" > "$output_file"
printf '%s' "$status"
}
if [[ "$method" == "GET" && "$path" == "$ISSUE_COMMENT_API_ROOT/user" ]]; then
[[ -n "$acting_identity" ]] || { write_response 401 '{"message":"unauthenticated"}'; exit 0; }
write_response 200 "$(ISSUE_COMMENT_LOGIN="$acting_identity" python3 - <<'PY'
import json
import os
print(json.dumps({"login": os.environ["ISSUE_COMMENT_LOGIN"]}))
PY
)"
elif [[ "$method" == "POST" && "$path" == "$ISSUE_COMMENT_API_BASE/issues/7/comments" ]]; then
result=$(ISSUE_COMMENT_ACTING_LOGIN="${acting_identity:-$ISSUE_COMMENT_ACTING_LOGIN}" ISSUE_COMMENT_DATA="$data" python3 - <<'PY'
import json
import os
state_path = os.environ["ISSUE_COMMENT_STATE"]
mode = os.environ["ISSUE_COMMENT_TEST_MODE"]
acting = os.environ["ISSUE_COMMENT_ACTING_LOGIN"]
foreign = os.environ["ISSUE_COMMENT_FOREIGN_LOGIN"]
repo = os.environ["ISSUE_COMMENT_REPO_SLUG"]
body = json.loads(os.environ["ISSUE_COMMENT_DATA"]).get("body")
with open(state_path, encoding="utf-8") as handle:
comments = json.load(handle)
# no-op-concurrent: the wrapper's own write is SUPPRESSED (returns 200 with no
# created object) even though a concurrent same-identity comment already exists
# in state. Nothing is persisted; there is no created id to verify.
if mode == "no-op-concurrent":
print("200")
print(json.dumps({}))
raise SystemExit(0)
author = foreign if mode == "author-mismatch" else acting
new_id = (max((c["id"] for c in comments), default=0)) + 1
# REAL Gitea comment shape: issue_url is the WEB (html) path, not an API path,
# and a plain issue comment leaves pull_request_url empty. The URL-injection
# modes persist a record whose id/author/body are all correct but whose
# issue_url is forged, so ONLY the origin+path verification can catch them.
issue_url = f"https://git.mosaicstack.dev/{repo}/issues/7"
if mode == "url-wrong-host":
issue_url = f"https://evil.example/{repo}/issues/7"
elif mode == "url-wrong-owner":
issue_url = "https://git.mosaicstack.dev/attacker/stack/issues/7"
elif mode == "url-wrong-repo":
issue_url = "https://git.mosaicstack.dev/mosaicstack/other/issues/7"
elif mode == "url-suffix-injection":
# Prefix-injected: a bare endswith("/<slug>/issues/7") test would ACCEPT this.
issue_url = f"https://git.mosaicstack.dev/deceptive/{repo}/issues/7"
record = {
"id": new_id,
"body": body,
"user": {"login": author},
"issue_url": issue_url,
"pull_request_url": "",
}
comments.append(record)
with open(state_path, "w", encoding="utf-8") as handle:
json.dump(comments, handle)
print("201")
print(json.dumps(record))
PY
)
write_response "$(printf '%s' "$result" | head -n1)" "$(printf '%s' "$result" | tail -n +2)"
elif [[ "$method" == "GET" && "$path" == "$ISSUE_COMMENT_API_BASE"/issues/comments/* ]]; then
result=$(ISSUE_COMMENT_GET_ID="${path##*/}" python3 - <<'PY'
import json
import os
state_path = os.environ["ISSUE_COMMENT_STATE"]
wanted = int(os.environ["ISSUE_COMMENT_GET_ID"])
with open(state_path, encoding="utf-8") as handle:
comments = json.load(handle)
match = next((c for c in comments if c["id"] == wanted), None)
if match is None:
print("404")
print(json.dumps({"message": "not found"}))
else:
print("200")
print(json.dumps(match))
PY
)
write_response "$(printf '%s' "$result" | head -n1)" "$(printf '%s' "$result" | tail -n +2)"
else
echo "Unexpected curl request: $method $url" >&2
exit 97
fi
SH
chmod +x "$BIN_DIR/curl"
# Seed persistent server state for a mode, then run the wrapper against it.
seed_state() {
local mode="$1"
ISSUE_COMMENT_SEED_MODE="$mode" ISSUE_COMMENT_SEED_BODY="$BODY" \
ISSUE_COMMENT_SEED_ACTING="$ACTING_LOGIN" ISSUE_COMMENT_SEED_REPO="$REPO_SLUG" \
python3 - "$STATE_FILE" <<'PY'
import json
import os
import sys
mode = os.environ["ISSUE_COMMENT_SEED_MODE"]
body = os.environ["ISSUE_COMMENT_SEED_BODY"]
acting = os.environ["ISSUE_COMMENT_SEED_ACTING"]
repo = os.environ["ISSUE_COMMENT_SEED_REPO"]
# REAL Gitea comment shape: issue_url is the WEB path, pull_request_url empty.
issue_url = f"https://git.mosaicstack.dev/{repo}/issues/7"
def comment(cid, text, author):
return {
"id": cid,
"body": text,
"user": {"login": author},
"issue_url": issue_url,
"pull_request_url": "",
}
if mode == "fresh-success":
# 50 pre-existing comments already exist; the comment this run creates
# becomes id 51, proving exact-id read-back works regardless of how many
# comments precede it (no list enumeration is involved).
comments = [comment(i, f"prior {i}", acting) for i in range(1, 51)]
elif mode == "no-op-concurrent":
# A concurrent SAME-IDENTITY comment with the IDENTICAL body already exists.
# The wrapper's own write will be a no-op; it must still fail closed because
# no created id is returned — it must not scan and accept this record.
comments = [comment(55, body, acting)]
else: # author-mismatch
comments = []
with open(sys.argv[1], "w", encoding="utf-8") as handle:
json.dump(comments, handle)
PY
}
run_comment() {
local mode="$1"
shift
: > "$TEA_LOG"
: > "$CURL_LOG"
: > "$CURL_ARGV_LOG"
: > "$AUTH_LOG"
: > "$OUTPUT_FILE"
seed_state "$mode"
(
cd "$REPO_DIR"
PATH="$BIN_DIR:$PATH" \
TMPDIR="$TMP_SCRATCH" \
XDG_CONFIG_HOME="$XDG_DIR" \
MOSAIC_CREDENTIALS_FILE="$CREDENTIALS_FILE" \
ISSUE_COMMENT_TEA_LOG="$TEA_LOG" \
ISSUE_COMMENT_CURL_LOG="$CURL_LOG" \
ISSUE_COMMENT_CURL_ARGV_LOG="$CURL_ARGV_LOG" \
ISSUE_COMMENT_AUTH_LOG="$AUTH_LOG" \
ISSUE_COMMENT_STATE="$STATE_FILE" \
ISSUE_COMMENT_TEST_MODE="$mode" \
ISSUE_COMMENT_ACTING_LOGIN="$ACTING_LOGIN" \
ISSUE_COMMENT_FOREIGN_LOGIN="$FOREIGN_LOGIN" \
ISSUE_COMMENT_OVERRIDE_LOGIN="$OVERRIDE_LOGIN" \
ISSUE_COMMENT_CROSS_HOST_LOGIN="$CROSS_HOST_LOGIN" \
ISSUE_COMMENT_DEFAULT_TOKEN="$DEFAULT_TOKEN" \
ISSUE_COMMENT_OVERRIDE_TOKEN="$OVERRIDE_TOKEN" \
ISSUE_COMMENT_CROSS_HOST_TOKEN="$CROSS_HOST_TOKEN" \
ISSUE_COMMENT_REPO_SLUG="$REPO_SLUG" \
ISSUE_COMMENT_API_BASE="$API_BASE" \
ISSUE_COMMENT_API_ROOT="$API_ROOT" \
"$SCRIPT_DIR/issue-comment.sh" -i "$ISSUE_NUMBER" -c "$BODY" "$@"
) > "$OUTPUT_FILE" 2>&1
}
# Assert the wrapper left no scratch temp files behind in TMPDIR (POST/GET
# request bodies + metadata). Called after both success and failure paths so a
# clobbered/leaked RETURN trap is caught on every exit route.
assert_no_temp_leak() {
local context="$1" leaked
# Includes the curl auth-config files (mosaic-gitea-auth-*), which carry the
# bearer token and must be unlinked on every exit path.
leaked=$(find "$TMP_SCRATCH" -type f \( -name 'mosaic-issue-comment-*' -o -name 'mosaic-gitea-auth-*' \) 2>/dev/null || true)
if [[ -n "$leaked" ]]; then
echo "FAIL: issue-comment temp files leaked ($context):" >&2
printf '%s\n' "$leaked" >&2
exit 1
fi
}
# Assert the presented bearer token NEVER appeared in curl's argv (it must travel
# via a curl --config file), and that --config auth was actually used. On the
# expected path grep matches nothing, so no token value is ever printed.
assert_token_not_in_argv() {
local context="$1"
if grep -qF -e "$DEFAULT_TOKEN" -e "$OVERRIDE_TOKEN" -e "$CROSS_HOST_TOKEN" "$CURL_ARGV_LOG"; then
echo "FAIL: a Gitea bearer token leaked into curl argv ($context)" >&2
exit 1
fi
if ! grep -q -- '--config' "$CURL_ARGV_LOG"; then
echo "FAIL: curl was not invoked with --config file auth ($context)" >&2
exit 1
fi
}
# Case 1: a genuine REST create (id 51) is verified end to end via its exact
# provider-returned id — no list enumeration is involved.
run_comment fresh-success
grep -q 'Added and verified comment on Gitea issue #7 (comment ID 51)' "$OUTPUT_FILE"
# The write is a REST POST, never a tea comment.
grep -q "^POST $API_BASE/issues/7/comments$" "$CURL_LOG"
if grep -Eq '^comment |^issue comment ' "$TEA_LOG"; then
echo "FAIL: wrapper wrote a comment via tea instead of REST" >&2
exit 1
fi
# Read-back is a DIRECT GET of the exact created id.
grep -q "^GET $API_BASE/issues/comments/51$" "$CURL_LOG"
# Acting identity resolved via GET /user.
grep -q "^GET $API_ROOT/user$" "$CURL_LOG"
# No comment-list enumeration is performed — the exact-id GET is authoritative.
if grep -Eq "^GET $API_BASE/issues/7/comments(\?|$)" "$CURL_LOG"; then
echo "FAIL: wrapper performed a redundant comment-list enumeration" >&2
exit 1
fi
# Default path (no --login): the host credential fallback resolves, and the
# write is performed AND self-verified under the host-default acting identity.
grep -q "^POST $API_BASE/issues/7/comments $ACTING_LOGIN$" "$AUTH_LOG"
grep -q "^GET $API_BASE/issues/comments/51 $ACTING_LOGIN$" "$AUTH_LOG"
# Success path leaves no scratch temp files behind.
assert_no_temp_leak "fresh-success"
# ITEM 3a: the token drove the write/read-back chain but never appeared in curl
# argv — it was passed via a curl --config file.
assert_token_not_in_argv "fresh-success default-token"
# Case 2: a no-op write with a concurrent SAME-IDENTITY, same-body comment
# already present must FAIL CLOSED — the closed concurrency window.
if run_comment no-op-concurrent; then
echo "FAIL: wrapper reported success when its write no-opped but a concurrent same-identity comment existed" >&2
cat "$OUTPUT_FILE" >&2
exit 1
fi
if grep -q 'Added and verified comment' "$OUTPUT_FILE"; then
echo "FAIL: wrapper accepted a concurrent record for a no-op write (window not closed)" >&2
exit 1
fi
# It must NOT have fallen back to a list scan that could find the concurrent id.
if grep -q "^GET $API_BASE/issues/comments/55$" "$CURL_LOG"; then
echo "FAIL: wrapper read back the concurrent comment id 55 (illegitimate fallback)" >&2
exit 1
fi
# Case 3: a created record NOT authored by the acting identity must FAIL CLOSED.
if run_comment author-mismatch; then
echo "FAIL: wrapper accepted a created comment authored by a different identity" >&2
cat "$OUTPUT_FILE" >&2
exit 1
fi
if grep -q 'Added and verified comment' "$OUTPUT_FILE"; then
echo "FAIL: read-back did not enforce acting-identity authorship" >&2
exit 1
fi
# Failure-after-read-back path must ALSO leave no scratch temp files behind
# (proves the RETURN traps clean up on the error-return route, not just success).
assert_no_temp_leak "author-mismatch"
# Case 4: a RESOLVABLE --login override — the write, the /user identity lookup,
# and the read-back must ALL be performed under THAT login's token/identity, not
# the host default. The override login has id 1 (empty seed).
run_comment override-success --login "$OVERRIDE_LOGIN"
grep -q 'Added and verified comment on Gitea issue #7 (comment ID 1)' "$OUTPUT_FILE"
grep -q "^GET $API_ROOT/user $OVERRIDE_LOGIN$" "$AUTH_LOG"
grep -q "^POST $API_BASE/issues/7/comments $OVERRIDE_LOGIN$" "$AUTH_LOG"
grep -q "^GET $API_BASE/issues/comments/1 $OVERRIDE_LOGIN$" "$AUTH_LOG"
# The host-default identity must NOT have performed ANY request in this run.
if grep -q " $ACTING_LOGIN\$" "$AUTH_LOG"; then
echo "FAIL: an explicit --login override request was performed under the host default identity" >&2
cat "$AUTH_LOG" >&2
exit 1
fi
# Case 5: an UNRESOLVABLE --login override (name absent from tea config) must
# FAIL CLOSED — no silent downgrade to the host default identity: nonzero exit,
# no success line, and NO write performed.
if run_comment override-unresolvable --login "nonexistent-typo-login"; then
echo "FAIL: unresolvable --login override did not fail closed" >&2
cat "$OUTPUT_FILE" >&2
exit 1
fi
if grep -q 'Added and verified comment' "$OUTPUT_FILE"; then
echo "FAIL: unresolvable --login override reported success" >&2
exit 1
fi
if grep -q "^POST $API_BASE/issues/7/comments" "$CURL_LOG"; then
echo "FAIL: unresolvable --login override still performed a write" >&2
exit 1
fi
# And it must not have silently fallen back to the host default identity.
if grep -q " $ACTING_LOGIN\$" "$AUTH_LOG"; then
echo "FAIL: unresolvable --login override fell back to the host default identity" >&2
exit 1
fi
# Case 6: a --login override that IS present in tea config but whose URL is a
# DIFFERENT host than the repo remote must FAIL CLOSED (host-bound selection).
# The cross-host token must NEVER be sent to the repo host, and no write occurs.
if run_comment cross-host --login "$CROSS_HOST_LOGIN"; then
echo "FAIL: cross-host --login override did not fail closed" >&2
cat "$OUTPUT_FILE" >&2
exit 1
fi
if grep -q 'Added and verified comment' "$OUTPUT_FILE"; then
echo "FAIL: cross-host --login override reported success" >&2
exit 1
fi
# The cross-host credential must not have performed ANY request against the repo
# host — no request may be attributed to the cross-host identity.
if grep -q " $CROSS_HOST_LOGIN\$" "$AUTH_LOG"; then
echo "FAIL: cross-host credential was sent to the repo host (cross-host leak)" >&2
cat "$AUTH_LOG" >&2
exit 1
fi
if grep -q "^POST $API_BASE/issues/7/comments" "$CURL_LOG"; then
echo "FAIL: cross-host --login override still performed a write" >&2
exit 1
fi
# It must not have silently downgraded to the host default identity either.
if grep -q " $ACTING_LOGIN\$" "$AUTH_LOG"; then
echo "FAIL: cross-host --login override fell back to the host default identity" >&2
exit 1
fi
assert_no_temp_leak "cross-host"
# Cases 7-10 (#865 Blocker 3): the created record's id/author/body are all
# correct, but its provider-returned issue_url is forged. Verification pins the
# URL's ORIGIN (scheme+host+effective-port) and its FULL path (deployment prefix
# + exact owner/repo + kind + number), so each forgery must FAIL CLOSED. A bare
# endswith/suffix test would wrongly accept the look-alike-host and
# prefix-injection variants.
for bad_mode in url-wrong-host url-wrong-owner url-wrong-repo url-suffix-injection; do
if run_comment "$bad_mode"; then
echo "FAIL: forged comment URL ($bad_mode) was accepted" >&2
cat "$OUTPUT_FILE" >&2
exit 1
fi
if grep -q 'Added and verified comment' "$OUTPUT_FILE"; then
echo "FAIL: forged comment URL ($bad_mode) passed verification" >&2
exit 1
fi
assert_no_temp_leak "$bad_mode"
done
# Sanity: the exact same verification path still ACCEPTS a legitimate web-shaped
# issue_url (already exercised by Case 1's fresh-success), so the tightened check
# is not rejecting genuine writes.
echo "issue-comment.sh REST create + exact-id read-back regression passed"

View File

@@ -0,0 +1,180 @@
#!/usr/bin/env python3
"""Enforcement-side version-coupling gate (issue #869, Point-1 card C4).
Root cause this exists to guard against (#828 version skew, restated from
the C1 activation probe in ``lease-activation-probe.ts``): the lease
broker's ENFORCEMENT half (this toolkit — ``launch-runtime.py``,
``mutator-gate.py``, ``revoke-lease.py``) and its ACTIVATION half
(``execLeaseGatedRuntime()`` in ``launch.ts``, which chains the gated
runtime through ``launch-runtime.py`` and injects ``MOSAIC_LEASE_*``) ship
on different channels — an npm package and a framework/CLI reseed. C1 gave
the activation half a versioned, machine-checkable identity
(``LEASE_ACTIVATION_CAPABILITY``, printed by the CLI's hidden
``mosaic __lease-capability`` subcommand). That identity is inert on its
own: nothing yet asserted that ENFORCEMENT actually requires the version
ACTIVATION advertises. This module is that assertion, owned by the
enforcement side.
``EXPECTED_ACTIVATION_CAPABILITY`` below is this toolkit's own contract
declaration — bump it only when this toolkit's launch/gate seam starts
requiring a different activation contract (new env vars it depends on,
changed chaining behavior, etc.), independent of any package semver, for
the same reason C1's constant is: #828 happened precisely because a
version number that should have moved did not.
This module never talks to a real broker or a real installed CLI in its
own tests — both the probe's command resolution and its ``run`` transport
are injectable so tests can drive every branch with fakes/stubs (see
``version_coupling_unittest.py``).
"""
from __future__ import annotations
import json
import os
import shlex
import shutil
import subprocess
from collections.abc import Callable, Mapping
from typing import Final, TypedDict
class ActivationCapability(TypedDict):
name: str
version: int
# ENFORCEMENT-side expected activation contract. OWNED by this toolkit (the
# enforcement half). Mirrors — but is deliberately a SEPARATE constant from
# — `LEASE_ACTIVATION_CAPABILITY` in
# `packages/mosaic/src/commands/lease-activation-probe.ts` (the activation
# half's own declaration of what it implements). The two are compared at
# runtime by `assert_activation_capability_matches()`; drift between them is
# exactly the version-skew failure mode #828/#869 exist to catch, and must
# FAIL LOUD, never a silent pass and never a dead (always-true) gate.
EXPECTED_ACTIVATION_CAPABILITY: Final[ActivationCapability] = {
"name": "lease-runtime-activation",
"version": 1,
}
# Matches `LEASE_CAPABILITY_PROBE_COMMAND` in lease-activation-probe.ts —
# the hidden CLI subcommand that prints the activation half's advertised
# capability as compact JSON.
LEASE_CAPABILITY_PROBE_COMMAND: Final = "__lease-capability"
PROBE_TIMEOUT_SECONDS: Final = 2.0
# Override hook: a full shell-style command line (parsed with `shlex.split`)
# to run INSTEAD of resolving `mosaic` on PATH and appending the probe
# subcommand. Real deployments should never need this — `mosaic` is on PATH
# whenever a runtime was launched via `mosaic <cmd>` in the first place, the
# only real caller of this seam. It exists for integration tests that spawn
# `launch-runtime.py` directly (never through the real CLI) to supply a
# fake/stub CLI probe, matching the existing convention of those tests
# supplying a fake broker and a fake runtime binary rather than depending on
# host state.
MOSAIC_COMMAND_OVERRIDE_VAR: Final = "MOSAIC_LEASE_VERSION_PROBE_COMMAND"
class VersionCouplingError(Exception):
"""Raised when the activation capability is absent, unreadable, or does
not match what enforcement expects. Callers MUST fail loud on this
(non-zero exit, clear actionable stderr) — never swallow it into a
silent pass, and never let its absence be treated as compatible."""
def _resolve_probe_command(environ: Mapping[str, str]) -> list[str] | None:
override = environ.get(MOSAIC_COMMAND_OVERRIDE_VAR)
if override:
parsed = shlex.split(override)
return parsed or None
resolved = shutil.which("mosaic")
if resolved is None:
return None
return [resolved, LEASE_CAPABILITY_PROBE_COMMAND]
def default_probe_activation_capability(
environ: Mapping[str, str] | None = None,
*,
run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> ActivationCapability | None:
"""Real capability lookup: resolves and executes the CLI's hidden
``__lease-capability`` probe subcommand out-of-process (the same
mechanism `defaultCapabilityProbe()` in lease-activation-probe.ts uses
from the activation side) and parses its JSON stdout. Any failure to
resolve a command, spawn it, have it exit zero, or produce a well-shaped
``{name, version}`` JSON object is treated as NO capability (``None``)
— fail-closed, never a fabricated/guessed capability.
"""
source_environment = os.environ if environ is None else environ
command = _resolve_probe_command(source_environment)
if command is None:
return None
try:
completed = run(
command,
capture_output=True,
text=True,
timeout=PROBE_TIMEOUT_SECONDS,
check=False,
)
except (OSError, subprocess.TimeoutExpired, ValueError):
return None
if completed.returncode != 0:
return None
try:
parsed = json.loads(completed.stdout)
except json.JSONDecodeError:
return None
if (
not isinstance(parsed, dict)
or not isinstance(parsed.get("name"), str)
or not isinstance(parsed.get("version"), int)
or isinstance(parsed.get("version"), bool)
):
return None
return {"name": parsed["name"], "version": parsed["version"]}
def format_mismatch_message(
activation: ActivationCapability | None,
expected: ActivationCapability,
) -> str:
"""Actionable, non-silent remediation message for either failure shape:
absent/unreadable capability, or a present-but-incompatible one."""
if activation is None:
return (
"Mosaic lease activation capability unreadable: enforcement "
f"expects '{expected['name']}' v{expected['version']} but the "
f"CLI's `mosaic {LEASE_CAPABILITY_PROBE_COMMAND}` probe produced "
"no usable result (mosaic not on PATH, non-zero exit, or "
"malformed output) — framework/CLI version skew; upgrade both "
"as one unit; see #869."
)
if activation["name"] != expected["name"]:
return (
f"activation capability name '{activation['name']}' != "
f"enforcement expects '{expected['name']}' — framework/CLI "
"version skew; upgrade both as one unit; see #869"
)
return (
f"activation capability v{activation['version']} != enforcement "
f"expects v{expected['version']} — framework/CLI version skew; "
"upgrade both as one unit; see #869"
)
def assert_activation_capability_matches(
activation: ActivationCapability | None,
expected: ActivationCapability = EXPECTED_ACTIVATION_CAPABILITY,
) -> None:
"""Raise `VersionCouplingError` unless `activation` is present AND its
`name`/`version` exactly match `expected`. Absence is treated the same
as a mismatch — never a silent pass."""
if (
activation is None
or activation.get("name") != expected["name"]
or activation.get("version") != expected["version"]
):
raise VersionCouplingError(format_mismatch_message(activation, expected))

View File

@@ -12,11 +12,24 @@ from collections.abc import Callable, Mapping, Sequence
from pathlib import Path from pathlib import Path
from typing import Final from typing import Final
from activation_version_gate import (
EXPECTED_ACTIVATION_CAPABILITY,
ActivationCapability,
VersionCouplingError,
assert_activation_capability_matches,
default_probe_activation_capability,
)
from lease_generation import initialize_runtime_generation from lease_generation import initialize_runtime_generation
MAX_FRAME: Final = 64 * 1024 MAX_FRAME: Final = 64 * 1024
BROKER_TIMEOUT_SECONDS: Final = 1.5 BROKER_TIMEOUT_SECONDS: Final = 1.5
CLAUDE_DANGEROUS_FLAG: Final = "--dangerously-skip-permissions" CLAUDE_DANGEROUS_FLAG: Final = "--dangerously-skip-permissions"
# Distinct, non-overlapping exit code for the C4 version-coupling gate (see
# `activation_version_gate.py`) — deliberately different from the `1`
# (broker registration failed closed) and `64` (usage error) codes already
# owned by this script, so a version-skew denial is unambiguous in caller
# logs/tests and is never confused with a broker-availability failure.
EXIT_VERSION_SKEW: Final = 65
def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]: def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
@@ -47,6 +60,10 @@ def main(
request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request, request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
execute: Callable[[str, list[str], dict[str, str]], object] = os.execvpe, execute: Callable[[str, list[str], dict[str, str]], object] = os.execvpe,
initialize_generation: Callable[[Path, int], None] = initialize_runtime_generation, initialize_generation: Callable[[Path, int], None] = initialize_runtime_generation,
probe_activation_capability: Callable[
[Mapping[str, str]], ActivationCapability | None
] = default_probe_activation_capability,
expected_activation_capability: ActivationCapability = EXPECTED_ACTIVATION_CAPABILITY,
) -> int: ) -> int:
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("--runtime", required=True, choices=("claude", "pi")) parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
@@ -66,6 +83,25 @@ def main(
command = [command[0], CLAUDE_DANGEROUS_FLAG, *command[1:]] command = [command[0], CLAUDE_DANGEROUS_FLAG, *command[1:]]
source_environment = os.environ if environ is None else environ source_environment = os.environ if environ is None else environ
# C4 version-coupling gate (#869 Point-1): before this ENFORCEMENT half
# chains into anything, assert that the ACTIVATION contract it is about
# to rely on (MOSAIC_LEASE_* injection, broker chaining) matches what
# this enforcement build expects. This is a build/deploy-defect check,
# not a broker-availability question, so it runs before — and
# independently of — broker registration below, and it FAILS LOUD: a
# clear stderr message plus a dedicated non-zero exit code, never a
# silent pass and never folded into the generic registration-failure
# branch.
try:
assert_activation_capability_matches(
probe_activation_capability(source_environment),
expected_activation_capability,
)
except VersionCouplingError as version_error:
print(str(version_error), file=sys.stderr)
return EXIT_VERSION_SKEW
try: try:
socket_path = Path(source_environment["MOSAIC_LEASE_BROKER_SOCKET"]) socket_path = Path(source_environment["MOSAIC_LEASE_BROKER_SOCKET"])
generation = int(source_environment.get("MOSAIC_RUNTIME_GENERATION", "1")) generation = int(source_environment.get("MOSAIC_RUNTIME_GENERATION", "1"))

View File

@@ -25,7 +25,7 @@
"lint": "eslint src", "lint": "eslint src",
"typecheck": "tsc --noEmit", "typecheck": "tsc --noEmit",
"test": "vitest run --passWithNoTests && pnpm run test:framework-shell", "test": "vitest run --passWithNoTests && pnpm run test:framework-shell",
"test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/lease-broker/framework_skill_portability_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh && bash framework/tools/qa/test-deps-preflight.sh && bash framework/tools/git/test-pr-review-gitea-comment.sh" "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_unittest.py && python3 src/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/lease-broker/framework_skill_portability_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 src/mutator-gate/version_coupling_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh && bash framework/tools/qa/test-deps-preflight.sh && bash framework/tools/git/test-pr-review-gitea-comment.sh && bash framework/tools/git/test-ci-queue-wait-branch-absent.sh && bash framework/tools/git/test-git-credential-mosaic.sh && bash framework/tools/git/test-gitea-token-identity.sh && bash framework/tools/_scripts/test-install-ordering-guard.sh"
}, },
"dependencies": { "dependencies": {
"@mosaicstack/brain": "workspace:*", "@mosaicstack/brain": "workspace:*",

View File

@@ -21,6 +21,8 @@ import { registerRestoreCommand } from './commands/restore.js';
import { registerSkillCommand } from './commands/skill.js'; import { registerSkillCommand } from './commands/skill.js';
// prdy is registered via launch.ts // prdy is registered via launch.ts
import { registerLaunchCommands } from './commands/launch.js'; import { registerLaunchCommands } from './commands/launch.js';
import { registerLeaseCapabilityProbe } from './commands/lease-activation-probe.js';
import { registerInstallOrderingGuardCommand } from './commands/install-ordering-guard.js';
import { registerAuthCommand } from './commands/auth.js'; import { registerAuthCommand } from './commands/auth.js';
import { registerFederationCommand } from './commands/federation.js'; import { registerFederationCommand } from './commands/federation.js';
import { registerGatewayCommand } from './commands/gateway.js'; import { registerGatewayCommand } from './commands/gateway.js';
@@ -78,6 +80,14 @@ Command Groups:
registerLaunchCommands(program); registerLaunchCommands(program);
// ─── lease activation capability probe (hidden; #869 Point-1 C1) ────────
registerLeaseCapabilityProbe(program);
// ─── install-ordering guard (hidden; #869 Point-1 C2) ───────────────────
registerInstallOrderingGuardCommand(program);
// ─── login ────────────────────────────────────────────────────────────── // ─── login ──────────────────────────────────────────────────────────────
program program

View File

@@ -0,0 +1,301 @@
import { describe, it, expect, afterEach } from 'vitest';
import { mkdtempSync, rmSync, writeFileSync, readFileSync, existsSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { leaseEnforcementActivatable } from './lease-activation-probe.js';
import {
ENFORCEMENT_HOOK_MARKERS,
FAIL_LOUD_MESSAGE,
guardClaudeSettingsWiring,
loudOptOutMessage,
runInstallOrderingGuard,
settingsHasEnforcementHooks,
stripEnforcementHooks,
} from './install-ordering-guard.js';
/**
* Red-first tests for issue #869 Point-1 C2 — the install-ordering guard.
*
* Root cause under test: `mosaic-link-runtime-assets` copies
* `runtime/claude/settings.json` (which embeds the PreToolUse
* `mutator-gate.py` hook and the Stop `receipt-observer-client.py` hook)
* straight into `~/.claude/settings.json`, unconditionally. If the
* activation half (C1: `leaseEnforcementActivatable()`) cannot be confirmed,
* wiring those hooks bricks the host with a fail-closed gate that can never
* be satisfied. This guard must refuse to wire in that case by default, and
* only wire anyway on an explicit, loud opt-out.
*
* All fixtures use temp directories — this suite never reads or writes the
* real `~/.claude/settings.json`.
*/
const FIXTURE_SETTINGS = {
model: 'opus',
hooks: {
PreCompact: [
{
matcher: '.*',
hooks: [{ type: 'command', command: 'python3 revoke-lease.py --reason pre-compact' }],
},
],
PreToolUse: [
{
matcher: '.*',
hooks: [
{
type: 'command',
command: 'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude',
timeout: 3,
},
],
},
{
matcher: 'Write|Edit|MultiEdit',
hooks: [{ type: 'command', command: '~/.config/mosaic/tools/qa/prevent-memory-write.sh' }],
},
],
PostToolUse: [
{
matcher: 'Edit|MultiEdit|Write',
hooks: [{ type: 'command', command: '~/.config/mosaic/tools/qa/qa-hook-stdin.sh' }],
},
],
Stop: [
{
hooks: [
{
type: 'command',
command:
'python3 ~/.config/mosaic/tools/lease-broker/receipt-observer-client.py --runtime claude',
timeout: 3,
},
{ type: 'command', command: '~/.config/mosaic/tools/qa/reflect-stop-hook.sh' },
],
},
],
},
enabledPlugins: { 'feature-dev@claude-plugins-official': true },
};
function fixtureJson(): string {
return JSON.stringify(FIXTURE_SETTINGS, null, 2) + '\n';
}
describe('stripEnforcementHooks', () => {
it('removes the PreToolUse mutator-gate trigger entirely', () => {
const { settings } = stripEnforcementHooks(FIXTURE_SETTINGS);
const hooks = settings['hooks'] as Record<string, unknown[]>;
const preToolUse = hooks['PreToolUse'] as Array<{ hooks: Array<{ command: string }> }>;
expect(preToolUse.some((t) => t.hooks.some((h) => h.command.includes('mutator-gate.py')))).toBe(
false,
);
});
it('preserves the sibling prevent-memory-write.sh PreToolUse trigger', () => {
const { settings } = stripEnforcementHooks(FIXTURE_SETTINGS);
const hooks = settings['hooks'] as Record<string, unknown[]>;
const preToolUse = hooks['PreToolUse'] as Array<{ hooks: Array<{ command: string }> }>;
expect(
preToolUse.some((t) => t.hooks.some((h) => h.command.includes('prevent-memory-write.sh'))),
).toBe(true);
});
it('removes only the receipt-observer-client.py hook from Stop, keeping reflect-stop-hook.sh', () => {
const { settings } = stripEnforcementHooks(FIXTURE_SETTINGS);
const hooks = settings['hooks'] as Record<string, unknown[]>;
const stop = hooks['Stop'] as Array<{ hooks: Array<{ command: string }> }>;
const commands = stop.flatMap((t) => t.hooks.map((h) => h.command));
expect(commands.some((c) => c.includes('receipt-observer-client.py'))).toBe(false);
expect(commands.some((c) => c.includes('reflect-stop-hook.sh'))).toBe(true);
});
it('leaves PreCompact/PostToolUse hooks byte-identical', () => {
const { settings } = stripEnforcementHooks(FIXTURE_SETTINGS);
const hooks = settings['hooks'] as Record<string, unknown>;
expect(hooks['PreCompact']).toEqual(FIXTURE_SETTINGS.hooks.PreCompact);
expect(hooks['PostToolUse']).toEqual(FIXTURE_SETTINGS.hooks.PostToolUse);
});
it('reports what it removed', () => {
const { removed } = stripEnforcementHooks(FIXTURE_SETTINGS);
expect(removed).toContain(`PreToolUse:${ENFORCEMENT_HOOK_MARKERS.preToolUse}`);
expect(removed).toContain(`Stop:${ENFORCEMENT_HOOK_MARKERS.stop}`);
});
});
describe('settingsHasEnforcementHooks', () => {
it('is true for the unmodified fixture', () => {
expect(settingsHasEnforcementHooks(FIXTURE_SETTINGS)).toBe(true);
});
it('is false after stripping', () => {
const { settings } = stripEnforcementHooks(FIXTURE_SETTINGS);
expect(settingsHasEnforcementHooks(settings)).toBe(false);
});
it('is false for settings with no hooks key at all', () => {
expect(settingsHasEnforcementHooks({ model: 'opus' })).toBe(false);
});
});
describe('guardClaudeSettingsWiring', () => {
it('probe=false (default, no opt-out): strips enforcement hooks and reports non-zero with a loud, actionable message', () => {
const outcome = guardClaudeSettingsWiring(fixtureJson(), {}, { activatable: () => false });
expect(outcome.exitCode).toBe(1);
expect(outcome.wired).toBe(false);
expect(settingsHasEnforcementHooks(JSON.parse(outcome.json) as Record<string, unknown>)).toBe(
false,
);
expect(outcome.logs).toHaveLength(1);
expect(outcome.logs[0]?.level).toBe('error');
expect(outcome.logs[0]?.message).toBe(FAIL_LOUD_MESSAGE);
expect(outcome.logs[0]?.message).toMatch(/refusing to wire a dead gate/i);
expect(outcome.logs[0]?.message).toMatch(/#869/);
expect(outcome.logs[0]?.message).toMatch(/--allow-inactive-enforcement/);
});
it('probe=false + explicit opt-out flag: wires hooks as-is and emits a loud warning', () => {
const outcome = guardClaudeSettingsWiring(
fixtureJson(),
{ allowInactiveEnforcement: true },
{ activatable: () => false },
);
expect(outcome.exitCode).toBe(0);
expect(outcome.wired).toBe(true);
expect(settingsHasEnforcementHooks(JSON.parse(outcome.json) as Record<string, unknown>)).toBe(
true,
);
expect(outcome.logs).toHaveLength(1);
expect(outcome.logs[0]?.level).toBe('warn');
expect(outcome.logs[0]?.message).toBe(loudOptOutMessage());
expect(outcome.logs[0]?.message).toMatch(/WITHOUT confirmed activation/);
});
it('probe=true: wires hooks normally with no logs, regardless of opt-out', () => {
const outcome = guardClaudeSettingsWiring(fixtureJson(), {}, { activatable: () => true });
expect(outcome.exitCode).toBe(0);
expect(outcome.wired).toBe(true);
expect(outcome.logs).toHaveLength(0);
expect(JSON.parse(outcome.json)).toEqual(FIXTURE_SETTINGS);
});
it('probe=true + opt-out flag set anyway: still wires normally, no spurious warning', () => {
const outcome = guardClaudeSettingsWiring(
fixtureJson(),
{ allowInactiveEnforcement: true },
{ activatable: () => true },
);
expect(outcome.exitCode).toBe(0);
expect(outcome.wired).toBe(true);
expect(outcome.logs).toHaveLength(0);
});
it('defaults to the real leaseEnforcementActivatable() when no activatable dep is injected', () => {
// Deliberately does not assume a fixed true/false value for the real
// probe (whether dist/cli.js happens to be built varies by environment —
// asserting a hardcoded expectation here would make the test flaky, not
// red-first). Instead it proves the wiring is genuinely delegated: the
// no-deps call must agree with an explicit call to the same real
// predicate, not some other hardcoded value.
const reallyActivatable = leaseEnforcementActivatable();
const outcome = guardClaudeSettingsWiring(fixtureJson());
if (reallyActivatable) {
expect(outcome.exitCode).toBe(0);
expect(outcome.wired).toBe(true);
} else {
expect(outcome.exitCode).toBe(1);
expect(outcome.wired).toBe(false);
}
});
});
describe('runInstallOrderingGuard (file-level, temp dirs only)', () => {
let dir: string;
afterEach(() => {
if (dir) rmSync(dir, { recursive: true, force: true });
});
function makeSrc(): string {
dir = mkdtempSync(join(tmpdir(), 'mosaic-install-ordering-guard-'));
const src = join(dir, 'settings.json');
writeFileSync(src, fixtureJson());
return src;
}
it('probe=false: writes a dest settings.json with hooks stripped and returns exitCode 1', () => {
const src = makeSrc();
const dest = join(dir, 'claude-settings.json');
const result = runInstallOrderingGuard(src, dest, {}, { activatable: () => false });
expect(result.exitCode).toBe(1);
expect(result.destWritten).toBe(true);
expect(existsSync(dest)).toBe(true);
const written = JSON.parse(readFileSync(dest, 'utf-8')) as Record<string, unknown>;
expect(settingsHasEnforcementHooks(written)).toBe(false);
});
it('probe=false + opt-out: writes dest with hooks intact and returns exitCode 0', () => {
const src = makeSrc();
const dest = join(dir, 'claude-settings.json');
const result = runInstallOrderingGuard(
src,
dest,
{ allowInactiveEnforcement: true },
{ activatable: () => false },
);
expect(result.exitCode).toBe(0);
const written = JSON.parse(readFileSync(dest, 'utf-8')) as Record<string, unknown>;
expect(settingsHasEnforcementHooks(written)).toBe(true);
});
it('probe=true: writes dest with hooks intact and returns exitCode 0', () => {
const src = makeSrc();
const dest = join(dir, 'claude-settings.json');
const result = runInstallOrderingGuard(src, dest, {}, { activatable: () => true });
expect(result.exitCode).toBe(0);
const written = JSON.parse(readFileSync(dest, 'utf-8')) as Record<string, unknown>;
expect(settingsHasEnforcementHooks(written)).toBe(true);
});
it('backs up a pre-existing divergent dest before overwriting (copy_file_managed parity)', () => {
const src = makeSrc();
const dest = join(dir, 'claude-settings.json');
writeFileSync(dest, JSON.stringify({ preexisting: true }));
const result = runInstallOrderingGuard(src, dest, {}, { activatable: () => true });
expect(result.destWritten).toBe(true);
expect(result.backupPath).toBeDefined();
expect(existsSync(result.backupPath!)).toBe(true);
expect(JSON.parse(readFileSync(result.backupPath!, 'utf-8'))).toEqual({ preexisting: true });
});
it('is a no-op write when dest already matches the guarded content (idempotent)', () => {
const src = makeSrc();
const dest = join(dir, 'claude-settings.json');
const first = runInstallOrderingGuard(src, dest, {}, { activatable: () => true });
expect(first.destWritten).toBe(true);
const second = runInstallOrderingGuard(src, dest, {}, { activatable: () => true });
expect(second.destWritten).toBe(false);
expect(second.backupPath).toBeUndefined();
});
it('never touches the real home directory settings path used by this test file', () => {
// Sanity guard for the suite itself: every dest path used above lives
// under the mkdtemp() scratch dir, never under homedir()/.claude.
expect(dir).toContain('mosaic-install-ordering-guard-');
});
});

View File

@@ -0,0 +1,327 @@
/**
* Install-ordering guard (issue #869, Point-1 card C2).
*
* Root cause this exists to guard against (#828 version skew, restated): the
* framework reseed / install path (`framework/install.sh` →
* `mosaic-link-runtime-assets` → copies `runtime/claude/settings.json` to
* `~/.claude/settings.json`) wires the ENFORCEMENT half of the lease broker —
* the `PreToolUse` `mutator-gate.py` hook and the `Stop`
* `receipt-observer-client.py` hook — unconditionally. If the ACTIVATION half
* (a CLI build advertising launch-runtime activation + a running broker
* supervisor — see `lease-activation-probe.ts`, C1) is absent, the fail-closed
* gate then denies every tool call with GATE_UNAVAILABLE: a bricked host.
*
* This module is the WIRING gate, not the enforcement gate: it decides
* whether the enforcement hook entries are written into the settings.json
* that ships to `~/.claude/`. It never touches `mutator-gate.py`'s own
* fail-closed-on-absent-identity runtime behavior (test-locked in
* `runtime_tools_unittest.py` / `fail-closed-regression.spec.ts`).
*
* Default (no opt-out): NOT activatable → strip the enforcement hook entries
* from the written settings.json and report a non-zero outcome with a loud,
* actionable message (see FAIL_LOUD_MESSAGE below).
*
* Opt-out: `--allow-inactive-enforcement` (an explicit, per-invocation CLI
* flag — deliberately NOT an environment variable, so it can never sit as a
* silently-inherited default in a shell profile). When set on a NOT
* activatable host, the hooks ARE wired but a loud warning is emitted saying
* so, and the outcome is reported ok (this is a conscious, informed choice).
*/
import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { dirname } from 'node:path';
import type { Command } from 'commander';
import { leaseEnforcementActivatable } from './lease-activation-probe.js';
// ─── Enforcement hook identification ────────────────────────────────────────
/** Substrings that identify the two enforcement hook commands #828 wired
* unconditionally. Matches the marker strings documented in
* `lease-activation-probe.ts`. */
export const ENFORCEMENT_HOOK_MARKERS = {
preToolUse: 'mutator-gate.py',
stop: 'receipt-observer-client.py',
} as const;
interface HookEntry {
command?: string;
[key: string]: unknown;
}
interface HookTrigger {
matcher?: string;
hooks?: HookEntry[];
[key: string]: unknown;
}
type HooksMap = Record<string, HookTrigger[]>;
function cloneJson<T>(value: T): T {
return JSON.parse(JSON.stringify(value)) as T;
}
function commandIncludes(hook: HookEntry, marker: string): boolean {
return String(hook.command ?? '').includes(marker);
}
/**
* Return a deep clone of `settings` with the enforcement hook entries removed:
* - Any `PreToolUse` trigger group containing a `mutator-gate.py` command is
* dropped in full (that trigger exists solely to run the gate).
* - Within `Stop` trigger groups, only the individual `receipt-observer-client.py`
* hook entry is dropped; sibling hooks in the same trigger (e.g.
* `reflect-stop-hook.sh`) are preserved.
* Every other hook (PreCompact/SessionStart revoke-lease, the
* `prevent-memory-write.sh` PreToolUse trigger, PostToolUse qa/typecheck
* hooks) is left byte-identical — this function only ever removes the two
* markers above.
*/
export function stripEnforcementHooks(settings: Record<string, unknown>): {
settings: Record<string, unknown>;
removed: string[];
} {
const cloned = cloneJson(settings);
const removed: string[] = [];
const hooks = cloned['hooks'] as HooksMap | undefined;
if (!hooks || typeof hooks !== 'object') {
return { settings: cloned, removed };
}
const preToolUse = hooks['PreToolUse'];
if (Array.isArray(preToolUse)) {
const kept = preToolUse.filter((trigger) => {
const hasGate = (trigger.hooks ?? []).some((h) =>
commandIncludes(h, ENFORCEMENT_HOOK_MARKERS.preToolUse),
);
if (hasGate) removed.push('PreToolUse:mutator-gate.py');
return !hasGate;
});
if (kept.length > 0) hooks['PreToolUse'] = kept;
else delete hooks['PreToolUse'];
}
const stop = hooks['Stop'];
if (Array.isArray(stop)) {
const rebuilt: HookTrigger[] = [];
for (const trigger of stop) {
const innerHooks = trigger.hooks ?? [];
const keptHooks = innerHooks.filter((h) => {
const isReceiptObserver = commandIncludes(h, ENFORCEMENT_HOOK_MARKERS.stop);
if (isReceiptObserver) removed.push('Stop:receipt-observer-client.py');
return !isReceiptObserver;
});
if (keptHooks.length > 0) {
rebuilt.push({ ...trigger, hooks: keptHooks });
}
}
if (rebuilt.length > 0) hooks['Stop'] = rebuilt;
else delete hooks['Stop'];
}
if (Object.keys(hooks).length === 0) {
delete cloned['hooks'];
} else {
cloned['hooks'] = hooks;
}
return { settings: cloned, removed };
}
/** True iff `settings` currently wires either enforcement hook. */
export function settingsHasEnforcementHooks(settings: Record<string, unknown>): boolean {
const hooks = settings['hooks'] as HooksMap | undefined;
if (!hooks || typeof hooks !== 'object') return false;
const preToolUse = hooks['PreToolUse'] ?? [];
const preHit = preToolUse.some((trigger) =>
(trigger.hooks ?? []).some((h) => commandIncludes(h, ENFORCEMENT_HOOK_MARKERS.preToolUse)),
);
if (preHit) return true;
const stop = hooks['Stop'] ?? [];
return stop.some((trigger) =>
(trigger.hooks ?? []).some((h) => commandIncludes(h, ENFORCEMENT_HOOK_MARKERS.stop)),
);
}
// ─── Guard predicate ────────────────────────────────────────────────────────
export const FAIL_LOUD_MESSAGE =
'[mosaic] ERROR: enforcement requested but activation half absent — needs a published CLI ' +
'carrying launch-runtime activation + a broker supervisor; refusing to wire a dead gate (see #869). ' +
'The PreToolUse mutator-gate.py hook and Stop receipt-observer-client.py hook were NOT written to ' +
'settings.json. Fix by installing/updating the CLI and broker, then re-run the framework reseed. ' +
'To wire anyway (NOT recommended — the fail-closed gate will deny every tool call with ' +
'GATE_UNAVAILABLE until activation is restored), re-run with --allow-inactive-enforcement.';
export function loudOptOutMessage(): string {
return (
'[mosaic] WARNING: wiring lease-enforcement hooks (mutator-gate.py / receipt-observer-client.py) ' +
'WITHOUT confirmed activation — --allow-inactive-enforcement was set explicitly. The fail-closed ' +
'gate will deny every tool call (GATE_UNAVAILABLE) until the activation half (launch-runtime ' +
'activation capability + a running broker supervisor) is present on this host (see #869).'
);
}
export type GuardLogLevel = 'error' | 'warn';
export interface GuardLogLine {
level: GuardLogLevel;
message: string;
}
export interface InstallOrderingGuardOptions {
/** Explicit, per-invocation opt-out. Never source this from an environment
* variable — see module doc. */
allowInactiveEnforcement?: boolean;
}
export interface InstallOrderingGuardDeps {
/** Defaults to {@link leaseEnforcementActivatable}. Injectable for tests. */
activatable?: () => boolean;
}
export interface InstallOrderingGuardOutcome {
/** The settings.json content to write (pretty-printed, trailing newline). */
json: string;
/** Whether the enforcement hooks are present in `json`. */
wired: boolean;
/** 0 = proceed normally; 1 = enforcement was refused (fail-loud default path). */
exitCode: 0 | 1;
logs: GuardLogLine[];
}
/**
* The install-ordering guard: decide whether the enforcement hooks embedded
* in the Claude settings.json template may be wired into the settings.json
* actually shipped to `~/.claude/`.
*
* - activatable → wire as-is. exitCode 0, no logs.
* - NOT activatable, no opt-out → strip enforcement hooks. exitCode 1,
* one 'error' log with the actionable FAIL_LOUD_MESSAGE.
* - NOT activatable, opt-out set → wire as-is anyway. exitCode 0, one
* 'warn' log making the risk explicit and loud.
*
* Pure function: takes the raw settings.json text, returns the text to write
* plus metadata. No filesystem access — callers (the hidden CLI subcommand
* below, or a test) own reading/writing so this stays trivially testable with
* fakes/temp files and never risks touching a real `~/.claude/settings.json`.
*/
export function guardClaudeSettingsWiring(
rawSettingsJson: string,
options: InstallOrderingGuardOptions = {},
deps: InstallOrderingGuardDeps = {},
): InstallOrderingGuardOutcome {
const parsed = JSON.parse(rawSettingsJson) as Record<string, unknown>;
const activatable = deps.activatable ?? leaseEnforcementActivatable;
const isActivatable = activatable();
const serialize = (settings: Record<string, unknown>): string =>
JSON.stringify(settings, null, 2) + '\n';
if (isActivatable) {
return {
json: serialize(parsed),
wired: settingsHasEnforcementHooks(parsed),
exitCode: 0,
logs: [],
};
}
if (options.allowInactiveEnforcement === true) {
return {
json: serialize(parsed),
wired: settingsHasEnforcementHooks(parsed),
exitCode: 0,
logs: [{ level: 'warn', message: loudOptOutMessage() }],
};
}
const { settings: stripped } = stripEnforcementHooks(parsed);
return {
json: serialize(stripped),
wired: settingsHasEnforcementHooks(stripped),
exitCode: 1,
logs: [{ level: 'error', message: FAIL_LOUD_MESSAGE }],
};
}
// ─── File-level runner (shared by the CLI action + tests) ──────────────────
export interface RunInstallOrderingGuardResult extends InstallOrderingGuardOutcome {
destWritten: boolean;
backupPath?: string;
}
/**
* Read `src`, guard it, and write the result to `dest` — mirroring
* `copy_file_managed`'s backup-on-change semantics from
* `mosaic-link-runtime-assets` (skip the write if content is unchanged;
* back up an existing divergent file once, timestamped). Exported standalone
* (not only reachable via the CLI action closure) so tests can exercise real
* file I/O against temp directories without ever touching `~/.claude/`.
*/
export function runInstallOrderingGuard(
src: string,
dest: string,
options: InstallOrderingGuardOptions = {},
deps: InstallOrderingGuardDeps = {},
): RunInstallOrderingGuardResult {
const raw = readFileSync(src, 'utf-8');
const outcome = guardClaudeSettingsWiring(raw, options, deps);
mkdirSync(dirname(dest), { recursive: true });
const existing = existsSync(dest) ? readFileSync(dest, 'utf-8') : null;
let destWritten = false;
let backupPath: string | undefined;
if (existing !== outcome.json) {
if (existing !== null) {
const stamp = new Date()
.toISOString()
.replace(/[-:]/g, '')
.replace(/\..+$/, '')
.replace('T', '');
backupPath = `${dest}.mosaic-bak-${stamp}`;
writeFileSync(backupPath, existing);
}
writeFileSync(dest, outcome.json);
destWritten = true;
}
return { ...outcome, destWritten, backupPath };
}
// ─── Hidden CLI bridge (bash → TS) ──────────────────────────────────────────
/** Hidden CLI subcommand name. `mosaic-link-runtime-assets` (bash) invokes
* this instead of its generic `copy_file_managed` for the settings.json
* runtime file specifically, so the guard's decision is made by importing
* `leaseEnforcementActivatable()` directly rather than re-implementing the
* capability/supervisor probes in shell. Deliberately undocumented (hidden
* from `--help`) — internal wiring, not a user-facing command. */
export const INSTALL_ORDERING_GUARD_COMMAND = '__link-claude-settings';
export function registerInstallOrderingGuardCommand(program: Command): void {
program
.command(`${INSTALL_ORDERING_GUARD_COMMAND} <src> <dest>`, { hidden: true })
.description(
'Internal: copy the Claude settings.json template, gating enforcement-hook ' +
'wiring on lease-activation capability (#869 Point-1 C2)',
)
.option(
'--allow-inactive-enforcement',
'Wire enforcement hooks even when activation cannot be confirmed on this host ' +
'(explicit, loud, non-default opt-out — see #869)',
)
.action((src: string, dest: string, opts: { allowInactiveEnforcement?: boolean }) => {
const result = runInstallOrderingGuard(src, dest, {
allowInactiveEnforcement: opts.allowInactiveEnforcement === true,
});
for (const line of result.logs) {
(line.level === 'error' ? console.error : console.warn)(line.message);
}
process.exit(result.exitCode);
});
}

View File

@@ -28,6 +28,7 @@ import { readRegularFileSecure } from '../fleet/secure-file.js';
import { readPersonaContractBlock } from '../fleet/persona-contract.js'; import { readPersonaContractBlock } from '../fleet/persona-contract.js';
import { canonicalizeRoleClass } from './fleet-personas.js'; import { canonicalizeRoleClass } from './fleet-personas.js';
import { launchClaudex, type ClaudexHarnessAdapter } from './claudex.js'; import { launchClaudex, type ClaudexHarnessAdapter } from './claudex.js';
import { runLeaseEnforcementDoctorCheck } from './lease-doctor-check.js';
const MOSAIC_HOME = process.env['MOSAIC_HOME'] ?? join(homedir(), '.config', 'mosaic'); const MOSAIC_HOME = process.env['MOSAIC_HOME'] ?? join(homedir(), '.config', 'mosaic');
const MAX_INSTALLED_TOOLS_BYTES = 256 * 1024; const MAX_INSTALLED_TOOLS_BYTES = 256 * 1024;
@@ -806,7 +807,14 @@ function launchRuntime(runtime: RuntimeName, args: string[], yolo: boolean): nev
process.exit(0); // Unreachable but satisfies never process.exit(0); // Unreachable but satisfies never
} }
function defaultLeaseBrokerSocket(env: NodeJS.ProcessEnv = process.env): string { /**
* Resolve the lease broker's control socket path. Exported (in addition to
* being used internally by execLeaseGatedRuntime) so the C1 activation probe
* (lease-activation-probe.ts) can perform the same resolution when checking
* whether the broker supervisor is reachable — detection only, this never
* connects to the socket itself.
*/
export function defaultLeaseBrokerSocket(env: NodeJS.ProcessEnv = process.env): string {
if (env['MOSAIC_LEASE_BROKER_SOCKET']) return env['MOSAIC_LEASE_BROKER_SOCKET']; if (env['MOSAIC_LEASE_BROKER_SOCKET']) return env['MOSAIC_LEASE_BROKER_SOCKET'];
const runtimeDir = env['XDG_RUNTIME_DIR']; const runtimeDir = env['XDG_RUNTIME_DIR'];
if (runtimeDir) return join(runtimeDir, 'mosaic-lease', 'broker.sock'); if (runtimeDir) return join(runtimeDir, 'mosaic-lease', 'broker.sock');
@@ -895,7 +903,12 @@ function delegateToScript(scriptPath: string, args: string[], env?: Record<strin
* bundled in the @mosaicstack/mosaic npm package (always matches the installed * bundled in the @mosaicstack/mosaic npm package (always matches the installed
* CLI version) over the deployed copy in ~/.config/mosaic/ (may be stale). * CLI version) over the deployed copy in ~/.config/mosaic/ (may be stale).
*/ */
function resolveTool(...segments: string[]): string { /**
* Exported so the C1 activation probe (lease-activation-probe.ts) can resolve
* the same lease-broker launcher/daemon artifacts execLeaseGatedRuntime()
* uses, for detection-only supervisor presence checks.
*/
export function resolveTool(...segments: string[]): string {
try { try {
const req = createRequire(import.meta.url); const req = createRequire(import.meta.url);
const mosaicPkg = dirname(req.resolve('@mosaicstack/mosaic/package.json')); const mosaicPkg = dirname(req.resolve('@mosaicstack/mosaic/package.json'));
@@ -1225,7 +1238,6 @@ export function registerLaunchCommands(program: Command): void {
// Direct framework script delegates // Direct framework script delegates
const directCommands: Record<string, { desc: string; script: string }> = { const directCommands: Record<string, { desc: string; script: string }> = {
init: { desc: 'Generate SOUL.md (agent identity contract)', script: 'mosaic-init' }, init: { desc: 'Generate SOUL.md (agent identity contract)', script: 'mosaic-init' },
doctor: { desc: 'Health audit — detect drift and missing files', script: 'mosaic-doctor' },
sync: { desc: 'Sync skills from canonical source', script: 'mosaic-sync-skills' }, sync: { desc: 'Sync skills from canonical source', script: 'mosaic-sync-skills' },
bootstrap: { bootstrap: {
desc: 'Bootstrap a repo with Mosaic standards', desc: 'Bootstrap a repo with Mosaic standards',
@@ -1244,4 +1256,67 @@ export function registerLaunchCommands(program: Command): void {
delegateToScript(fwScript(script), cmd.args); delegateToScript(fwScript(script), cmd.args);
}); });
} }
// `doctor` — the framework drift audit (bash script) PLUS the #869
// Point-1 C5 lease-enforcement activation check (TS, reusing C1's
// `leaseEnforcementActivatable()` and C3's `checkBrokerSupervisorHealth()`).
// Kept out of the generic `directCommands` loop above because this check
// must run and report BEFORE the bash script's own exit, and must be able
// to force a non-zero exit on its own — a silent pass on "enforcement
// hooks wired but activation absent" would leave a bricked host
// undiagnosed (see lease-doctor-check.ts docstring).
program
.command('doctor')
.description('Health audit — detect drift, missing files, and #869 lease-activation gaps')
.allowUnknownOption(true)
.allowExcessArguments(true)
.action(async (_opts: unknown, cmd: Command) => {
checkMosaicHome();
const leaseCheck = await runLeaseEnforcementDoctorCheck();
const leaseCheckFailed = printLeaseDoctorCheck(leaseCheck);
runDoctorScriptAndExit(fwScript('mosaic-doctor'), cmd.args, leaseCheckFailed);
});
}
/**
* Print the #869 C5 lease-enforcement doctor result using the same
* `[mosaic-doctor]` prefix the bash audit script uses, but with a distinct
* `[ERROR]` severity token (louder than the script's own `[WARN]`) — this is
* a hard, actionable brick warning, not a soft drift warning, and must never
* read as just one more line among the script's routine warnings. Silent on
* an `ok` result, matching this file's other pre-flight checks
* (`checkMosaicHome`, `checkFile`, `checkRuntime`) which only print on
* failure. Returns whether the check failed, so the caller can force a
* non-zero exit regardless of the bash script's own exit code.
*/
function printLeaseDoctorCheck(
result: Awaited<ReturnType<typeof runLeaseEnforcementDoctorCheck>>,
): boolean {
if (result.status === 'error') {
console.error(`[mosaic-doctor] [ERROR] ${result.message}`);
return true;
}
return false;
}
/**
* Run the bash `mosaic-doctor` audit script (inheriting stdio, same as
* {@link delegateToScript}) and exit with a non-zero code if EITHER the
* script itself reported failure OR the lease-enforcement check above did —
* so `--fail-on-warn` and other script-level exit semantics are preserved,
* but the lease-enforcement ERROR can never be masked by an otherwise-green
* script run.
*/
function runDoctorScriptAndExit(scriptPath: string, args: string[], forceFailure: boolean): never {
if (!existsSync(scriptPath)) {
console.error(`[mosaic] Script not found: ${scriptPath}`);
process.exit(1);
}
let scriptExitCode = 0;
try {
execFileSync('bash', [scriptPath, ...args], { stdio: 'inherit', env: process.env });
} catch (err) {
scriptExitCode = (err as { status?: number }).status ?? 1;
}
process.exit(forceFailure ? 1 : scriptExitCode);
} }

View File

@@ -0,0 +1,243 @@
import { describe, it, expect } from 'vitest';
import { Command } from 'commander';
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { tmpdir } from 'node:os';
import { fileURLToPath } from 'node:url';
import {
LEASE_ACTIVATION_CAPABILITY,
LEASE_CAPABILITY_PROBE_COMMAND,
defaultCapabilityProbe,
defaultResolveCliEntry,
defaultSupervisorProbe,
leaseEnforcementActivatable,
registerLeaseCapabilityProbe,
type LeaseActivationCapability,
type SupervisorProbeResult,
} from './lease-activation-probe.js';
/**
* Red-first tests for issue #869 Point-1 C1 — leaseEnforcementActivatable().
*
* Root cause under test: #828 shipped the lease broker's ENFORCEMENT half
* (hooks) and ACTIVATION half (execLeaseGatedRuntime + a running daemon.py
* broker) on different channels, and they drifted — the published CLI
* tarball lacked the activation half even though it existed in source. The
* predicate here must say NO when either half of activation is unavailable,
* and only YES when both are genuinely present — never based on "does the
* source file exist", but on a real capability signal + real supervisor
* detection.
*/
const compatibleCapability: LeaseActivationCapability = { ...LEASE_ACTIVATION_CAPABILITY };
const presentSupervisor: SupervisorProbeResult = {
supervisorPresent: true,
socketPath: '/run/user/1000/mosaic-lease/broker.sock',
};
describe('leaseEnforcementActivatable', () => {
it('is false when the activation capability is absent (null)', () => {
const result = leaseEnforcementActivatable({
getCapability: () => null,
probeSupervisor: () => presentSupervisor,
});
expect(result).toBe(false);
});
it('is false when the activation capability name does not match', () => {
const result = leaseEnforcementActivatable({
getCapability: () => ({
name: 'some-other-capability',
version: LEASE_ACTIVATION_CAPABILITY.version,
}),
probeSupervisor: () => presentSupervisor,
});
expect(result).toBe(false);
});
it('is false when the activation capability version is incompatible (stale/newer build)', () => {
const result = leaseEnforcementActivatable({
getCapability: () => ({
name: LEASE_ACTIVATION_CAPABILITY.name,
version: LEASE_ACTIVATION_CAPABILITY.version + 1,
}),
probeSupervisor: () => presentSupervisor,
});
expect(result).toBe(false);
});
it('is false when the supervisor artifacts (launcher/daemon) are not present', () => {
const result = leaseEnforcementActivatable({
getCapability: () => compatibleCapability,
probeSupervisor: () => ({
supervisorPresent: false,
socketPath: presentSupervisor.socketPath,
}),
});
expect(result).toBe(false);
});
it('is false when the supervisor socket path is not resolvable', () => {
const result = leaseEnforcementActivatable({
getCapability: () => compatibleCapability,
probeSupervisor: () => ({ supervisorPresent: true, socketPath: null }),
});
expect(result).toBe(false);
});
it('is false when BOTH capability and supervisor are absent', () => {
const result = leaseEnforcementActivatable({
getCapability: () => null,
probeSupervisor: () => ({ supervisorPresent: false, socketPath: null }),
});
expect(result).toBe(false);
});
it('is true when a compatible capability AND a resolvable supervisor are both present', () => {
const result = leaseEnforcementActivatable({
getCapability: () => compatibleCapability,
probeSupervisor: () => presentSupervisor,
});
expect(result).toBe(true);
});
it('uses the real default probes when no deps are injected (does not throw)', () => {
// No live broker / built CLI is guaranteed in a test environment, so this
// only asserts the predicate degrades to a safe boolean rather than
// throwing — the fail-closed behavior itself is covered by the injected
// cases above.
expect(() => leaseEnforcementActivatable()).not.toThrow();
expect(typeof leaseEnforcementActivatable()).toBe('boolean');
});
});
describe('defaultCapabilityProbe', () => {
it('returns null (fail-closed) when no built CLI artifact is resolvable', () => {
// Deterministic regardless of ambient host state (e.g. a host that has
// already run `pnpm build`, which would otherwise make this pass or fail
// depending on whether dist/cli.js happens to exist) — inject a resolver
// pointing at a path that cannot exist, rather than relying on this
// checkout being unbuilt. The probe must report "no capability" rather
// than fabricate one from source-tree presence — this is the exact
// distinction #828's version skew needed: source existing is not the
// same as the published artifact advertising the capability.
const result = defaultCapabilityProbe({
resolveCliEntry: () => '/nonexistent/mosaic-lease-activation-probe-test/cli.js',
});
expect(result).toBeNull();
});
describe('positive path — injected resolver, isolated scratch dir (never the real dist/)', () => {
// A prior version of this test staged the stub cli.js at the package's
// REAL resolved dist/ path and relied on afterEach to clean up "only
// what it created" — which meant a host with a real pre-built
// dist/cli.js (ordinary `pnpm build && pnpm test`) would have its real
// ~26KB compiled CLI silently overwritten by an 87-byte stub, with no
// restoration of the original content. That is exactly the kind of
// build-artifact corruption #869 exists to prevent. This version uses
// dependency injection exclusively: defaultCapabilityProbe() is never
// called with its default resolver here, so it can never touch the real
// package dist/ at all — proven below by asserting that path's
// existence is unchanged by the test.
it('returns the real {name, version} capability from a stub cli.js in a temp dir, and leaves the real dist/ untouched', () => {
const packageRoot = join(dirname(fileURLToPath(import.meta.url)), '..', '..');
const realDistDir = join(packageRoot, 'dist');
const realDistPreexisted = existsSync(realDistDir);
const scratchDir = mkdtempSync(join(tmpdir(), 'mosaic-lease-capability-probe-'));
try {
const scratchCliPath = join(scratchDir, 'cli.js');
// Minimal stand-in for the built CLI's hidden __lease-capability
// subcommand — prints exactly what registerLeaseCapabilityProbe()
// wires the real `mosaic __lease-capability` command to print.
writeFileSync(
scratchCliPath,
`process.stdout.write(JSON.stringify(${JSON.stringify(LEASE_ACTIVATION_CAPABILITY)}));\n`,
);
const result = defaultCapabilityProbe({ resolveCliEntry: () => scratchCliPath });
expect(result).toEqual(LEASE_ACTIVATION_CAPABILITY);
// The real package dist/ must be byte-for-byte untouched: this test
// never invokes the default resolver, so the path's mere existence
// (created or not) must be unchanged by having run this test.
expect(existsSync(realDistDir)).toBe(realDistPreexisted);
} finally {
rmSync(scratchDir, { recursive: true, force: true });
}
});
});
});
describe('defaultResolveCliEntry', () => {
it('resolves the bare "@mosaicstack/mosaic" specifier (the exported "." entry), never the non-exported "./package.json" subpath', () => {
// Fully isolated from the real filesystem/package state (no dependency
// on whether @mosaicstack/mosaic has been built on this host) via an
// injected fake resolver that mirrors Node's real behavior: the "."
// export resolves fine, but "./package.json" is NOT in package.json's
// `exports` map, so real `require.resolve` throws
// ERR_PACKAGE_PATH_NOT_EXPORTED for it. This is genuinely red-first
// against the reviewer-found bug: the old implementation resolved the
// "./package.json" subpath here, which this fake throws on — the new
// implementation must resolve only the bare specifier.
const requestedSpecifiers: string[] = [];
const fakeResolve = (specifier: string): string => {
requestedSpecifiers.push(specifier);
if (specifier === '@mosaicstack/mosaic') return '/fake/pkg/dist/index.js';
throw new Error(`ERR_PACKAGE_PATH_NOT_EXPORTED: ${specifier}`);
};
const result = defaultResolveCliEntry(fakeResolve);
expect(result).toBe(join('/fake/pkg/dist', 'cli.js'));
expect(requestedSpecifiers).toEqual(['@mosaicstack/mosaic']);
});
});
describe('defaultSupervisorProbe', () => {
it('returns a well-shaped result without starting or connecting to anything', () => {
const result = defaultSupervisorProbe({});
expect(typeof result.supervisorPresent).toBe('boolean');
expect(result.socketPath === null || typeof result.socketPath === 'string').toBe(true);
});
it('resolves a socket path from an explicit MOSAIC_LEASE_BROKER_SOCKET override', () => {
const result = defaultSupervisorProbe({ MOSAIC_LEASE_BROKER_SOCKET: '/tmp/explicit.sock' });
expect(result.socketPath).toBe('/tmp/explicit.sock');
});
});
describe('registerLeaseCapabilityProbe', () => {
it('registers a hidden subcommand named __lease-capability', () => {
const program = new Command();
program.exitOverride();
registerLeaseCapabilityProbe(program);
const registered = program.commands.find((c) => c.name() === LEASE_CAPABILITY_PROBE_COMMAND);
expect(registered).toBeDefined();
// Commander exposes "hidden" only as help-output suppression (no public
// getter) — assert the observable behavior instead of a private field.
expect(program.helpInformation()).not.toContain(LEASE_CAPABILITY_PROBE_COMMAND);
});
it('prints the capability constant as JSON when invoked', () => {
const program = new Command();
program.exitOverride();
registerLeaseCapabilityProbe(program);
let written = '';
const originalWrite = process.stdout.write.bind(process.stdout);
process.stdout.write = ((chunk: string) => {
written += chunk;
return true;
}) as typeof process.stdout.write;
try {
program.parse(['node', 'mosaic', LEASE_CAPABILITY_PROBE_COMMAND]);
} finally {
process.stdout.write = originalWrite;
}
expect(JSON.parse(written)).toEqual(LEASE_ACTIVATION_CAPABILITY);
});
});

View File

@@ -0,0 +1,232 @@
/**
* Lease-enforcement activation probe (issue #869, Point-1 card C1).
*
* Root cause this exists to guard against (#828 version skew): the
* ENFORCEMENT half of the lease broker (PreToolUse/Stop hooks —
* `mutator-gate.py`, `receipt-observer-client.py` — wired via the framework
* reseed) and the ACTIVATION half (`execLeaseGatedRuntime()` in `launch.ts`,
* which chains the runtime through `launch-runtime.py`, injects
* `MOSAIC_LEASE_*`, and requires a running `daemon.py` broker) ship on
* different channels. When the published CLI tarball lags behind an
* enforcement reseed, the gate correctly fails CLOSED on absent identity —
* but every tool call then denies with GATE_UNAVAILABLE. That fail-closed
* behavior is intentional and must not change (see the C-REGRESS note in
* `runtime_tools_unittest.py`); this module exists so a downstream
* install-ordering guard (C2, out of scope here) can refuse to WIRE
* enforcement in the first place on a host that cannot ACTIVATE it.
*
* `leaseEnforcementActivatable()` answers one narrow question: "if
* enforcement were wired right now, could activation actually satisfy it?"
* It is a real capability probe — not a "does the source file exist" check
* — and both of its inputs are injectable so tests can drive every branch
* without a live broker or an installed CLI on PATH.
*/
import { execFileSync } from 'node:child_process';
import { existsSync } from 'node:fs';
import { createRequire } from 'node:module';
import { dirname, join } from 'node:path';
import type { Command } from 'commander';
import { defaultLeaseBrokerSocket, resolveTool } from './launch.js';
// ─── Capability signal (owned by the activation half) ──────────────────────
/**
* Versioned identity for the activation contract `execLeaseGatedRuntime()`
* implements. OWNED by the activation half of the lease broker. Bump
* `version` only when the activation contract itself changes (env vars
* injected, chaining behavior, socket protocol, etc.) — deliberately
* independent of the package's npm semver, because #828 happened precisely
* because the npm version was NOT bumped even though the shipped artifact
* fell out of sync. A build that cannot advertise this exact
* `{ name, version }` pair does not implement the contract a caller is
* relying on, whatever its package.json claims.
*/
export interface LeaseActivationCapability {
readonly name: string;
readonly version: number;
}
export const LEASE_ACTIVATION_CAPABILITY: LeaseActivationCapability = {
name: 'lease-runtime-activation',
version: 1,
};
/** Hidden CLI probe subcommand name — wired via {@link registerLeaseCapabilityProbe}. */
export const LEASE_CAPABILITY_PROBE_COMMAND = '__lease-capability';
function capabilityMatches(candidate: LeaseActivationCapability | null): boolean {
return (
candidate !== null &&
candidate.name === LEASE_ACTIVATION_CAPABILITY.name &&
candidate.version === LEASE_ACTIVATION_CAPABILITY.version
);
}
/**
* Register the hidden `__lease-capability` probe subcommand. Prints the
* capability this BUILD advertises as compact JSON to stdout and exits 0.
* Deliberately undocumented (hidden from `--help`): it is an internal signal
* for {@link defaultCapabilityProbe}, not a user-facing command.
*/
export function registerLeaseCapabilityProbe(program: Command): void {
program
.command(LEASE_CAPABILITY_PROBE_COMMAND, { hidden: true })
.description('Internal: print the lease-activation capability this build advertises')
.action(() => {
process.stdout.write(JSON.stringify(LEASE_ACTIVATION_CAPABILITY));
});
}
/** Injectable Node module resolver — matches `require.resolve`'s signature
* narrowly (specifier in, absolute path out, or throws). Defaults to the
* real `createRequire(import.meta.url).resolve`. Injectable so tests can
* exercise WHICH specifier {@link defaultResolveCliEntry} resolves (the
* reviewer-found bug was resolving the wrong one) without depending on
* whether `@mosaicstack/mosaic` has actually been built on the test host —
* and without ever touching the real package's `dist/` to find out. */
export type ModuleResolver = (specifier: string) => string;
/**
* Resolve the CLI's built entrypoint (`dist/cli.js`). Resolves via the
* package's "." export (already present in package.json's `exports` map)
* rather than a "./package.json" subpath — the latter is NOT exported, so
* `require.resolve('@mosaicstack/mosaic/package.json')` throws
* ERR_PACKAGE_PATH_NOT_EXPORTED on every real install. The "." export
* resolves to `dist/index.js`; `cli.js` is its sibling in the same built
* `dist/` directory (see package.json's `bin.mosaic`).
*
* Exported standalone (and injectable via {@link CapabilityProbeDeps}) so
* tests can exercise this resolution logic in isolation, or point
* {@link defaultCapabilityProbe} at a scratch directory instead of ever
* touching the real installed package's `dist/` — a test corrupting a real
* build artifact is exactly the artifact-integrity failure class this card
* exists to prevent (#828).
*/
export function defaultResolveCliEntry(
resolve: ModuleResolver = createRequire(import.meta.url).resolve,
): string {
const mainEntry = resolve('@mosaicstack/mosaic');
return join(dirname(mainEntry), 'cli.js');
}
/** Injectable inputs for {@link defaultCapabilityProbe}. */
export interface CapabilityProbeDeps {
/** Resolve the CLI entrypoint (`cli.js`) to probe. Defaults to
* {@link defaultResolveCliEntry}. Inject to point at an isolated scratch
* location in tests — never at the real package's `dist/`. */
resolveCliEntry?: () => string;
}
/**
* Real capability lookup. Resolves the installed `@mosaicstack/mosaic`
* package's BUILT entrypoint (`dist/cli.js` — the published artifact a user
* actually runs, not this TypeScript source file) and executes its hidden
* `__lease-capability` probe subcommand out-of-process. A build that lacks
* the subcommand, fails to execute, or reports an incompatible
* `{ name, version }` is treated as having NO activation capability.
*
* This is the check that would have caught #828's version skew: the
* source-tree activation half existed, but the published tarball's `dist/`
* did not carry it, so this probe — reading the actually-resolvable built
* artifact rather than trusting source-tree presence — would report null.
*/
export function defaultCapabilityProbe(
deps: CapabilityProbeDeps = {},
): LeaseActivationCapability | null {
try {
const resolveCliEntry = deps.resolveCliEntry ?? defaultResolveCliEntry;
const cliEntry = resolveCliEntry();
if (!existsSync(cliEntry)) return null;
const output = execFileSync(process.execPath, [cliEntry, LEASE_CAPABILITY_PROBE_COMMAND], {
encoding: 'utf-8',
timeout: 2000,
stdio: ['ignore', 'pipe', 'ignore'],
});
const parsed: unknown = JSON.parse(output);
if (
typeof parsed !== 'object' ||
parsed === null ||
typeof (parsed as Record<string, unknown>)['name'] !== 'string' ||
typeof (parsed as Record<string, unknown>)['version'] !== 'number'
) {
return null;
}
const candidate = parsed as { name: string; version: number };
return { name: candidate.name, version: candidate.version };
} catch {
return null;
}
}
// ─── Supervisor / socket resolution (detection only) ───────────────────────
/** Detection-only supervisor/socket probe result. Never starts the broker
* and never connects to the socket — presence and path resolution only. */
export interface SupervisorProbeResult {
/** The lease-broker supervisor artifacts (launcher + daemon) are present. */
readonly supervisorPresent: boolean;
/** Resolved broker socket path, or null if it could not be resolved. */
readonly socketPath: string | null;
}
/**
* Real supervisor/socket resolution: checks that the lease-broker's launcher
* (`launch-runtime.py`) and supervisor (`daemon.py`) artifacts resolve on
* disk via the same tool-resolution `execLeaseGatedRuntime()` uses, and that
* a broker socket path resolves via the same logic as
* `defaultLeaseBrokerSocket()`. Detection only — this never starts the
* daemon and never connects to the socket.
*/
export function defaultSupervisorProbe(
env: NodeJS.ProcessEnv = process.env,
): SupervisorProbeResult {
const launcherPath = resolveTool('lease-broker', 'launch-runtime.py');
const daemonPath = resolveTool('lease-broker', 'daemon.py');
const supervisorPresent = existsSync(launcherPath) && existsSync(daemonPath);
let socketPath: string | null = null;
try {
const resolved = defaultLeaseBrokerSocket(env);
socketPath = resolved.trim().length > 0 ? resolved : null;
} catch {
socketPath = null;
}
return { supervisorPresent, socketPath };
}
// ─── Predicate ───────────────────────────────────────────────────────────
/** Injectable inputs for {@link leaseEnforcementActivatable}, so tests (and
* downstream callers such as the C2 install-ordering guard) can drive every
* branch without a live broker or an installed CLI on PATH. */
export interface ActivationProbeDeps {
getCapability?: () => LeaseActivationCapability | null;
probeSupervisor?: () => SupervisorProbeResult;
}
/**
* True IFF lease enforcement can actually be ACTIVATED on this host:
*
* (a) the resolvable CLI advertises a {@link LeaseActivationCapability}
* compatible with {@link LEASE_ACTIVATION_CAPABILITY}, AND
* (b) the broker supervisor is resolvable — launcher + `daemon.py`
* artifacts present AND a broker socket path resolves.
*
* Pure/testable: both probes default to the real, side-effect-free lookups
* above but can be injected, so this predicate never itself starts a broker
* or performs enforcement — it only reports whether activation *could*
* satisfy enforcement if wired.
*/
export function leaseEnforcementActivatable(deps: ActivationProbeDeps = {}): boolean {
const getCapability = deps.getCapability ?? defaultCapabilityProbe;
const probeSupervisor = deps.probeSupervisor ?? defaultSupervisorProbe;
if (!capabilityMatches(getCapability())) return false;
const supervisor = probeSupervisor();
return supervisor.supervisorPresent && supervisor.socketPath !== null;
}

View File

@@ -0,0 +1,196 @@
import { describe, expect, it } from 'vitest';
import {
detectEnforcementHooksWired,
runLeaseEnforcementDoctorCheck,
} from './lease-doctor-check.js';
/**
* Red-first tests for issue #869 Point-1 C5 — the `mosaic doctor`
* lease-enforcement surfacing check.
*
* Root cause under test: enforcement hooks (`mutator-gate.py`,
* `receipt-observer-client.py`) can be wired into `~/.claude/settings.json`
* on a host where C1's `leaseEnforcementActivatable()` is false and/or C3's
* `checkBrokerSupervisorHealth()` reports unhealthy. That combination fails
* closed correctly, but must be surfaced LOUDLY by `mosaic doctor` rather
* than silently passing — this test suite exercises the three primary
* branches (wired+not-activatable, wired+healthy, not-wired) plus the
* broker-unhealthy variant.
*
* Every dependency is injected — no real `~/.claude/settings.json` and no
* real broker are ever touched.
*/
const WIRED_SETTINGS_JSON = JSON.stringify({
hooks: {
PreToolUse: [
{
matcher: '.*',
hooks: [
{
type: 'command',
command: 'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude',
},
],
},
],
Stop: [
{
hooks: [
{
type: 'command',
command:
'python3 ~/.config/mosaic/tools/lease-broker/receipt-observer-client.py --runtime claude --latest-entry',
},
],
},
],
},
});
const UNWIRED_SETTINGS_JSON = JSON.stringify({
hooks: {
PostToolUse: [
{
matcher: 'Edit|MultiEdit|Write',
hooks: [{ type: 'command', command: '~/.config/mosaic/tools/qa/qa-hook-stdin.sh' }],
},
],
},
});
describe('detectEnforcementHooksWired', () => {
it('detects the mutator-gate + receipt-observer markers when wired', () => {
const result = detectEnforcementHooksWired(JSON.parse(WIRED_SETTINGS_JSON));
expect(result.wired).toBe(true);
expect(result.matchedMarkers).toEqual(
expect.arrayContaining(['mutator-gate.py', 'receipt-observer-client.py']),
);
});
it('reports not wired when no enforcement markers are present', () => {
const result = detectEnforcementHooksWired(JSON.parse(UNWIRED_SETTINGS_JSON));
expect(result.wired).toBe(false);
expect(result.matchedMarkers).toEqual([]);
});
it('reports not wired for an empty settings object', () => {
expect(detectEnforcementHooksWired({}).wired).toBe(false);
});
it('detects wiring from just ONE marker (partial wiring is still dangerous)', () => {
const onlyMutatorGate = JSON.stringify({
hooks: {
PreToolUse: [
{
hooks: [{ type: 'command', command: 'python3 .../mutator-gate.py --runtime claude' }],
},
],
},
});
const result = detectEnforcementHooksWired(JSON.parse(onlyMutatorGate));
expect(result.wired).toBe(true);
expect(result.matchedMarkers).toEqual(['mutator-gate.py']);
});
});
describe('runLeaseEnforcementDoctorCheck', () => {
it('RED: wired + not-activatable ⇒ LOUD error (not a silent pass)', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => WIRED_SETTINGS_JSON,
isActivatable: () => false,
isBrokerHealthy: async () => true,
});
expect(result.status).toBe('error');
expect(result.wired).toBe(true);
expect(result.activatable).toBe(false);
expect(result.message).toMatch(/activation absent/);
expect(result.message).toMatch(/#869/);
expect(result.message.toLowerCase()).toMatch(/brick/);
});
it('wired + activatable + broker-unhealthy ⇒ LOUD error', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => WIRED_SETTINGS_JSON,
isActivatable: () => true,
isBrokerHealthy: async () => false,
});
expect(result.status).toBe('error');
expect(result.wired).toBe(true);
expect(result.brokerHealthy).toBe(false);
expect(result.message).toMatch(/broker not healthy/);
});
it('wired + not-activatable + broker-unhealthy ⇒ LOUD error citing both reasons', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => WIRED_SETTINGS_JSON,
isActivatable: () => false,
isBrokerHealthy: async () => false,
});
expect(result.status).toBe('error');
expect(result.message).toMatch(/activation absent/);
expect(result.message).toMatch(/broker not healthy/);
});
it('GREEN: wired + activatable + broker-healthy ⇒ ok', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => WIRED_SETTINGS_JSON,
isActivatable: () => true,
isBrokerHealthy: async () => true,
});
expect(result.status).toBe('ok');
expect(result.wired).toBe(true);
expect(result.activatable).toBe(true);
expect(result.brokerHealthy).toBe(true);
});
it('GREEN: not-wired ⇒ ok, no false alarm (activation/broker never probed)', async () => {
let activatableCalled = false;
let brokerCalled = false;
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => UNWIRED_SETTINGS_JSON,
isActivatable: () => {
activatableCalled = true;
return false;
},
isBrokerHealthy: async () => {
brokerCalled = true;
return false;
},
});
expect(result.status).toBe('ok');
expect(result.wired).toBe(false);
expect(result.activatable).toBeNull();
expect(result.brokerHealthy).toBeNull();
// Not wired must short-circuit — never even consult activation/broker.
expect(activatableCalled).toBe(false);
expect(brokerCalled).toBe(false);
});
it('GREEN: settings.json absent ⇒ ok (never touches a real file — readSettingsRaw is injected)', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => null,
isActivatable: () => false,
isBrokerHealthy: async () => false,
});
expect(result.status).toBe('ok');
expect(result.wired).toBe(false);
});
it('GREEN: malformed settings.json ⇒ ok (parse errors are not this cards failure class)', async () => {
const result = await runLeaseEnforcementDoctorCheck({
readSettingsRaw: () => '{ not valid json',
isActivatable: () => false,
isBrokerHealthy: async () => false,
});
expect(result.status).toBe('ok');
});
});

View File

@@ -0,0 +1,210 @@
/**
* Lease-enforcement doctor check (issue #869, Point-1 card C5).
*
* Root cause this guards against (#828 version skew, the same one C1/C3
* exist for): the Claude Code enforcement hooks (`mutator-gate.py` gating
* PreToolUse, `receipt-observer-client.py` observing Stop) can be WIRED into
* `~/.claude/settings.json` on a host where the ACTIVATION half is absent —
* no compatible CLI build (C1's `leaseEnforcementActivatable()`), or no
* healthy broker supervisor (C3's `checkBrokerSupervisorHealth()`). That
* combination is a silent brick: every gated tool call denies with
* GATE_UNAVAILABLE, and the fail-closed behavior is *correct* — but nothing
* surfaces it to an operator running `mosaic doctor` on an already-bricked
* host.
*
* This module answers one question — "if I ran right now, would I be
* bricked?" — by combining:
*
* 1. wiring detection: does `~/.claude/settings.json` reference either
* enforcement-hook marker (`mutator-gate.py` / `receipt-observer-client.py`)?
* 2. C1's `leaseEnforcementActivatable()` — could activation satisfy
* enforcement if it were exercised right now?
* 3. C3's `checkBrokerSupervisorHealth()` — is the broker supervisor
* actually healthy?
*
* Not wired ⇒ ok (nothing to activate, no false alarm). Wired AND activatable
* AND broker-healthy ⇒ ok. Wired AND (NOT activatable OR broker unhealthy) ⇒
* a LOUD, actionable error — this module never silently passes that state.
*
* Every dependency (settings read, activation probe, broker-health check) is
* injectable so tests can drive every branch without ever touching a real
* `~/.claude/settings.json` or a real broker.
*/
import { readFileSync } from 'node:fs';
import { homedir } from 'node:os';
import { join } from 'node:path';
import { leaseEnforcementActivatable, type ActivationProbeDeps } from './lease-activation-probe.js';
import {
checkBrokerSupervisorHealth,
resolveBrokerSupervisorPaths,
} from '../lease-broker/broker-supervisor.js';
import { DEFAULT_MOSAIC_HOME } from '../constants.js';
/** Markers identifying the two enforcement-hook halves wired via the
* framework reseed. Either marker's presence in `settings.json` means
* enforcement is wired — a host can be bricked with just one half present. */
const ENFORCEMENT_HOOK_MARKERS = ['mutator-gate.py', 'receipt-observer-client.py'] as const;
export interface EnforcementHooksWiredResult {
readonly wired: boolean;
readonly matchedMarkers: readonly string[];
}
/**
* Detect whether the Claude Code enforcement hooks (mutator-gate /
* receipt-observer) are wired into an already-parsed `settings.json`.
* Pure/testable — takes parsed JSON, never touches the filesystem itself.
*/
export function detectEnforcementHooksWired(settings: unknown): EnforcementHooksWiredResult {
const serialized = JSON.stringify(settings ?? {});
const matchedMarkers = ENFORCEMENT_HOOK_MARKERS.filter((marker) => serialized.includes(marker));
return { wired: matchedMarkers.length > 0, matchedMarkers };
}
export interface LeaseDoctorCheckDeps {
/**
* Read raw `settings.json` text; return `null` if the file is absent.
* Defaults to reading the real `~/.claude/settings.json`. ALWAYS inject a
* fake in tests — never point this at a real host's settings file.
*/
readSettingsRaw?: () => string | null;
/** Defaults to {@link leaseEnforcementActivatable} (C1). Inject for tests. */
isActivatable?: (deps?: ActivationProbeDeps) => boolean;
/**
* Defaults to a real broker-supervisor health check (C3) rooted at
* `mosaicHome`. Inject for tests — never point this at a real broker.
*/
isBrokerHealthy?: () => Promise<boolean>;
/** Mosaic home used to resolve default broker-supervisor paths. Defaults to
* `$MOSAIC_HOME` or `~/.config/mosaic`. */
mosaicHome?: string;
}
export type LeaseDoctorCheckStatus = 'ok' | 'error';
export interface LeaseDoctorCheckResult {
readonly status: LeaseDoctorCheckStatus;
readonly wired: boolean;
/** `null` when hooks are not wired (activation/broker were never probed). */
readonly activatable: boolean | null;
/** `null` when hooks are not wired (activation/broker were never probed). */
readonly brokerHealthy: boolean | null;
readonly message: string;
}
function defaultReadSettingsRaw(): string | null {
const settingsPath = join(homedir(), '.claude', 'settings.json');
try {
return readFileSync(settingsPath, 'utf8');
} catch (error) {
if (isEnoent(error)) return null;
throw error;
}
}
function isEnoent(error: unknown): boolean {
return (
typeof error === 'object' &&
error !== null &&
'code' in error &&
(error as NodeJS.ErrnoException).code === 'ENOENT'
);
}
function defaultMosaicHome(): string {
return process.env['MOSAIC_HOME'] ?? DEFAULT_MOSAIC_HOME;
}
async function defaultIsBrokerHealthy(mosaicHome: string): Promise<boolean> {
// `frameworkRoot` only feeds SOURCE paths (unit/wrapper/daemon file
// locations for `applyBrokerSupervisor`); the health check only reads
// TARGET paths (`unitTargetPath`, `socketPath`), both derived from
// `mosaicHome`/`homeDir`/`env` alone. Passing `mosaicHome` again here is
// therefore safe and never resolves or touches a framework checkout.
const paths = resolveBrokerSupervisorPaths({ mosaicHome, frameworkRoot: mosaicHome });
return (await checkBrokerSupervisorHealth(paths)).healthy;
}
/**
* Surface the #869 fail-closed brick scenario as a LOUD `mosaic doctor`
* error. See module docstring for the full decision table.
*/
export async function runLeaseEnforcementDoctorCheck(
deps: LeaseDoctorCheckDeps = {},
): Promise<LeaseDoctorCheckResult> {
const readSettingsRaw = deps.readSettingsRaw ?? defaultReadSettingsRaw;
const mosaicHome = deps.mosaicHome ?? defaultMosaicHome();
const isActivatable = deps.isActivatable ?? leaseEnforcementActivatable;
const isBrokerHealthy = deps.isBrokerHealthy ?? (() => defaultIsBrokerHealthy(mosaicHome));
const raw = readSettingsRaw();
if (raw === null) {
return {
status: 'ok',
wired: false,
activatable: null,
brokerHealthy: null,
message: 'Claude Code settings.json not found — lease-enforcement hooks not wired.',
};
}
let parsed: unknown;
try {
parsed = JSON.parse(raw);
} catch {
// Malformed settings.json is a different failure class than this card
// owns (C2 guards install-time writes); report ok rather than
// misattributing a parse error to the #869 activation gap.
return {
status: 'ok',
wired: false,
activatable: null,
brokerHealthy: null,
message:
'Claude Code settings.json could not be parsed — skipping lease-enforcement wiring check.',
};
}
const { wired, matchedMarkers } = detectEnforcementHooksWired(parsed);
if (!wired) {
return {
status: 'ok',
wired: false,
activatable: null,
brokerHealthy: null,
message:
'Lease-enforcement hooks not wired in ~/.claude/settings.json — nothing to activate.',
};
}
const activatable = isActivatable();
const brokerHealthy = await isBrokerHealthy();
if (activatable && brokerHealthy) {
return {
status: 'ok',
wired: true,
activatable,
brokerHealthy,
message: `Lease-enforcement hooks wired (${matchedMarkers.join(', ')}) — activation capability present and broker healthy.`,
};
}
const reasons: string[] = [];
if (!activatable) reasons.push('activation absent (leaseEnforcementActivatable() is false)');
if (!brokerHealthy) {
reasons.push('broker not healthy (checkBrokerSupervisorHealth() reports unhealthy)');
}
return {
status: 'error',
wired: true,
activatable,
brokerHealthy,
message:
`Lease-enforcement hooks (${matchedMarkers.join(', ')}) are wired in ~/.claude/settings.json, but ${reasons.join(' and ')}. ` +
'Every gated tool call will fail closed and BRICK this agent (see #869). ' +
'Remediate by activating the lease-broker supervisor (systemd unit + socket) or by removing the enforcement hooks from ~/.claude/settings.json.',
};
}

View File

@@ -0,0 +1,41 @@
import { spawnSync } from 'node:child_process';
import { join } from 'node:path';
import { describe, expect, it } from 'vitest';
/**
* C-REGRESS (issue #869, Point-1) — proves the fail-closed gate is untouched
* by the C1 activation probe added alongside this test.
*
* `mutator-gate.py`'s fail-closed-on-absent-identity behavior is INTENTIONAL
* and TEST-LOCKED: #869 C1 gates the WIRING decision for enforcement (via
* `leaseEnforcementActivatable()`), it does not — and must not — touch the
* gate's own runtime denial behavior. This spec runs the two test-locked
* cases from `runtime_tools_unittest.py` directly (rather than merely
* re-asserting the same logic in TypeScript) so a regression in the actual
* Python gate is caught here too, not just documented in prose.
*/
const MUTATOR_GATE_DIR = new URL('.', import.meta.url).pathname;
const UNITTEST_FILE = join(MUTATOR_GATE_DIR, 'runtime_tools_unittest.py');
const LOCKED_TEST_CASES = [
'ExecutableEntrypointTest.test_gate_entrypoint_denies_when_identity_environment_is_absent',
'MutatorGateTest.test_environment_generation_and_request_failures_deny',
] as const;
describe('mutator-gate fail-closed behavior (C-REGRESS, unchanged by #869 C1)', () => {
it.each(LOCKED_TEST_CASES)('%s still passes', (testCase) => {
const result = spawnSync('python3', ['-m', 'unittest', `${moduleName()}.${testCase}`, '-v'], {
cwd: MUTATOR_GATE_DIR,
encoding: 'utf-8',
});
expect(result.status, `stderr:\n${result.stderr}`).toBe(0);
});
});
function moduleName(): string {
// runtime_tools_unittest.py, addressed as a bare module name for `python3 -m unittest`.
return UNITTEST_FILE.split('/').pop()!.replace(/\.py$/, '');
}

View File

@@ -47,6 +47,22 @@ const piLifecyclePath = join(frameworkRoot, 'runtime/pi/lease-lifecycle.ts');
const prdyInitPath = join(frameworkRoot, 'tools/prdy/prdy-init.sh'); const prdyInitPath = join(frameworkRoot, 'tools/prdy/prdy-init.sh');
const prdyUpdatePath = join(frameworkRoot, 'tools/prdy/prdy-update.sh'); const prdyUpdatePath = join(frameworkRoot, 'tools/prdy/prdy-update.sh');
const remediationHandlerPath = join(frameworkRoot, 'tools/qa/remediation-hook-handler.sh'); const remediationHandlerPath = join(frameworkRoot, 'tools/qa/remediation-hook-handler.sh');
// C4 (#869 Point-1): launch-runtime.py now asserts, before anything else,
// that the CLI's advertised lease-activation capability (normally read via
// the hidden `mosaic __lease-capability` subcommand) matches what
// enforcement expects — see framework/tools/lease-broker/
// activation_version_gate.py. This suite drives launch-runtime.py directly
// as a subprocess (never through the real `mosaic` CLI), so — exactly like
// the fake broker (daemon.py) and fake `claude` binaries already used
// below — it must supply a fake activation-capability probe rather than
// depend on a real `mosaic` binary being on PATH. `MOSAIC_LEASE_VERSION_PROBE_COMMAND`
// is launch-runtime.py's injection point for that fake; this literal
// {name, version} pair must be kept in sync with
// `EXPECTED_ACTIVATION_CAPABILITY` (activation_version_gate.py) and
// `LEASE_ACTIVATION_CAPABILITY` (lease-activation-probe.ts) — all three
// currently agree on v1.
const leaseCapabilityProbeStub = `python3 -c "import json; print(json.dumps({'name': 'lease-runtime-activation', 'version': 1}))"`;
const children: ChildProcess[] = []; const children: ChildProcess[] = [];
const temporaryRoots: string[] = []; const temporaryRoots: string[] = [];
@@ -184,6 +200,7 @@ raise SystemExit(0 if len(session_id) == 64 and denied else 1)
MOSAIC_PRDY_RUNTIME: 'claude', MOSAIC_PRDY_RUNTIME: 'claude',
MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_LEASE_BROKER_SOCKET: socket,
MOSAIC_RUNTIME_GENERATION: '1', MOSAIC_RUNTIME_GENERATION: '1',
MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub,
}, },
}); });
} }
@@ -743,6 +760,7 @@ describe('whole mutator-class lease gate', () => {
...process.env, ...process.env,
MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_LEASE_BROKER_SOCKET: socket,
MOSAIC_RUNTIME_GENERATION: '1', MOSAIC_RUNTIME_GENERATION: '1',
MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub,
}, },
}, },
); );
@@ -761,6 +779,7 @@ describe('whole mutator-class lease gate', () => {
...process.env, ...process.env,
MOSAIC_LEASE_BROKER_SOCKET: join(tmpdir(), 'missing-mosaic-broker.sock'), MOSAIC_LEASE_BROKER_SOCKET: join(tmpdir(), 'missing-mosaic-broker.sock'),
MOSAIC_RUNTIME_GENERATION: '1', MOSAIC_RUNTIME_GENERATION: '1',
MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub,
}, },
}, },
); );
@@ -878,6 +897,7 @@ raise SystemExit(0 if len(session_id) == 64 and hook_present and observers_prese
PATH: `${binDir}:${process.env.PATH ?? ''}`, PATH: `${binDir}:${process.env.PATH ?? ''}`,
MOSAIC_LEASE_BROKER_SOCKET: socket, MOSAIC_LEASE_BROKER_SOCKET: socket,
MOSAIC_RUNTIME_GENERATION: '1', MOSAIC_RUNTIME_GENERATION: '1',
MOSAIC_LEASE_VERSION_PROBE_COMMAND: leaseCapabilityProbeStub,
}, },
proxyGate: () => proxyGate: () =>
Promise.resolve({ Promise.resolve({

View File

@@ -39,6 +39,18 @@ LAUNCHER = load_tool("lease_runtime_launcher", "launch-runtime.py")
GATE = load_tool("lease_mutator_gate", "mutator-gate.py") GATE = load_tool("lease_mutator_gate", "mutator-gate.py")
def matching_activation_probe(*_args: object, **_kwargs: object) -> dict[str, object]:
"""Fake activation-capability probe matching what enforcement expects
(C4, #869 Point-1). Injected into `LAUNCHER.main()` calls below that are
exercising OTHER branches (registration, exec, generation init, ...) so
the new version-coupling gate — which runs before those — never blocks
on host state (no real `mosaic` CLI on PATH in a test sandbox). The
version-coupling gate's OWN behavior (match/mismatch/absent) is covered
by its dedicated red-first tests in `version_coupling_unittest.py`."""
return dict(LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY)
class FakeSocket: class FakeSocket:
def __init__(self, *chunks: bytes): def __init__(self, *chunks: bytes):
self.chunks = list(chunks) self.chunks = list(chunks)
@@ -95,6 +107,7 @@ class LaunchRuntimeTest(unittest.TestCase):
request=request, request=request,
execute=execute, execute=execute,
initialize_generation=initialize_generation, initialize_generation=initialize_generation,
probe_activation_capability=matching_activation_probe,
) )
self.assertEqual(result, 0) self.assertEqual(result, 0)
@@ -127,6 +140,7 @@ class LaunchRuntimeTest(unittest.TestCase):
request=lambda *_args: {"ok": True, "session_id": "e" * 64}, request=lambda *_args: {"ok": True, "session_id": "e" * 64},
execute=lambda *args: executed.append(args), execute=lambda *args: executed.append(args),
initialize_generation=lambda *_args: None, initialize_generation=lambda *_args: None,
probe_activation_capability=matching_activation_probe,
) )
self.assertEqual(result, 0) self.assertEqual(result, 0)
self.assertEqual( self.assertEqual(
@@ -153,6 +167,7 @@ class LaunchRuntimeTest(unittest.TestCase):
request=lambda *_args: {"ok": True, "session_id": "f" * 64}, request=lambda *_args: {"ok": True, "session_id": "f" * 64},
execute=lambda *args: executed.append(args), execute=lambda *args: executed.append(args),
initialize_generation=lambda *_args: None, initialize_generation=lambda *_args: None,
probe_activation_capability=matching_activation_probe,
) )
self.assertEqual(result, 0) self.assertEqual(result, 0)
self.assertEqual(executed[0][0:2], ("pi", ["pi", "--print", "hello"])) self.assertEqual(executed[0][0:2], ("pi", ["pi", "--print", "hello"]))
@@ -188,6 +203,7 @@ class LaunchRuntimeTest(unittest.TestCase):
environ=environment, environ=environment,
request=lambda *_args, value=reply: value, request=lambda *_args, value=reply: value,
execute=lambda *args: executed.append(args), execute=lambda *args: executed.append(args),
probe_activation_capability=matching_activation_probe,
) )
self.assertEqual(result, 1) self.assertEqual(result, 1)
self.assertEqual(executed, []) self.assertEqual(executed, [])
@@ -203,6 +219,7 @@ class LaunchRuntimeTest(unittest.TestCase):
initialize_generation=lambda *_args: (_ for _ in ()).throw( initialize_generation=lambda *_args: (_ for _ in ()).throw(
OSError("unsafe state") OSError("unsafe state")
), ),
probe_activation_capability=matching_activation_probe,
), ),
1, 1,
) )
@@ -220,6 +237,7 @@ class LaunchRuntimeTest(unittest.TestCase):
environ={"MOSAIC_LEASE_BROKER_SOCKET": "/x"}, environ={"MOSAIC_LEASE_BROKER_SOCKET": "/x"},
request=request, request=request,
execute=lambda *_args: self.fail("must not execute"), execute=lambda *_args: self.fail("must not execute"),
probe_activation_capability=matching_activation_probe,
), ),
1, 1,
) )
@@ -233,6 +251,7 @@ class LaunchRuntimeTest(unittest.TestCase):
request=lambda *_args: {"ok": True, "session_id": "c" * 64}, request=lambda *_args: {"ok": True, "session_id": "c" * 64},
execute=lambda *_args: (_ for _ in ()).throw(OSError("missing")), execute=lambda *_args: (_ for _ in ()).throw(OSError("missing")),
initialize_generation=lambda *_args: None, initialize_generation=lambda *_args: None,
probe_activation_capability=matching_activation_probe,
), ),
1, 1,
) )

View File

@@ -0,0 +1,287 @@
#!/usr/bin/env python3
"""Red-first tests for issue #869 Point-1 C4 — the enforcement/activation
version-coupling assertion at the `launch-runtime.py` seam.
Root cause under test (#828 restated): the lease broker's ENFORCEMENT half
(this toolkit) and its ACTIVATION half (`execLeaseGatedRuntime()` in
`launch.ts`, chained through `launch-runtime.py`) shipped on different
channels and drifted. C1 (`lease-activation-probe.ts`) gave the activation
half a versioned, machine-checkable identity
(`LEASE_ACTIVATION_CAPABILITY`, printed via the hidden CLI subcommand
`mosaic __lease-capability`). C4 (this module + `activation_version_gate.py`)
is the assertion that actually USES that identity: enforcement must refuse
to proceed — loudly, with an actionable remediation message, never a
silent pass — unless the activation capability it observes exactly matches
what enforcement expects.
Every case here drives the seam with injected fakes/stubs (a fake
`probe_activation_capability` callable at the `launch-runtime.py` level, or
a fake `run` transport at the `activation_version_gate` level) — never a
real broker, a real installed CLI, or a real `mosaic` binary on PATH.
"""
from __future__ import annotations
import importlib.util
import io
import subprocess
import sys
import unittest
from contextlib import redirect_stderr
from pathlib import Path
TOOLS_DIR = Path(__file__).parents[2] / "framework/tools/lease-broker"
if str(TOOLS_DIR) not in sys.path:
sys.path.insert(0, str(TOOLS_DIR))
def load_tool(module_name: str, filename: str):
spec = importlib.util.spec_from_file_location(module_name, TOOLS_DIR / filename)
if spec is None or spec.loader is None:
raise RuntimeError(f"unable to load {filename}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
# Loaded under distinct module names from runtime_tools_unittest.py's own
# LAUNCHER/GATE loads — importlib.util.module_from_spec() gives each load a
# fresh module object regardless of name collisions, but distinct names keep
# tracebacks/debugging unambiguous when both files run in the same process.
LAUNCHER = load_tool("lease_runtime_launcher_version_coupling", "launch-runtime.py")
VERSION_GATE = load_tool("lease_activation_version_gate_test", "activation_version_gate.py")
def matching_capability() -> dict[str, object]:
return dict(VERSION_GATE.EXPECTED_ACTIVATION_CAPABILITY)
class AssertActivationCapabilityMatchesTest(unittest.TestCase):
"""Unit-level coverage of `activation_version_gate.py`'s own assertion,
isolated from the launch-runtime.py seam it is wired into below."""
def test_matching_capability_passes_silently(self) -> None:
VERSION_GATE.assert_activation_capability_matches(matching_capability())
# No exception is the assertion; nothing further to check.
def test_absent_capability_fails_closed_not_silent_pass(self) -> None:
with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
VERSION_GATE.assert_activation_capability_matches(None)
message = str(raised.exception)
self.assertIn("#869", message)
self.assertIn("upgrade", message.lower())
def test_version_mismatch_message_is_actionable(self) -> None:
expected = {"name": "lease-runtime-activation", "version": 1}
mismatched = {"name": "lease-runtime-activation", "version": 2}
with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
VERSION_GATE.assert_activation_capability_matches(mismatched, expected)
message = str(raised.exception)
self.assertIn("v2", message)
self.assertIn("v1", message)
self.assertIn("#869", message)
self.assertIn("upgrade", message.lower())
self.assertIn("version skew", message.lower())
def test_name_mismatch_fails_loud(self) -> None:
expected = {"name": "lease-runtime-activation", "version": 1}
mismatched = {"name": "some-other-capability", "version": 1}
with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
VERSION_GATE.assert_activation_capability_matches(mismatched, expected)
message = str(raised.exception)
self.assertIn("some-other-capability", message)
self.assertIn("lease-runtime-activation", message)
self.assertIn("#869", message)
def test_reversed_drift_newer_activation_than_enforcement_expects_also_fails(self) -> None:
# A build/deploy where ACTIVATION shipped ahead of ENFORCEMENT is
# exactly as much version skew as the reverse (#828's actual shape
# was enforcement ahead of activation) — the assertion must not special
# case direction.
expected = {"name": "lease-runtime-activation", "version": 1}
newer_activation = {"name": "lease-runtime-activation", "version": 2}
with self.assertRaises(VERSION_GATE.VersionCouplingError):
VERSION_GATE.assert_activation_capability_matches(newer_activation, expected)
class ProbeActivationCapabilityTest(unittest.TestCase):
"""Coverage of the probe's command resolution and fail-closed transport
handling — never spawns a real `mosaic` process."""
def test_returns_none_when_mosaic_is_not_resolvable_on_path(self) -> None:
result = VERSION_GATE.default_probe_activation_capability(
{"PATH": "/nonexistent-bin-dir-for-869-c4-test"}
)
self.assertIsNone(result)
def test_override_command_is_parsed_and_the_probe_subcommand_is_not_double_appended(
self,
) -> None:
captured: list[list[str]] = []
class FakeCompleted:
returncode = 0
stdout = '{"name": "lease-runtime-activation", "version": 1}'
def fake_run(argv: list[str], **_kwargs: object) -> FakeCompleted:
captured.append(argv)
return FakeCompleted()
result = VERSION_GATE.default_probe_activation_capability(
{VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic __lease-capability"},
run=fake_run,
)
self.assertEqual(result, {"name": "lease-runtime-activation", "version": 1})
self.assertEqual(captured, [["/fake/mosaic", "__lease-capability"]])
def test_fails_closed_on_nonzero_exit_malformed_json_and_missing_fields(self) -> None:
class NonZeroExit:
returncode = 1
stdout = '{"name": "lease-runtime-activation", "version": 1}'
class MalformedOutput:
returncode = 0
stdout = "not-json"
class MissingVersion:
returncode = 0
stdout = '{"name": "lease-runtime-activation"}'
class WrongShapeVersion:
returncode = 0
stdout = '{"name": "lease-runtime-activation", "version": "1"}'
class BooleanVersion:
# bool is a subclass of int in Python; must not be accepted as
# a version number.
returncode = 0
stdout = '{"name": "lease-runtime-activation", "version": true}'
for fake in (
NonZeroExit(),
MalformedOutput(),
MissingVersion(),
WrongShapeVersion(),
BooleanVersion(),
):
with self.subTest(stdout=fake.stdout, returncode=fake.returncode):
result = VERSION_GATE.default_probe_activation_capability(
{VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"},
run=lambda *_a, fake=fake, **_kw: fake,
)
self.assertIsNone(result)
def test_fails_closed_on_timeout_and_transport_error(self) -> None:
def timeout_run(*_args: object, **_kwargs: object) -> None:
raise subprocess.TimeoutExpired(cmd="mosaic", timeout=2.0)
def oserror_run(*_args: object, **_kwargs: object) -> None:
raise OSError("no such file or directory")
for run_fake in (timeout_run, oserror_run):
with self.subTest(run=run_fake.__name__):
result = VERSION_GATE.default_probe_activation_capability(
{VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"},
run=run_fake,
)
self.assertIsNone(result)
class LaunchRuntimeVersionCouplingSeamTest(unittest.TestCase):
"""End-to-end (still fully faked) coverage of the seam as wired into
`launch-runtime.py`'s `main()` — the strongest natural enforcement point
per the C4 card, run before any broker registration."""
def _run(self, *, probe):
calls: dict[str, object] = {}
def request(_path: Path, payload: dict[str, object]) -> dict[str, object]:
calls["registered"] = True
calls["request"] = payload
return {"ok": True, "session_id": "a" * 64}
def execute(command: str, argv: list[str], environment: dict[str, str]) -> None:
calls["executed"] = (command, argv, environment)
def initialize_generation(_path: Path, _generation: int) -> None:
calls["generation_initialized"] = True
stderr = io.StringIO()
with redirect_stderr(stderr):
result = LAUNCHER.main(
["--runtime", "claude", "--", "claude", "--print", "hello"],
environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"},
request=request,
execute=execute,
initialize_generation=initialize_generation,
probe_activation_capability=probe,
)
return result, stderr.getvalue(), calls
def test_matching_activation_version_passes_and_the_gate_proceeds(self) -> None:
result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: matching_capability())
self.assertEqual(result, 0)
self.assertEqual(stderr_text, "")
self.assertTrue(calls.get("registered"))
self.assertIn("executed", calls)
def test_version_mismatch_fails_loud_denies_and_never_registers_or_execs(self) -> None:
expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY
mismatched = {"name": expected["name"], "version": expected["version"] + 1}
result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched)
self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
self.assertNotEqual(result, 0)
self.assertIn("#869", stderr_text)
self.assertIn(f"v{mismatched['version']}", stderr_text)
self.assertIn(f"v{expected['version']}", stderr_text)
self.assertIn("upgrade", stderr_text.lower())
# Never reaches broker registration or exec — the version gate is a
# hard stop, not advisory.
self.assertNotIn("registered", calls)
self.assertNotIn("executed", calls)
def test_name_mismatch_fails_loud(self) -> None:
expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY
mismatched = {"name": "some-other-capability", "version": expected["version"]}
result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched)
self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
self.assertIn("#869", stderr_text)
self.assertIn("some-other-capability", stderr_text)
self.assertNotIn("registered", calls)
self.assertNotIn("executed", calls)
def test_absent_activation_capability_fails_closed_not_a_silent_pass(self) -> None:
result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: None)
self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
self.assertNotEqual(result, 0)
self.assertIn("#869", stderr_text)
self.assertNotIn("registered", calls)
self.assertNotIn("executed", calls)
def test_version_gate_runs_before_and_independently_of_broker_registration(self) -> None:
def request_must_not_be_called(*_args: object, **_kwargs: object) -> dict[str, object]:
self.fail("broker must not be contacted when activation version is mismatched")
stderr = io.StringIO()
with redirect_stderr(stderr):
result = LAUNCHER.main(
["--runtime", "claude", "--", "claude"],
environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"},
request=request_must_not_be_called,
probe_activation_capability=lambda *_a, **_kw: None,
)
self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
def test_dedicated_exit_code_never_collides_with_usage_or_registration_codes(self) -> None:
# Distinctness guard: a version-skew denial must never be mistaken
# for the pre-existing usage error (64) or registration/exec
# fail-closed code (1) this script already owns.
self.assertNotIn(LAUNCHER.EXIT_VERSION_SKEW, (0, 1, 64))
if __name__ == "__main__":
unittest.main()

View File

@@ -13,22 +13,37 @@ import {
type SkillSyncResult as ClaudeSkillSyncResult, type SkillSyncResult as ClaudeSkillSyncResult,
} from '../commands/skill.js'; } from '../commands/skill.js';
function linkRuntimeAssets(mosaicHome: string, skipClaudeHooks: boolean): void { /**
* Link runtime assets. Returns a warning string when the install-ordering
* guard (#869 Point-1 C2) reported a degraded outcome — i.e. the
* lease-enforcement hooks were NOT wired into ~/.claude/settings.json because
* this host could not confirm it can activate them — so the caller can
* surface it via `p.warn(...)` instead of it being swallowed by `stdio:
* 'pipe'`. Non-fatal either way: the wizard always continues.
*/
function linkRuntimeAssets(mosaicHome: string, skipClaudeHooks: boolean): string | undefined {
const script = join(mosaicHome, 'bin', 'mosaic-link-runtime-assets'); const script = join(mosaicHome, 'bin', 'mosaic-link-runtime-assets');
if (existsSync(script)) { if (!existsSync(script)) return undefined;
try { try {
spawnSync('bash', [script], { const result = spawnSync('bash', [script], {
timeout: 30000, timeout: 30000,
stdio: 'pipe', stdio: 'pipe',
env: { encoding: 'utf-8',
...process.env, env: {
...(skipClaudeHooks ? { MOSAIC_SKIP_CLAUDE_HOOKS: '1' } : {}), ...process.env,
}, ...(skipClaudeHooks ? { MOSAIC_SKIP_CLAUDE_HOOKS: '1' } : {}),
}); },
} catch { });
// Non-fatal: wizard continues if (result.status !== 0) {
const stderr = (result.stderr ?? '').trim();
return (
stderr || 'Runtime asset linking reported a non-zero exit (see mosaic doctor for details).'
);
} }
} catch {
// Non-fatal: wizard continues
} }
return undefined;
} }
interface SyncSkillsResult { interface SyncSkillsResult {
@@ -201,7 +216,7 @@ export async function finalizeStage(
// copied into ~/.claude/ while still linking the other runtime files. // copied into ~/.claude/ while still linking the other runtime files.
spin.update('Linking runtime assets...'); spin.update('Linking runtime assets...');
const skipClaudeHooks = state.hooks?.accepted === false; const skipClaudeHooks = state.hooks?.accepted === false;
linkRuntimeAssets(state.mosaicHome, skipClaudeHooks); const linkWarning = linkRuntimeAssets(state.mosaicHome, skipClaudeHooks);
// 4. Sync skills (only installs the user-selected subset) // 4. Sync skills (only installs the user-selected subset)
let skillsResult: SyncSkillsResult = { success: true, installedCount: 0 }; let skillsResult: SyncSkillsResult = { success: true, installedCount: 0 };
@@ -236,6 +251,10 @@ export async function finalizeStage(
spin.stop('Installation complete'); spin.stop('Installation complete');
// Surface the install-ordering guard's outcome (#869 Point-1 C2) — never
// silent, even though the wizard continues either way.
if (linkWarning) p.warn(linkWarning);
// Report skill install failure clearly (non-fatal but user should know) // Report skill install failure clearly (non-fatal but user should know)
if (!skillsResult.success && skillsResult.failureReason) { if (!skillsResult.success && skillsResult.failureReason) {
p.warn(skillsResult.failureReason); p.warn(skillsResult.failureReason);