diff --git a/orchestrator/api.py b/orchestrator/api.py index e4cb70d53f..9a0bd8a6d6 100644 --- a/orchestrator/api.py +++ b/orchestrator/api.py @@ -48,6 +48,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from routes.metrics import metrics_bp from routes.phases import phases_bp from routes.pipelines import pipelines_bp + from routes.sdlc_tokens import sdlc_tokens_bp from routes.signals import signals_bp from webhooks import webhooks_bp @@ -59,6 +60,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] app.register_blueprint(decisions_bp) app.register_blueprint(metrics_bp) app.register_blueprint(webhooks_bp) + app.register_blueprint(sdlc_tokens_bp) except ImportError: from .routes.containers import containers_bp # type: ignore[no-redef] from .routes.decisions import decisions_bp # type: ignore[no-redef] @@ -66,6 +68,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from .routes.metrics import metrics_bp # type: ignore[no-redef] from .routes.phases import phases_bp # type: ignore[no-redef] from .routes.pipelines import pipelines_bp # type: ignore[no-redef] + from .routes.sdlc_tokens import sdlc_tokens_bp # type: ignore[no-redef] from .routes.signals import signals_bp # type: ignore[no-redef] from .webhooks import webhooks_bp # type: ignore[no-redef] @@ -77,6 +80,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] app.register_blueprint(decisions_bp) app.register_blueprint(metrics_bp) app.register_blueprint(webhooks_bp) + app.register_blueprint(sdlc_tokens_bp) @app.before_request diff --git a/orchestrator/models.py b/orchestrator/models.py index fb0179b5e9..747e4b1016 100644 --- a/orchestrator/models.py +++ b/orchestrator/models.py @@ -207,6 +207,9 @@ class Pipeline(BaseModel): ) updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update time") contract_synced: bool = Field(default=True, description="Whether state is synced with contract") + sdlc_token_gated: bool = Field( + default=False, description="Whether this pipeline requires token-gated approval" + ) error: str | None = Field(default=None, description="Error if failed") version: int = Field( default=1, ge=1, description="Optimistic locking version (incremented on each save)" diff --git a/orchestrator/routes/decisions.py b/orchestrator/routes/decisions.py index 8c6919a1be..9ee645afc9 100644 --- a/orchestrator/routes/decisions.py +++ b/orchestrator/routes/decisions.py @@ -35,7 +35,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] DecisionNotFoundError, get_decision_queue, ) -from state_store import InvalidPipelineIdError +from state_store import InvalidPipelineIdError, PipelineNotFoundError, get_state_store logger = get_logger("orchestrator.decisions") @@ -287,6 +287,25 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int] if not resolution: return make_error_response("Missing resolution") + # Check if pipeline is token-gated — block direct resolution if so + try: + store = get_state_store(repo_path) + pipeline = store.load_pipeline(pipeline_id) + if pipeline.sdlc_token_gated: + return make_error_response( + "This pipeline requires token-gated approval. " + "Type !approve in your terminal.", + status_code=403, + ) + except PipelineNotFoundError: + pass # Pipeline may not exist yet, allow resolution + except Exception: + logger.error("Failed to check token gate", pipeline_id=pipeline_id, exc_info=True) + return make_error_response( + "Unable to verify pipeline token gate. Try again.", + status_code=503, + ) + try: queue = get_decision_queue(pipeline_id, repo_path) decision = queue.resolve_decision(decision_id, resolution) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 7c11cd783c..a9659c07cf 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -414,6 +414,9 @@ def create_pipeline() -> tuple[Response, int]: repo = data.get("repo") branch = data.get("branch") + # Check for pre-generated SDLC tokens (set by entrypoint before Claude starts) + from routes.sdlc_tokens import has_tokens_for_pipeline + if not issue_number: return make_error_response("Missing issue_number") if not repo: @@ -436,6 +439,12 @@ def create_pipeline() -> tuple[Response, int]: # Contract creation is deferred to _run_pipeline so it writes # into the per-pipeline worktree instead of the main repo. + # Enable token gating if tokens were pre-generated for this pipeline + if has_tokens_for_pipeline(pipeline.id): + pipeline.sdlc_token_gated = True + store.save_pipeline(pipeline, commit=False) + logger.info("Pipeline token-gated", pipeline_id=pipeline.id) + logger.info( "Pipeline created", pipeline_id=pipeline.id, diff --git a/orchestrator/routes/sdlc_tokens.py b/orchestrator/routes/sdlc_tokens.py new file mode 100644 index 0000000000..c5369a49a2 --- /dev/null +++ b/orchestrator/routes/sdlc_tokens.py @@ -0,0 +1,333 @@ +""" +SDLC token-gated approval endpoints. + +Provides token generation and validation for SDLC pipeline phase approvals. +Tokens are stored in-memory (ephemeral, single-session lifetime). +""" + +import functools +import hashlib +import os +import secrets +import sys +from pathlib import Path +from typing import Any + +from flask import Blueprint, Response, jsonify, request + +# Add parent directory to path for imports +_parent_path = Path(__file__).parent.parent +if str(_parent_path) not in sys.path: + sys.path.insert(0, str(_parent_path)) + +# Add shared directory to path for logging +_shared_path = Path(__file__).parent.parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +try: + from egg_logging import get_logger +except ImportError: + import logging + + def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] + return logging.getLogger(name) + + +from sdlc_wordlist import WORD_LIST + +logger = get_logger("orchestrator.sdlc_tokens") + +sdlc_tokens_bp = Blueprint("sdlc_tokens", __name__, url_prefix="/api/v1/sdlc-tokens") + + +def _check_launcher_auth() -> tuple[bool, str]: + """Validate the launcher secret from the Authorization header. + + The /generate and /reset endpoints are privileged — only the entrypoint + (running as root before Claude starts) should call them. The launcher + secret is unavailable to the sandbox egg user, preventing Claude from + calling these endpoints directly. + """ + expected = os.environ.get("EGG_LAUNCHER_SECRET", "") + if not expected: + # If no secret is configured, deny all requests to privileged endpoints. + # This prevents accidental exposure in misconfigured deployments. + return False, "Launcher secret not configured" + + auth_header = request.headers.get("Authorization", "") + if not auth_header.startswith("Bearer "): + return False, "Missing or invalid Authorization header" + + provided = auth_header[7:] + if secrets.compare_digest(provided, expected): + return True, "" + + return False, "Invalid launcher authorization token" + + +def _require_launcher_auth(f: Any) -> Any: + """Decorator requiring launcher secret auth on privileged endpoints.""" + + @functools.wraps(f) + def decorated(*args: Any, **kwargs: Any) -> Any: + is_valid, error = _check_launcher_auth() + if not is_valid: + logger.warning( + "SDLC endpoint auth failed", + endpoint=request.path, + error=error, + source_ip=request.remote_addr, + ) + return _make_error(error, status_code=401) + return f(*args, **kwargs) + + return decorated + + +# In-memory token store: pipeline_id -> token data +# Tokens are ephemeral (one session lifetime); orchestrator is a singleton. +# CAVEAT: If the orchestrator restarts while a pipeline is token-gated, +# the in-memory tokens are lost but Pipeline.sdlc_token_gated remains True +# in persistent storage. Use the /reset endpoint to recover from this state. +_token_store: dict[str, dict[str, Any]] = {} + +# Only the first two PipelinePhase values (refine, plan) are token-gated. +# The implement and pr phases proceed automatically after plan approval. +# NOTE: If this set changes, also update the regex in sdlc-approve.sh. +VALID_PHASES = {"refine", "plan"} + + +def _generate_token() -> str: + """Generate a 3-word token like APPLE-HORSE-RIVER.""" + words = [secrets.choice(WORD_LIST) for _ in range(3)] + return "-".join(words) + + +def _hash_token(token: str) -> str: + """SHA-256 hash a token for storage.""" + return hashlib.sha256(token.upper().encode()).hexdigest() + + +def has_tokens_for_pipeline(pipeline_id: str) -> bool: + """Check if tokens have been generated for a pipeline.""" + return pipeline_id in _token_store + + +def _make_error(message: str, status_code: int = 400) -> tuple[Response, int]: + """Create an error response.""" + return jsonify({"success": False, "message": message}), status_code + + +def _make_success(message: str, data: dict[str, Any] | None = None) -> tuple[Response, int]: + """Create a success response.""" + response: dict[str, Any] = {"success": True, "message": message} + if data: + response["data"] = data + return jsonify(response), 200 + + +@sdlc_tokens_bp.route("/generate", methods=["POST"]) +@_require_launcher_auth +def generate_tokens() -> tuple[Response, int]: + """Generate approval tokens for an SDLC pipeline. + + Requires launcher secret authentication. Only callable by the entrypoint + (running as root) — not by Claude in the sandbox. + + Request body: + {"pipeline_id": "issue-596"} + + Returns plaintext tokens (one-time display to human). + """ + data = request.get_json() or {} + pipeline_id = data.get("pipeline_id") + + if not pipeline_id: + return _make_error("Missing pipeline_id") + + if pipeline_id in _token_store: + return _make_error(f"Tokens already generated for pipeline {pipeline_id}", 409) + + refine_token = _generate_token() + plan_token = _generate_token() + + # Ensure tokens are different + while plan_token == refine_token: + plan_token = _generate_token() + + _token_store[pipeline_id] = { + "refine_hash": _hash_token(refine_token), + "plan_hash": _hash_token(plan_token), + "refine_used": False, + "plan_used": False, + } + + logger.info("SDLC tokens generated", pipeline_id=pipeline_id) + + return _make_success( + "Tokens generated", + data={ + "pipeline_id": pipeline_id, + "refine_token": refine_token, + "plan_token": plan_token, + }, + ) + + +@sdlc_tokens_bp.route("/approve", methods=["POST"]) +def approve_phase() -> tuple[Response, int]: + """Validate a token and approve an SDLC phase. + + Request body: + { + "pipeline_id": "issue-596", + "phase": "refine", + "token": "APPLE-HORSE-RIVER" + } + + Returns: + 200: Phase approved + 400: Bad input + 403: Wrong token + 404: No tokens for pipeline + 409: Token already used + """ + data = request.get_json() or {} + pipeline_id = data.get("pipeline_id") + phase = data.get("phase") + token = data.get("token") + + if not pipeline_id: + return _make_error("Missing pipeline_id") + if not phase: + return _make_error("Missing phase") + if not token: + return _make_error("Missing token") + + if phase not in VALID_PHASES: + return _make_error(f"Invalid phase: {phase}. Must be one of: {', '.join(sorted(VALID_PHASES))}") + + if pipeline_id not in _token_store: + return _make_error(f"No tokens found for pipeline {pipeline_id}", 404) + + store = _token_store[pipeline_id] + hash_key = f"{phase}_hash" + used_key = f"{phase}_used" + + if store[used_key]: + return _make_error(f"Token for phase '{phase}' has already been used", 409) + + # Timing-safe comparison of token hashes + provided_hash = _hash_token(token) + if not secrets.compare_digest(provided_hash, store[hash_key]): + logger.warning( + "SDLC token validation failed", + pipeline_id=pipeline_id, + phase=phase, + ) + return _make_error("Invalid token", 403) + + # Mark token as used + store[used_key] = True + + # Resolve any pending decisions for this pipeline/phase + _resolve_phase_decisions(pipeline_id, phase) + + logger.info("SDLC phase approved", pipeline_id=pipeline_id, phase=phase) + + return _make_success( + f"Phase '{phase}' approved", + data={"pipeline_id": pipeline_id, "phase": phase}, + ) + + +@sdlc_tokens_bp.route("/reset", methods=["POST"]) +@_require_launcher_auth +def reset_token_gate() -> tuple[Response, int]: + """Clear token-gated state for a pipeline. + + Requires launcher secret authentication. Only callable by privileged + callers — not by Claude in the sandbox. + + Recovery endpoint for when the orchestrator restarts and in-memory tokens + are lost while Pipeline.sdlc_token_gated is still True in persistent storage. + + Request body: + {"pipeline_id": "issue-596"} + """ + data = request.get_json() or {} + pipeline_id = data.get("pipeline_id") + + if not pipeline_id: + return _make_error("Missing pipeline_id") + + # Clear persistent flag first — if this fails, don't clear in-memory + # tokens, since the pipeline would remain gated in persistent storage + # while appearing cleared in memory. + try: + from routes import get_repo_path + from state_store import PipelineNotFoundError, get_state_store + + repo_path = get_repo_path() + store = get_state_store(repo_path) + pipeline = store.load_pipeline(pipeline_id) + if pipeline.sdlc_token_gated: + pipeline.sdlc_token_gated = False + store.save_pipeline(pipeline) + logger.info("Token gate cleared", pipeline_id=pipeline_id) + except PipelineNotFoundError: + pass # Pipeline doesn't exist in store — just clear in-memory tokens + except Exception as e: + logger.error( + "Failed to clear persistent token gate", + pipeline_id=pipeline_id, + error=str(e), + exc_info=True, + ) + return _make_error("Failed to clear token gate in persistent storage", 503) + + # Only clear in-memory tokens after persistent flag is successfully cleared + _token_store.pop(pipeline_id, None) + + return _make_success("Token gate cleared", data={"pipeline_id": pipeline_id}) + + +def _is_phase_transition_decision(decision: Any, phase: str) -> bool: + """Check if a decision is a phase-transition gate for the given phase. + + Matches the specific question format produced by the HITL gate in + pipelines.py: "The {phase} phase has completed. ..." + """ + expected_prefix = f"the {phase} phase has completed" + return decision.question.lower().startswith(expected_prefix) + + +def _resolve_phase_decisions(pipeline_id: str, phase: str) -> None: + """Resolve pending HITL decisions for a pipeline phase.""" + from routes import get_repo_path + + try: + from decision_queue import get_decision_queue + + repo_path = get_repo_path() + queue = get_decision_queue(pipeline_id, repo_path) + pending = queue.get_pending_decisions() + + for decision in pending: + if _is_phase_transition_decision(decision, phase): + queue.resolve_decision(decision.id, f"Approved via SDLC token ({phase})") + logger.info( + "Auto-resolved decision via SDLC token", + pipeline_id=pipeline_id, + decision_id=decision.id, + phase=phase, + ) + except Exception as e: + # Don't fail the approval if decision resolution fails + logger.warning( + "Failed to auto-resolve decisions", + pipeline_id=pipeline_id, + phase=phase, + error=str(e), + ) diff --git a/orchestrator/sdlc_wordlist.py b/orchestrator/sdlc_wordlist.py new file mode 100644 index 0000000000..f7f3576a61 --- /dev/null +++ b/orchestrator/sdlc_wordlist.py @@ -0,0 +1,44 @@ +""" +Word list for SDLC approval token generation. + +210 unique concrete, unambiguous, easy-to-spell English nouns (3-7 chars each). +3-word tokens yield 210^3 ≈ 9.3M combinations, sufficient for ephemeral single-session use. +""" + +WORD_LIST = [ + # Animals + "BEAR", "BIRD", "CAT", "CRAB", "CROW", "DEER", "DOG", "DOVE", "DUCK", + "EAGLE", "EEL", "ELK", "FISH", "FOX", "FROG", "GOAT", "GOOSE", "HAWK", + "HARE", "HORSE", "LAMB", "LION", "MOLE", "MOOSE", "MOTH", "MOUSE", + "NEWT", "OWL", "PANDA", "PIG", "PONY", "RAM", "ROBIN", "SEAL", "SHARK", + "SHEEP", "SLUG", "SNAIL", "SNAKE", "STORK", "SWAN", "TIGER", "TOAD", + "TROUT", "VIPER", "WASP", "WHALE", "WOLF", "WORM", "ZEBRA", + # Colors + "AMBER", "BLACK", "BLUE", "BROWN", "CORAL", "CREAM", "GOLD", "GREEN", + "GREY", "IVORY", "LILAC", "NAVY", "OLIVE", "PEACH", "PINK", "PLUM", + "RED", "RUBY", "RUST", "TAN", "TEAL", "WHITE", + # Fruits & food + "APPLE", "BASIL", "BEAN", "BERRY", "BREAD", "CANDY", "CEDAR", "CHERRY", + "CHILI", "COCOA", "DATE", "FIG", "GRAPE", "GUAVA", "HONEY", "KIWI", + "LEMON", "LIME", "MANGO", "MAPLE", "MELON", "MINT", + "PEAR", "RICE", "SAGE", "WHEAT", + # Weather & sky + "BOLT", "BREEZE", "CLOUD", "DEW", "FLAME", "FLOOD", "FOG", "FROST", + "GALE", "HAIL", "ICE", "MIST", "MOON", "RAIN", "SKY", "SLEET", + "SNOW", "STAR", "STORM", "SUN", "TIDE", "WIND", + # Nature & landscape + "ASH", "BAY", "BLUFF", "BOG", "BROOK", "CAPE", "CAVE", "CLIFF", + "COAST", "COVE", "CREEK", "DALE", "DELTA", "DUNE", "FALLS", "FIELD", + "FJORD", "GLEN", "GROVE", "GULF", "HILL", "ISLE", "LAKE", "MARSH", + "MESA", "MOSS", "OAK", "PALM", "PEAK", "PINE", "POND", "REEF", + "RIDGE", "RIVER", "ROCK", "SAND", "SHORE", "SLOPE", "STONE", "TRAIL", + "VALE", "VINE", "WOOD", + # Tools & objects + "ANVIL", "AXE", "BELL", "BOW", "BRICK", "BRUSH", "CHAIN", + "CHEST", "CLOCK", "COIN", "CROWN", "DRUM", "FLUTE", "FORGE", "GATE", + "GEAR", "GLASS", "GLOBE", "GONG", "HARP", "HELM", "HORN", "KEY", + "KNOT", "LANCE", "LATCH", "LENS", "LOCK", "MASK", "NAIL", "OPAL", + "PEARL", "PIKE", "PIPE", "PRISM", "RING", "ROPE", "SCALE", "SHIELD", + "SPEAR", "SPIKE", "SWORD", "TORCH", "TOWER", "VAULT", "WAGON", + "WHEEL", +] diff --git a/orchestrator/tests/test_sdlc_tokens.py b/orchestrator/tests/test_sdlc_tokens.py new file mode 100644 index 0000000000..973dab85c8 --- /dev/null +++ b/orchestrator/tests/test_sdlc_tokens.py @@ -0,0 +1,713 @@ +""" +Tests for SDLC token-gated approval endpoints and word list. +""" + +import re +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from flask import Flask + +# Add parent directory to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from sdlc_wordlist import WORD_LIST + +TEST_LAUNCHER_SECRET = "test-launcher-secret-for-sdlc-tokens" + + +class TestWordList: + """Tests for the SDLC word list.""" + + def test_minimum_word_count(self): + """Word list should have at least 200 words for sufficient entropy.""" + assert len(WORD_LIST) >= 200 + + def test_no_duplicates(self): + """Word list should have no duplicates.""" + assert len(WORD_LIST) == len(set(WORD_LIST)) + + def test_word_length(self): + """All words should be 2-7 characters.""" + for word in WORD_LIST: + assert 2 <= len(word) <= 7, f"Word '{word}' is {len(word)} chars (expected 2-7)" + + def test_all_uppercase(self): + """All words should be uppercase.""" + for word in WORD_LIST: + assert word == word.upper(), f"Word '{word}' is not uppercase" + + def test_all_alpha(self): + """All words should contain only letters.""" + for word in WORD_LIST: + assert word.isalpha(), f"Word '{word}' contains non-alpha characters" + + +@pytest.fixture() +def _set_launcher_secret(monkeypatch): + """Set EGG_LAUNCHER_SECRET for tests that interact with the SDLC token endpoints.""" + monkeypatch.setenv("EGG_LAUNCHER_SECRET", TEST_LAUNCHER_SECRET) + + +@pytest.fixture() +def auth_headers(): + """Return Authorization headers with valid launcher secret.""" + return {"Authorization": f"Bearer {TEST_LAUNCHER_SECRET}"} + + +@pytest.fixture() +def app(): + """Create a minimal Flask app with just the sdlc_tokens blueprint.""" + from routes.sdlc_tokens import _token_store, sdlc_tokens_bp + + test_app = Flask(__name__) + test_app.register_blueprint(sdlc_tokens_bp) + _token_store.clear() + yield test_app + _token_store.clear() + + +@pytest.fixture() +def client(app): + """Create a test client.""" + return app.test_client() + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestTokenGeneration: + """Tests for token generation endpoint.""" + + def test_generate_returns_two_tokens(self, client, auth_headers): + """Generate should return refine and plan tokens.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-100"}, + headers=auth_headers, + ) + assert resp.status_code == 200 + data = resp.get_json() + assert data["success"] is True + assert "refine_token" in data["data"] + assert "plan_token" in data["data"] + + def test_token_format(self, client, auth_headers): + """Tokens should be WORD-WORD-WORD format, uppercase.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-101"}, + headers=auth_headers, + ) + data = resp.get_json()["data"] + for key in ("refine_token", "plan_token"): + token = data[key] + assert re.match(r"^[A-Z]+-[A-Z]+-[A-Z]+$", token), f"Bad format: {token}" + parts = token.split("-") + assert len(parts) == 3 + for part in parts: + assert part in WORD_LIST + + def test_tokens_are_different(self, client, auth_headers): + """Refine and plan tokens should be different.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-102"}, + headers=auth_headers, + ) + data = resp.get_json()["data"] + assert data["refine_token"] != data["plan_token"] + + def test_generate_missing_pipeline_id(self, client, auth_headers): + """Generate without pipeline_id should return 400.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={}, + headers=auth_headers, + ) + assert resp.status_code == 400 + + def test_generate_duplicate_pipeline(self, client, auth_headers): + """Generating tokens twice for same pipeline should return 409.""" + client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-103"}, + headers=auth_headers, + ) + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-103"}, + headers=auth_headers, + ) + assert resp.status_code == 409 + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestTokenApproval: + """Tests for token approval endpoint.""" + + @pytest.fixture(autouse=True) + def setup_tokens(self, client, auth_headers, _set_launcher_secret): + """Generate tokens for testing.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-200"}, + headers=auth_headers, + ) + data = resp.get_json()["data"] + self.refine_token = data["refine_token"] + self.plan_token = data["plan_token"] + + def test_approve_correct_refine_token(self, client): + """Approving with correct refine token should return 200.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "refine", + "token": self.refine_token, + }, + ) + assert resp.status_code == 200 + data = resp.get_json() + assert data["success"] is True + assert data["data"]["phase"] == "refine" + + def test_approve_correct_plan_token(self, client): + """Approving with correct plan token should return 200.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "plan", + "token": self.plan_token, + }, + ) + assert resp.status_code == 200 + + def test_approve_wrong_token(self, client): + """Approving with wrong token should return 403.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "refine", + "token": "WRONG-BAD-TOKEN", + }, + ) + assert resp.status_code == 403 + + def test_approve_used_token(self, client): + """Using same token twice should return 409.""" + client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "refine", + "token": self.refine_token, + }, + ) + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "refine", + "token": self.refine_token, + }, + ) + assert resp.status_code == 409 + + def test_approve_no_tokens_for_pipeline(self, client): + """Approving for unknown pipeline should return 404.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-999", + "phase": "refine", + "token": "SOME-TOKEN-HERE", + }, + ) + assert resp.status_code == 404 + + def test_approve_invalid_phase(self, client): + """Approving with invalid phase should return 400.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "implement", + "token": self.refine_token, + }, + ) + assert resp.status_code == 400 + + def test_approve_missing_fields(self, client): + """Approving with missing fields should return 400.""" + for missing in ("pipeline_id", "phase", "token"): + payload = { + "pipeline_id": "issue-200", + "phase": "refine", + "token": self.refine_token, + } + del payload[missing] + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json=payload, + ) + assert resp.status_code == 400, f"Expected 400 when missing {missing}" + + def test_approve_case_insensitive(self, client): + """Token validation should be case-insensitive.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "refine", + "token": self.refine_token.lower(), + }, + ) + assert resp.status_code == 200 + + def test_approve_cross_phase_token_rejected(self, client): + """Using refine token for plan phase should fail.""" + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-200", + "phase": "plan", + "token": self.refine_token, + }, + ) + assert resp.status_code == 403 + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestHasTokensForPipeline: + """Tests for has_tokens_for_pipeline helper.""" + + def test_no_tokens(self, app): + """Should return False when no tokens exist.""" + from routes.sdlc_tokens import has_tokens_for_pipeline + + assert has_tokens_for_pipeline("issue-300") is False + + def test_with_tokens(self, client, auth_headers): + """Should return True after tokens are generated.""" + from routes.sdlc_tokens import has_tokens_for_pipeline + + client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-300"}, + headers=auth_headers, + ) + assert has_tokens_for_pipeline("issue-300") is True + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestDecisionGating: + """Tests for resolve endpoint gating on token-gated pipelines.""" + + def test_pipeline_model_has_sdlc_token_gated_field(self): + """Pipeline model should have sdlc_token_gated field defaulting to False.""" + from models import Pipeline + + pipeline = Pipeline(id="issue-400", issue_number=400, repo="owner/repo", branch="egg/test") + assert pipeline.sdlc_token_gated is False + + def test_pipeline_model_accepts_sdlc_token_gated_true(self): + """Pipeline model should accept sdlc_token_gated=True.""" + from models import Pipeline + + pipeline = Pipeline( + id="issue-401", + issue_number=401, + repo="owner/repo", + branch="egg/test", + sdlc_token_gated=True, + ) + assert pipeline.sdlc_token_gated is True + + def test_pipeline_serialization_includes_sdlc_token_gated(self): + """Pipeline serialization should include sdlc_token_gated field.""" + from models import Pipeline + + pipeline = Pipeline( + id="issue-402", + issue_number=402, + repo="owner/repo", + branch="egg/test", + sdlc_token_gated=True, + ) + data = pipeline.model_dump() + assert "sdlc_token_gated" in data + assert data["sdlc_token_gated"] is True + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestDecisionGateFailClosed: + """Tests for the fail-closed behavior of the token gate in resolve_decision.""" + + def test_token_gated_pipeline_blocks_direct_resolution(self): + """resolve_decision should return 403 for token-gated pipelines.""" + from routes.decisions import decisions_bp + + test_app = Flask(__name__) + test_app.register_blueprint(decisions_bp) + + mock_pipeline = MagicMock() + mock_pipeline.sdlc_token_gated = True + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = mock_pipeline + + with test_app.test_client() as c: + with patch("routes.decisions.get_state_store", return_value=mock_store), \ + patch("routes.decisions.get_repo_path", return_value="/tmp/test"): + resp = c.post( + "/api/v1/pipelines/issue-500/decisions/decision-1/resolve", + json={"resolution": "approve"}, + ) + assert resp.status_code == 403 + assert "token-gated" in resp.get_json()["message"] + + def test_state_store_exception_returns_503(self): + """resolve_decision should return 503 (fail closed) on state store errors.""" + from routes.decisions import decisions_bp + + test_app = Flask(__name__) + test_app.register_blueprint(decisions_bp) + + mock_store = MagicMock() + mock_store.load_pipeline.side_effect = RuntimeError("connection lost") + + with test_app.test_client() as c: + with patch("routes.decisions.get_state_store", return_value=mock_store), \ + patch("routes.decisions.get_repo_path", return_value="/tmp/test"): + resp = c.post( + "/api/v1/pipelines/issue-500/decisions/decision-1/resolve", + json={"resolution": "approve"}, + ) + assert resp.status_code == 503 + assert "Unable to verify" in resp.get_json()["message"] + + def test_pipeline_not_found_allows_resolution(self): + """resolve_decision should allow resolution when pipeline doesn't exist.""" + from routes.decisions import decisions_bp + from state_store import PipelineNotFoundError + + test_app = Flask(__name__) + test_app.register_blueprint(decisions_bp) + + mock_store = MagicMock() + mock_store.load_pipeline.side_effect = PipelineNotFoundError("not found") + + mock_decision = MagicMock() + mock_decision.id = "decision-1" + mock_decision.question = "test?" + mock_decision.status.value = "resolved" + mock_decision.resolution = "approve" + mock_decision.resolved_at = None + + mock_queue = MagicMock() + mock_queue.resolve_decision.return_value = mock_decision + + with test_app.test_client() as c: + with patch("routes.decisions.get_state_store", return_value=mock_store), \ + patch("routes.decisions.get_repo_path", return_value="/tmp/test"), \ + patch("routes.decisions.get_decision_queue", return_value=mock_queue): + resp = c.post( + "/api/v1/pipelines/issue-500/decisions/decision-1/resolve", + json={"resolution": "approve"}, + ) + assert resp.status_code == 200 + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestPhaseTransitionDecisionMatching: + """Tests for _is_phase_transition_decision matching logic.""" + + def test_matches_exact_phase_transition_question(self): + """Should match the standard phase transition question format.""" + from routes.sdlc_tokens import _is_phase_transition_decision + + decision = MagicMock() + decision.question = "The refine phase has completed. Please review the analysis and approve to continue." + assert _is_phase_transition_decision(decision, "refine") is True + + def test_does_not_match_different_phase(self): + """Should not match when checking for a different phase.""" + from routes.sdlc_tokens import _is_phase_transition_decision + + decision = MagicMock() + decision.question = "The refine phase has completed. Please review the analysis and approve to continue." + assert _is_phase_transition_decision(decision, "plan") is False + + def test_does_not_match_unrelated_question_containing_phase_name(self): + """Should not match questions that happen to contain the phase name.""" + from routes.sdlc_tokens import _is_phase_transition_decision + + decision = MagicMock() + decision.question = "What's the test plan?" + assert _is_phase_transition_decision(decision, "plan") is False + + def test_does_not_match_ambiguous_question(self): + """Should not match 'Should we refine the plan?' for either phase.""" + from routes.sdlc_tokens import _is_phase_transition_decision + + decision = MagicMock() + decision.question = "Should we refine the plan?" + assert _is_phase_transition_decision(decision, "refine") is False + assert _is_phase_transition_decision(decision, "plan") is False + + def test_matches_plan_phase_transition(self): + """Should match the plan phase transition question.""" + from routes.sdlc_tokens import _is_phase_transition_decision + + decision = MagicMock() + decision.question = "The plan phase has completed. Please review the plan and approve to continue." + assert _is_phase_transition_decision(decision, "plan") is True + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestResolvePhaseDecisions: + """Tests for _resolve_phase_decisions auto-resolution.""" + + def test_resolves_matching_decision(self): + """Should resolve a decision that matches the phase transition pattern.""" + from routes.sdlc_tokens import _resolve_phase_decisions + + mock_decision = MagicMock() + mock_decision.id = "decision-1" + mock_decision.question = "The refine phase has completed. Please review the analysis and approve to continue." + + mock_queue = MagicMock() + mock_queue.get_pending_decisions.return_value = [mock_decision] + + with patch("routes.get_repo_path", return_value="/tmp/test"), \ + patch("decision_queue.get_decision_queue", return_value=mock_queue): + _resolve_phase_decisions("issue-600", "refine") + + mock_queue.resolve_decision.assert_called_once_with( + "decision-1", "Approved via SDLC token (refine)" + ) + + def test_does_not_resolve_unrelated_decision(self): + """Should not resolve decisions that don't match the phase pattern.""" + from routes.sdlc_tokens import _resolve_phase_decisions + + mock_decision = MagicMock() + mock_decision.id = "decision-1" + mock_decision.question = "What's the test plan for this feature?" + + mock_queue = MagicMock() + mock_queue.get_pending_decisions.return_value = [mock_decision] + + with patch("routes.get_repo_path", return_value="/tmp/test"), \ + patch("decision_queue.get_decision_queue", return_value=mock_queue): + _resolve_phase_decisions("issue-600", "plan") + + mock_queue.resolve_decision.assert_not_called() + + def test_handles_exception_gracefully(self): + """Should not raise when decision resolution fails.""" + from routes.sdlc_tokens import _resolve_phase_decisions + + with patch("routes.get_repo_path", side_effect=RuntimeError("no repo")): + # Should not raise + _resolve_phase_decisions("issue-600", "refine") + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestResetEndpoint: + """Tests for the /reset endpoint.""" + + def test_reset_clears_in_memory_tokens(self, client, auth_headers): + """Reset should remove tokens from the in-memory store.""" + from routes.sdlc_tokens import has_tokens_for_pipeline + from state_store import PipelineNotFoundError + + client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-700"}, + headers=auth_headers, + ) + assert has_tokens_for_pipeline("issue-700") is True + + # Mock state store — pipeline doesn't exist in persistent storage + mock_store = MagicMock() + mock_store.load_pipeline.side_effect = PipelineNotFoundError("not found") + + with patch("state_store.get_state_store", return_value=mock_store), \ + patch("routes.get_repo_path", return_value="/tmp/test"): + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-700"}, + headers=auth_headers, + ) + assert resp.status_code == 200 + assert has_tokens_for_pipeline("issue-700") is False + + def test_reset_missing_pipeline_id(self, client, auth_headers): + """Reset without pipeline_id should return 400.""" + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={}, + headers=auth_headers, + ) + assert resp.status_code == 400 + + def test_reset_nonexistent_pipeline(self, client, auth_headers): + """Reset for unknown pipeline should succeed (idempotent).""" + from state_store import PipelineNotFoundError + + mock_store = MagicMock() + mock_store.load_pipeline.side_effect = PipelineNotFoundError("not found") + + with patch("state_store.get_state_store", return_value=mock_store), \ + patch("routes.get_repo_path", return_value="/tmp/test"): + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-999"}, + headers=auth_headers, + ) + assert resp.status_code == 200 + + def test_reset_returns_503_on_persistent_store_failure(self, client, auth_headers): + """Reset should return 503 when persistent store write fails.""" + from routes.sdlc_tokens import has_tokens_for_pipeline + + # Pre-populate in-memory tokens + client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-750"}, + headers=auth_headers, + ) + assert has_tokens_for_pipeline("issue-750") is True + + mock_pipeline = MagicMock() + mock_pipeline.sdlc_token_gated = True + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = mock_pipeline + mock_store.save_pipeline.side_effect = RuntimeError("disk full") + + with patch("state_store.get_state_store", return_value=mock_store), \ + patch("routes.get_repo_path", return_value="/tmp/test"): + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-750"}, + headers=auth_headers, + ) + assert resp.status_code == 503 + assert "Failed to clear" in resp.get_json()["message"] + + # In-memory tokens should NOT be cleared since persistent store failed + assert has_tokens_for_pipeline("issue-750") is True + + def test_reset_clears_both_stores_on_success(self, client, auth_headers): + """Reset should clear persistent flag and in-memory tokens on success.""" + from routes.sdlc_tokens import has_tokens_for_pipeline + + client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-760"}, + headers=auth_headers, + ) + + mock_pipeline = MagicMock() + mock_pipeline.sdlc_token_gated = True + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = mock_pipeline + + with patch("state_store.get_state_store", return_value=mock_store), \ + patch("routes.get_repo_path", return_value="/tmp/test"): + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-760"}, + headers=auth_headers, + ) + assert resp.status_code == 200 + + assert has_tokens_for_pipeline("issue-760") is False + assert mock_pipeline.sdlc_token_gated is False + mock_store.save_pipeline.assert_called_once() + + +@pytest.mark.usefixtures("_set_launcher_secret") +class TestLauncherAuth: + """Tests for launcher secret authentication on privileged endpoints.""" + + def test_generate_without_auth_returns_401(self, client): + """Generate without Authorization header should return 401.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-800"}, + ) + assert resp.status_code == 401 + + def test_generate_with_wrong_secret_returns_401(self, client): + """Generate with wrong launcher secret should return 401.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-801"}, + headers={"Authorization": "Bearer wrong-secret"}, + ) + assert resp.status_code == 401 + + def test_generate_missing_bearer_prefix_returns_401(self, client): + """Generate without Bearer prefix should return 401.""" + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-802"}, + headers={"Authorization": TEST_LAUNCHER_SECRET}, + ) + assert resp.status_code == 401 + + def test_reset_without_auth_returns_401(self, client): + """Reset without Authorization header should return 401.""" + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-803"}, + ) + assert resp.status_code == 401 + + def test_reset_with_wrong_secret_returns_401(self, client): + """Reset with wrong launcher secret should return 401.""" + resp = client.post( + "/api/v1/sdlc-tokens/reset", + json={"pipeline_id": "issue-804"}, + headers={"Authorization": "Bearer wrong-secret"}, + ) + assert resp.status_code == 401 + + def test_approve_does_not_require_launcher_auth(self, client, auth_headers): + """Approve endpoint should not require launcher auth (uses token auth).""" + # Generate tokens first (with auth) + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-805"}, + headers=auth_headers, + ) + token = resp.get_json()["data"]["refine_token"] + + # Approve without launcher auth — should work (uses SDLC token instead) + resp = client.post( + "/api/v1/sdlc-tokens/approve", + json={ + "pipeline_id": "issue-805", + "phase": "refine", + "token": token, + }, + ) + assert resp.status_code == 200 + + def test_generate_with_no_secret_configured_returns_401(self, client, monkeypatch): + """Generate should return 401 when no launcher secret is configured.""" + monkeypatch.delenv("EGG_LAUNCHER_SECRET", raising=False) + resp = client.post( + "/api/v1/sdlc-tokens/generate", + json={"pipeline_id": "issue-806"}, + headers={"Authorization": "Bearer anything"}, + ) + assert resp.status_code == 401 diff --git a/sandbox/.claude/hooks/sdlc-approve.sh b/sandbox/.claude/hooks/sdlc-approve.sh new file mode 100755 index 0000000000..fc9318063a --- /dev/null +++ b/sandbox/.claude/hooks/sdlc-approve.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +set -euo pipefail + +# SDLC Phase Approval Hook +# +# Intercepts "!approve " prompts in Claude Code's UserPromptSubmit hook. +# Reads the approval token directly from /dev/tty so Claude never sees it, +# then validates against the orchestrator's token store. +# +# Installed as root-owned (0555) to prevent Claude from modifying it. + +# Read hook input from stdin (Claude Code passes JSON with prompt) +INPUT=$(cat) +PROMPT=$(echo "$INPUT" | jq -r '.user_prompt // .prompt // ""' 2>/dev/null || echo "$INPUT") + +# Only intercept !approve commands for token-gated phases (refine|plan). +# NOTE: If gated phases change, also update VALID_PHASES in sdlc_tokens.py. +if [[ ! "$PROMPT" =~ ^!approve[[:space:]]+(refine|plan)$ ]]; then + exit 0 # Pass through all other prompts +fi + +PHASE="${BASH_REMATCH[1]}" +PIPELINE_ID=$(cat /tmp/.egg-sdlc-pipeline-id 2>/dev/null || true) + +if [[ -z "$PIPELINE_ID" ]]; then + echo "No SDLC pipeline active." > /dev/tty + exit 0 +fi + +# Read token directly from terminal — Claude cannot see this +echo "" > /dev/tty +echo "=== SDLC Phase Approval ===" > /dev/tty +echo "Phase: $PHASE | Pipeline: $PIPELINE_ID" > /dev/tty +echo -n "Enter approval token: " > /dev/tty +read -r TOKEN < /dev/tty + +if [[ -z "$TOKEN" ]]; then + echo "No token entered." > /dev/tty + exit 0 +fi + +ORCH_URL="${EGG_ORCHESTRATOR_URL:-http://egg-orchestrator:9849}" +JSON_PAYLOAD=$(jq -n \ + --arg pid "$PIPELINE_ID" \ + --arg phase "$PHASE" \ + --arg token "$TOKEN" \ + '{pipeline_id: $pid, phase: $phase, token: $token}') +RESPONSE=$(curl -s -w "\n%{http_code}" -X POST \ + "$ORCH_URL/api/v1/sdlc-tokens/approve" \ + -H "Content-Type: application/json" \ + -d "$JSON_PAYLOAD") + +HTTP_CODE=$(echo "$RESPONSE" | tail -1) +BODY=$(echo "$RESPONSE" | sed '$d') + +if [[ "$HTTP_CODE" == "200" ]]; then + echo "Phase '$PHASE' approved!" > /dev/tty +else + ERROR=$(echo "$BODY" | jq -r '.message // "Unknown error"' 2>/dev/null || echo "$BODY") + echo "Approval failed: $ERROR" > /dev/tty +fi + +# Allow "!approve " prompt to reach Claude so it checks pipeline status +exit 0 diff --git a/sandbox/egg_lib/cli.py b/sandbox/egg_lib/cli.py index 387f293c47..0dd99aad0a 100644 --- a/sandbox/egg_lib/cli.py +++ b/sandbox/egg_lib/cli.py @@ -111,6 +111,14 @@ def main() -> int | None: help="Rebuild compose images before starting (use with --compose)", ) + # SDLC pipeline with token-gated approvals + parser.add_argument( + "--sdlc", + type=int, + metavar="ISSUE", + help="Start SDLC pipeline with token-gated approvals for the given issue number", + ) + # Private mode arguments (mutually exclusive) mode_group = parser.add_mutually_exclusive_group() mode_group.add_argument( @@ -211,7 +219,7 @@ def main() -> int | None: return 0 # Normal run - if not run_claude(repo_mode=repo_mode): + if not run_claude(repo_mode=repo_mode, sdlc_issue=args.sdlc): return 1 return 0 diff --git a/sandbox/egg_lib/runtime.py b/sandbox/egg_lib/runtime.py index e994cd95c4..c5198d4929 100644 --- a/sandbox/egg_lib/runtime.py +++ b/sandbox/egg_lib/runtime.py @@ -475,7 +475,7 @@ def _setup_session_repos( return session_token, repos, filtered_repos -def run_claude(repo_mode: str | None = None) -> bool: +def run_claude(repo_mode: str | None = None, sdlc_issue: int | None = None) -> bool: """Run Claude Code CLI in the sandboxed container (interactive mode). Args: @@ -483,6 +483,8 @@ def run_claude(repo_mode: str | None = None) -> bool: - None: Legacy mode (all repos accessible, global env vars) - "private": Only mount private/internal repos - "public": Only mount public repos + sdlc_issue: Optional issue number for SDLC pipeline with token-gated approvals. + When set, EGG_SDLC_ISSUE is passed to the container. Returns: True if container ran successfully, False otherwise @@ -636,6 +638,10 @@ def run_claude(repo_mode: str | None = None) -> bool: if api_key: caller_env["ANTHROPIC_API_KEY"] = api_key + # Pass SDLC issue number for token-gated approvals + if sdlc_issue is not None: + caller_env["EGG_SDLC_ISSUE"] = str(sdlc_issue) + cmd = build_sandbox_docker_cmd( container_name=container_id, image=ctx.sandbox_image, diff --git a/sandbox/entrypoint.py b/sandbox/entrypoint.py index e5bd5521be..5d4566716d 100644 --- a/sandbox/entrypoint.py +++ b/sandbox/entrypoint.py @@ -601,6 +601,176 @@ def setup_anthropic_api(config: Config, logger: Logger) -> None: logger.info(" Credentials injected by gateway (not in container)") +# ============================================================================= +# SDLC Token-Gated Approvals +# ============================================================================= + + +def setup_sdlc_tokens(config: Config, logger: Logger) -> None: + """Generate and display SDLC approval tokens, install hook and watchdog. + + Called when EGG_SDLC_ISSUE is set. This runs as root before dropping to egg user. + """ + sdlc_issue = os.environ.get("EGG_SDLC_ISSUE") + if not sdlc_issue: + return + + import requests + + pipeline_id = f"issue-{sdlc_issue}" + orch_url = os.environ.get( + "EGG_ORCHESTRATOR_URL", + f"http://egg-orchestrator:{GATEWAY_PORT + 1}", + ) + + # Generate tokens via orchestrator (requires launcher secret auth) + launcher_secret = os.environ.get("EGG_LAUNCHER_SECRET", "") + headers = {} + if launcher_secret: + headers["Authorization"] = f"Bearer {launcher_secret}" + try: + resp = requests.post( + f"{orch_url}/api/v1/sdlc-tokens/generate", + json={"pipeline_id": pipeline_id}, + headers=headers, + timeout=10, + ) + resp.raise_for_status() + data = resp.json()["data"] + except Exception as e: + logger.error(f"Failed to generate SDLC tokens: {e}") + logger.error("Continuing without token-gated approvals.") + return + + refine_token = data["refine_token"] + plan_token = data["plan_token"] + + # Display tokens to human (visible on terminal before Claude starts) + print() + print("\033[1;33m" + "=" * 51 + "\033[0m") + print(f"\033[1;33m SDLC Approval Tokens for {pipeline_id:<22}\033[0m") + print("\033[1;33m" + "=" * 51 + "\033[0m") + print() + print(f" Refine: \033[1;32m{refine_token}\033[0m") + print(f" Plan: \033[1;32m{plan_token}\033[0m") + print() + print(" Write these down. Claude will NOT see them.") + print("\033[1;33m" + "=" * 51 + "\033[0m") + print() + input(" Press Enter when ready...") + print() + + # Write pipeline ID for hook script + Path("/tmp/.egg-sdlc-pipeline-id").write_text(pipeline_id) + os.chmod("/tmp/.egg-sdlc-pipeline-id", 0o444) + + # Install hook script (root-owned, 0555 — Claude can't modify) + hooks_dir = config.claude_dir / "hooks" + hooks_dir.mkdir(parents=True, exist_ok=True) + hook_src = Path("/opt/egg-runtime/sandbox/.claude/hooks/sdlc-approve.sh") + hook_dst = hooks_dir / "sdlc-approve.sh" + if hook_src.exists(): + shutil.copy2(hook_src, hook_dst) + else: + # Fallback: copy from repo if runtime path doesn't exist + logger.warn("Hook source not found at /opt/egg-runtime, using inline fallback") + return + # Set root ownership and read+execute only (Claude runs as egg, can't modify) + os.chown(hook_dst, 0, 0) + os.chmod(hook_dst, 0o555) + + logger.success(f"SDLC hook installed: {hook_dst} (root-owned, 0555)") + + # Add hook to settings.json + settings_file = config.claude_dir / "settings.json" + if settings_file.exists(): + settings = json.loads(settings_file.read_text()) + else: + settings = {} + + settings.setdefault("hooks", {}) + existing_hooks = settings["hooks"].get("UserPromptSubmit", []) + if not any(h.get("command") == str(hook_dst) for h in existing_hooks if isinstance(h, dict)): + existing_hooks.append({"type": "command", "command": str(hook_dst)}) + settings["hooks"]["UserPromptSubmit"] = existing_hooks + settings_file.write_text(json.dumps(settings, indent=2)) + os.chown(settings_file, config.runtime_uid, config.runtime_gid) + + logger.success("SDLC hook registered in settings.json") + + # Also set in project-level .claude/settings.json for redundancy + project_claude_dir = config.repos_dir / ".claude" + project_claude_dir.mkdir(parents=True, exist_ok=True) + project_settings_file = project_claude_dir / "settings.json" + project_settings = {} + if project_settings_file.exists(): + try: + project_settings = json.loads(project_settings_file.read_text()) + except json.JSONDecodeError: + pass + project_settings.setdefault("hooks", {}) + existing_project_hooks = project_settings["hooks"].get("UserPromptSubmit", []) + if not any(h.get("command") == str(hook_dst) for h in existing_project_hooks if isinstance(h, dict)): + existing_project_hooks.append({"type": "command", "command": str(hook_dst)}) + project_settings["hooks"]["UserPromptSubmit"] = existing_project_hooks + project_settings_file.write_text(json.dumps(project_settings, indent=2)) + os.chown(project_settings_file, config.runtime_uid, config.runtime_gid) + chown_recursive(project_claude_dir, config.runtime_uid, config.runtime_gid) + + # Append auto-start instruction to CLAUDE.md + claude_md = config.user_home / "CLAUDE.md" + if claude_md.exists(): + with open(claude_md, "a") as f: + f.write(f"\n\n---\n\n## Auto-Start SDLC\n\n" + f"IMMEDIATELY run `/sdlc {sdlc_issue}` as your first action. " + f"Do not wait for user input.\n") + logger.success(f"Auto-start instruction added to CLAUDE.md for issue {sdlc_issue}") + + # Start settings watchdog (background thread, runs as root) + _start_settings_watchdog(config, logger, str(hook_dst)) + + +def _start_settings_watchdog(config: Config, logger: Logger, hook_path: str) -> None: + """Start a background thread that monitors settings.json for hook removal. + + Runs as root — Claude (egg user) cannot kill threads in the parent process. + Checks every 5 seconds and re-adds the hook if it's missing. + """ + settings_file = config.claude_dir / "settings.json" + + def watchdog() -> None: + while True: + time.sleep(5) + try: + if not settings_file.exists(): + continue + settings = json.loads(settings_file.read_text()) + hooks = settings.get("hooks", {}) + user_prompt_hooks = hooks.get("UserPromptSubmit", []) + + # Check if our hook is present + hook_present = any( + h.get("command") == hook_path + for h in user_prompt_hooks + if isinstance(h, dict) + ) + + if not hook_present: + logger.warn("SDLC hook removed from settings.json — re-adding") + settings.setdefault("hooks", {}) + existing = settings["hooks"].get("UserPromptSubmit", []) + existing.append({"type": "command", "command": hook_path}) + settings["hooks"]["UserPromptSubmit"] = existing + settings_file.write_text(json.dumps(settings, indent=2)) + os.chown(settings_file, config.runtime_uid, config.runtime_gid) + except Exception: + logger.debug("SDLC settings watchdog error", exc_info=True) + + thread = threading.Thread(target=watchdog, daemon=True, name="sdlc-settings-watchdog") + thread.start() + logger.success("SDLC settings watchdog started (background thread)") + + def setup_worktrees(config: Config, logger: Logger) -> bool: """Validate gateway-managed worktree configuration. @@ -1418,6 +1588,11 @@ def run_interactive(config: Config, logger: Logger) -> int: for proxy_var in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]: env.pop(proxy_var, None) + # Remove launcher secret from Claude's environment — it's a privileged + # credential used only by the entrypoint (root) for orchestrator auth. + # Leaving it accessible would let Claude bypass SDLC token gating. + env.pop("EGG_LAUNCHER_SECRET", None) + logger.info("Launching Claude Code interactive mode...") # Print timing summary right before launching LLM @@ -1448,6 +1623,8 @@ def run_exec(config: Config, logger: Logger, args: list[str]) -> int: Exit code from the subprocess """ env = os.environ.copy() + # Remove launcher secret — privileged credential not for Claude's use + env.pop("EGG_LAUNCHER_SECRET", None) # Print timing summary before exec _startup_timer.print_summary() @@ -1544,6 +1721,14 @@ def signal_handler(signum: int, frame: Any) -> None: with timed_phase("setup_anthropic_api", logger): setup_anthropic_api(config, logger) + # Set up SDLC token-gated approvals if requested + with timed_phase("setup_sdlc_tokens", logger): + setup_sdlc_tokens(config, logger) + + # Remove launcher secret from process environment before launching Claude. + # setup_sdlc_tokens (above) was the last operation that needed it. + os.environ.pop("EGG_LAUNCHER_SECRET", None) + # Run appropriate mode (timing summary is printed inside each mode) if len(sys.argv) == 1: exit_code = run_interactive(config, logger)