diff --git a/.github/workflows/codex_pr_review.yml b/.github/workflows/codex_pr_review.yml index 6d35e872..b038b38b 100644 --- a/.github/workflows/codex_pr_review.yml +++ b/.github/workflows/codex_pr_review.yml @@ -115,6 +115,7 @@ jobs: working-directory: source run: | set -euo pipefail + rm -rf "${GITHUB_WORKSPACE}/source/data/output/codex_pr_review" bridge_script="${GITHUB_WORKSPACE}/bridge/scripts/run_codex_pr_review.py" if [ -f "${bridge_script}" ]; then script_path="${bridge_script}" @@ -154,3 +155,22 @@ jobs: name: codex-pr-review-${{ github.event.pull_request.number || github.run_id }}-${{ github.run_id }} path: source/data/output/codex_pr_review/ if-no-files-found: warn + + - name: Emit review completion event + if: always() + continue-on-error: true + env: + CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} + CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }} + CODEX_REVIEW_STEP_OUTCOME: ${{ steps.review.outcome }} + CODEX_REVIEW_DECISION_PATH: ${{ github.workspace }}/source/data/output/codex_pr_review/decision.json + CODEX_REVIEW_PR_NUMBER: ${{ github.event.pull_request.number }} + CODEX_REVIEW_HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: | + set -euo pipefail + emitter="${GITHUB_WORKSPACE}/bridge/scripts/emit_pr_review_event.py" + if [ ! -f "${emitter}" ]; then + echo "::warning::Trusted review event emitter not found." + exit 1 + fi + timeout --signal=TERM --kill-after=5s 60s python -I "${emitter}" diff --git a/cloudflare/codex-audit-proxy/README.md b/cloudflare/codex-audit-proxy/README.md index ccf7e5b1..07cb10a8 100644 --- a/cloudflare/codex-audit-proxy/README.md +++ b/cloudflare/codex-audit-proxy/README.md @@ -6,7 +6,7 @@ This Worker should stay separate from the Pigbibi CodexGateway Worker so origin The Worker is intentionally thin: -- serves `GET /healthz` locally, proxies legacy `POST /v1/codex-audit`, and proxies async `POST /v1/codex-audit/jobs` plus `GET /v1/codex-audit/jobs/{job_id}`; +- serves `GET /healthz` locally, proxies legacy/async Codex routes, and permits authenticated `POST /v1/ai/automation/runs` for sanitized review completion events; - requires a bearer token before proxying and forwards the GitHub Actions OIDC `Authorization` header to the existing Codex audit service; - keeps the VPS origin URL in a Cloudflare Worker secret, not in git; - does not store provider keys, GitHub tokens, or Codex credentials. diff --git a/cloudflare/codex-audit-proxy/src/index.mjs b/cloudflare/codex-audit-proxy/src/index.mjs index 370e7343..7a0cc728 100644 --- a/cloudflare/codex-audit-proxy/src/index.mjs +++ b/cloudflare/codex-audit-proxy/src/index.mjs @@ -13,6 +13,7 @@ POST /v1/ai/execute POST /v1/ai/execute/jobs POST /v1/ai/review + POST /v1/ai/automation/runs POST /v1/ai/feedback/register POST /v1/ai/feedback/evaluate POST /v1/ai/feedback/shadow @@ -32,6 +33,7 @@ const ALLOWED_ROUTES = [ "/v1/ai/execute", "/v1/ai/execute/jobs", "/v1/ai/review", + "/v1/ai/automation/runs", "/v1/ai/health", "/v1/ai/quota", "/v1/ai/changes", @@ -116,7 +118,7 @@ function methodAllowed(method, pathname) { if (pathname === "/v1/ai/changes") return method === "GET"; // All other routes: POST only if (["/v1/ai/analyze", "/v1/ai/execute", "/v1/ai/execute/jobs", - "/v1/ai/review", "/v1/ai/feedback/register", "/v1/ai/feedback/evaluate", + "/v1/ai/review", "/v1/ai/automation/runs", "/v1/ai/feedback/register", "/v1/ai/feedback/evaluate", "/v1/codex-audit", "/v1/codex-audit/jobs"].includes(pathname)) { return method === "POST"; } diff --git a/cloudflare/codex-audit-proxy/tests/index.test.mjs b/cloudflare/codex-audit-proxy/tests/index.test.mjs index 8c8b7b5f..c12a6598 100644 --- a/cloudflare/codex-audit-proxy/tests/index.test.mjs +++ b/cloudflare/codex-audit-proxy/tests/index.test.mjs @@ -108,3 +108,39 @@ test("worker allows authenticated job status polling subroutes", async () => { globalThis.fetch = originalFetch; } }); + +test("worker proxies authenticated review event records only via POST", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = async (url, init) => { + assert.equal(url, "https://origin.example/v1/ai/automation/runs"); + assert.equal(init.method, "POST"); + assert.equal(init.headers.get("Authorization"), "Bearer test-token"); + return new Response(JSON.stringify({ status: "ok", notification: { status: "sent" } }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }; + + try { + const response = await worker.fetch( + new Request("https://proxy.example/v1/ai/automation/runs", { + method: "POST", + headers: { Authorization: "Bearer test-token", "Content-Type": "application/json" }, + body: "{}", + }), + { CODEX_AUDIT_ORIGIN_URL: "https://origin.example" }, + ); + assert.equal(response.status, 200); + assert.deepEqual(await response.json(), { status: "ok", notification: { status: "sent" } }); + + const rejected = await worker.fetch( + new Request("https://proxy.example/v1/ai/automation/runs", { + headers: { Authorization: "Bearer test-token" }, + }), + { CODEX_AUDIT_ORIGIN_URL: "https://origin.example" }, + ); + assert.equal(rejected.status, 405); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/scripts/deploy_codex_audit_service.sh b/scripts/deploy_codex_audit_service.sh index cc581f34..9245c7bd 100644 --- a/scripts/deploy_codex_audit_service.sh +++ b/scripts/deploy_codex_audit_service.sh @@ -21,6 +21,7 @@ ALLOWED_DIRECT_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES:- ALLOWED_SOURCE_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES:-QuantStrategyLab/AIAuditBridge,QuantStrategyLab/BinancePlatform,QuantStrategyLab/CharlesSchwabPlatform,QuantStrategyLab/CnEquitySnapshotPipelines,QuantStrategyLab/CnEquityStrategies,QuantStrategyLab/CryptoLivePoolPipelines,QuantStrategyLab/CryptoStrategies,QuantStrategyLab/FirstradePlatform,QuantStrategyLab/HkEquitySnapshotPipelines,QuantStrategyLab/HkEquityStrategies,QuantStrategyLab/IBKRGatewayManager,QuantStrategyLab/InteractiveBrokersPlatform,QuantStrategyLab/LongBridgePlatform,QuantStrategyLab/MarketSignalSources,QuantStrategyLab/PoliticalEventTrackingResearch,QuantStrategyLab/QmtPlatform,QuantStrategyLab/QuantAdvisorResearch,QuantStrategyLab/QuantPlatformKit,QuantStrategyLab/QuantRuntimeSettings,QuantStrategyLab/QuantStrategyPlugins,QuantStrategyLab/ResearchSignalContextPipelines,QuantStrategyLab/SchwabTokenAutoRefresher,QuantStrategyLab/UsEquitySnapshotPipelines,QuantStrategyLab/UsEquityStrategies}" JOB_DIR="${CODEX_AUDIT_SERVICE_JOB_DIR:-/var/lib/codex-audit-bridge/jobs}" ADMIN_ENV_FILE="${CODEX_AUDIT_SERVICE_ADMIN_ENV_FILE:-/etc/codex-audit-bridge/admin.env}" +TELEGRAM_ENV_FILE="${CODEX_AUDIT_SERVICE_TELEGRAM_ENV_FILE:-/etc/codex-audit-bridge/telegram.env}" EXECUTION_POLICY_FILE="${CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH:-/etc/codex-audit-bridge-policy/execution_policy.json}" AUDIT_MODEL="${CODEX_AUDIT_SERVICE_MODEL:-}" AUDIT_REASONING_EFFORT="${CODEX_AUDIT_SERVICE_REASONING_EFFORT:-}" @@ -385,6 +386,7 @@ Environment=CODEX_AUDIT_SERVICE_OPENAI_USAGE_WINDOW_DAYS=${OPENAI_USAGE_WINDOW_D Environment=CODEX_AUDIT_SERVICE_ANTHROPIC_USAGE_WINDOW_DAYS=${ANTHROPIC_USAGE_WINDOW_DAYS} Environment=CODEX_AUDIT_SERVICE_SANDBOX=read-only EnvironmentFile=-${ADMIN_ENV_FILE} +EnvironmentFile=-${TELEGRAM_ENV_FILE} ${audit_model_line} ${audit_reasoning_effort_line} ${audit_token_line} diff --git a/scripts/emit_pr_review_event.py b/scripts/emit_pr_review_event.py new file mode 100644 index 00000000..e210f928 --- /dev/null +++ b/scripts/emit_pr_review_event.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Emit one bounded, non-sensitive PR review completion event.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import re +import sys +import time +from typing import Any, Mapping +import urllib.error +import urllib.parse +import urllib.request + + +EVENT_SCHEMA = "qsl.pr_review_event.v1" +EVENT_TASK = "pr_review_completed" +DEFAULT_AUDIENCE = "quant-codex-audit" +MAX_INPUT_BYTES = 128 * 1024 +MAX_RESPONSE_BYTES = 128 * 1024 +REPOSITORY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}/[A-Za-z0-9][A-Za-z0-9._-]{0,99}") +HEAD_SHA_RE = re.compile(r"[0-9a-f]{40}") +RUN_ID_RE = re.compile(r"[1-9][0-9]{0,19}") +REVIEW_OUTCOMES = frozenset({"success", "failure", "cancelled", "skipped"}) + + +class ReviewEventEmissionError(ValueError): + """Raised when the workflow event cannot be safely emitted.""" + + +def _required_string(env: Mapping[str, str], name: str) -> str: + value = env.get(name) + if type(value) is not str or not value.strip(): + raise ReviewEventEmissionError(f"{name} is required") + return value.strip() + + +def _read_json_object(path: Path) -> dict[str, Any]: + try: + raw = path.read_bytes() + except OSError as exc: + raise ReviewEventEmissionError(f"cannot read {path.name}") from exc + if len(raw) > MAX_INPUT_BYTES: + raise ReviewEventEmissionError(f"{path.name} exceeds size limit") + try: + value = json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ReviewEventEmissionError(f"{path.name} is not valid JSON") from exc + if type(value) is not dict: + raise ReviewEventEmissionError(f"{path.name} must contain an object") + return value + + +def _decision_flags(path_value: str) -> tuple[bool | None, bool | None]: + if not path_value: + return None, None + path = Path(path_value) + if not path.is_file(): + return None, None + try: + decision = _read_json_object(path) + except ReviewEventEmissionError: + return None, None + blocked = decision.get("blocked") + conflict = decision.get("contract_conflict") + return ( + blocked if type(blocked) is bool else None, + conflict if type(conflict) is bool else None, + ) + + +def build_payload(env: Mapping[str, str]) -> dict[str, Any]: + repository = _required_string(env, "GITHUB_REPOSITORY") + if REPOSITORY_RE.fullmatch(repository) is None: + raise ReviewEventEmissionError("GITHUB_REPOSITORY is invalid") + workflow_run_id = _required_string(env, "GITHUB_RUN_ID") + if RUN_ID_RE.fullmatch(workflow_run_id) is None: + raise ReviewEventEmissionError("GITHUB_RUN_ID is invalid") + review_outcome = _required_string(env, "CODEX_REVIEW_STEP_OUTCOME").lower() + if review_outcome not in REVIEW_OUTCOMES: + raise ReviewEventEmissionError("CODEX_REVIEW_STEP_OUTCOME is invalid") + + pr_number_value = _required_string(env, "CODEX_REVIEW_PR_NUMBER") + if not pr_number_value.isascii() or not pr_number_value.isdigit(): + raise ReviewEventEmissionError("CODEX_REVIEW_PR_NUMBER is invalid") + pr_number = int(pr_number_value) + if pr_number <= 0 or pr_number > 2**53 - 1: + raise ReviewEventEmissionError("CODEX_REVIEW_PR_NUMBER is invalid") + head_sha = _required_string(env, "CODEX_REVIEW_HEAD_SHA") + if HEAD_SHA_RE.fullmatch(head_sha) is None: + raise ReviewEventEmissionError("CODEX_REVIEW_HEAD_SHA is invalid") + + blocked, contract_conflict = _decision_flags(str(env.get("CODEX_REVIEW_DECISION_PATH") or "")) + metadata = { + "schema": EVENT_SCHEMA, + "repository": repository, + "pr_number": pr_number, + "head_sha": head_sha, + "workflow_run_id": workflow_run_id, + "review_outcome": review_outcome, + "blocked": blocked, + "contract_conflict": contract_conflict, + } + return { + "run_id": f"review:{repository}:{pr_number}:{head_sha}:{workflow_run_id}", + "task": EVENT_TASK, + "task_state": "completed", + "mode": "review_only", + "source_repository": repository, + "metadata": metadata, + } + + +def normalize_service_url(value: str) -> str: + raw = str(value or "").strip().rstrip("/") + parsed = urllib.parse.urlparse(raw) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.netloc + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ReviewEventEmissionError("CODEX_AUDIT_SERVICE_URL is invalid") + if parsed.scheme != "https" and parsed.hostname not in {"127.0.0.1", "localhost", "::1"}: + raise ReviewEventEmissionError("CODEX_AUDIT_SERVICE_URL must use HTTPS") + return raw + + +def request_github_oidc_token(env: Mapping[str, str], audience: str) -> str: + request_url = _required_string(env, "ACTIONS_ID_TOKEN_REQUEST_URL") + request_token = _required_string(env, "ACTIONS_ID_TOKEN_REQUEST_TOKEN") + separator = "&" if "?" in request_url else "?" + url = f"{request_url}{separator}audience={urllib.parse.quote(audience)}" + request = urllib.request.Request( + url, + method="GET", + headers={ + "Authorization": f"Bearer {request_token}", + "Accept": "application/json", + "User-Agent": "qsl-pr-review-event-bridge", + }, + ) + with urllib.request.urlopen(request, timeout=20) as response: + raw = response.read(MAX_RESPONSE_BYTES + 1) + if len(raw) > MAX_RESPONSE_BYTES: + raise ReviewEventEmissionError("GitHub OIDC response exceeds size limit") + payload = json.loads(raw.decode("utf-8")) + token = payload.get("value") if type(payload) is dict else None + if type(token) is not str or not token: + raise ReviewEventEmissionError("GitHub OIDC response is invalid") + return token + + +def post_event(service_url: str, token: str, payload: dict[str, Any]) -> None: + body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + request = urllib.request.Request( + f"{service_url}/v1/ai/automation/runs", + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "qsl-pr-review-event-bridge", + }, + ) + with urllib.request.urlopen(request, timeout=20) as response: + raw = response.read(MAX_RESPONSE_BYTES + 1) + if len(raw) > MAX_RESPONSE_BYTES: + raise ReviewEventEmissionError("event service response exceeds size limit") + response_body = json.loads(raw.decode("utf-8")) + if type(response_body) is not dict or response_body.get("status") != "ok": + raise ReviewEventEmissionError("event service response is invalid") + notification = response_body.get("notification") + if type(notification) is not dict or notification.get("status") not in {"sent", "deduplicated"}: + raise ReviewEventEmissionError("review notification was not delivered") + + +def main() -> int: + service_url_value = str(os.environ.get("CODEX_AUDIT_SERVICE_URL") or "").strip() + if not service_url_value: + print("::notice::Review event bridge skipped: CODEX_AUDIT_SERVICE_URL is not configured") + return 0 + try: + service_url = normalize_service_url(service_url_value) + payload = build_payload(os.environ) + audience = str(os.environ.get("CODEX_AUDIT_SERVICE_AUDIENCE") or DEFAULT_AUDIENCE).strip() + token = request_github_oidc_token(os.environ, audience) + for attempt in range(1, 4): + try: + post_event(service_url, token, payload) + print(f"Review completion event recorded for {payload['source_repository']}") + return 0 + except (OSError, urllib.error.URLError, ReviewEventEmissionError, json.JSONDecodeError): + if attempt == 3: + raise + time.sleep(attempt) + except (OSError, urllib.error.URLError, ReviewEventEmissionError, json.JSONDecodeError) as exc: + print(f"::warning::Review event bridge delivery failed: {type(exc).__name__}", file=sys.stderr) + return 1 + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/service/ai_gateway_service.py b/service/ai_gateway_service.py index e8a5921e..0839bf92 100644 --- a/service/ai_gateway_service.py +++ b/service/ai_gateway_service.py @@ -93,6 +93,14 @@ from service.task_state import TERMINAL_STATES, job_task_state from service.health import get_health_monitor from service.org_health import read_org_health +from service.review_event_notification import ( + EVENT_TASK as PR_REVIEW_EVENT_TASK, + assert_review_event_provenance, + dispatch_review_event_notification, + expected_review_event_run_id, + parse_review_event_metadata, +) +from service.review_event_store import get_review_event_store try: from quant_platform_kit.strategy_lifecycle.performance_monitor import try_record_platform_execution except ModuleNotFoundError: # pragma: no cover - optional in CI @@ -138,6 +146,7 @@ def try_record_platform_execution(*_args, **_kwargs): _audit.propagate = False _JOB_WRITE_LOCK = threading.Lock() +_REVIEW_EVENT_LOCK = threading.Lock() # ── helpers ──────────────────────────────────────────────────────────── @@ -1723,10 +1732,42 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st for key, value in metadata.items() if str(key).strip().lower() not in {"requested_mode", "mode"} } - ledger = get_automation_run_ledger() run_id = str(payload.get("run_id") or payload.get("job_id") or "") if not run_id.strip(): raise ValueError("run_id must be a non-empty string") + task_name = str(payload.get("task") or payload.get("task_name") or "") + if task_name == PR_REVIEW_EVENT_TASK: + review_event = parse_review_event_metadata(repo, metadata) + assert_review_event_provenance(claims, review_event) + if run_id != expected_review_event_run_id(review_event): + raise ValueError("review event run_id does not match metadata") + if payload.get("task_state") != "completed": + raise ValueError("review event task_state must be completed") + if payload.get("mode") != MODE_REVIEW_ONLY: + raise ValueError("review event mode must be review_only") + with _REVIEW_EVENT_LOCK: + review_store = get_review_event_store() + if review_store.get_status(run_id) == "sent": + notification = {"status": "deduplicated"} + else: + review_store.set_status(run_id, "pending") + notification = dispatch_review_event_notification(review_event) + notification_status = str(notification.get("status") or "failed") + review_store.set_status(run_id, notification_status) + _json_response( + self, + HTTPStatus.OK, + { + "status": "ok", + "event": { + "event_id": run_id, + "notification_status": str(notification.get("status") or "failed"), + }, + "notification": notification, + }, + ) + return + ledger = get_automation_run_ledger() existing = ledger.get(run_id) if existing is not None: existing_owner = _automation_run_owner_repository(existing) @@ -1736,7 +1777,6 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st if existing_metadata.get("origin") == "service_job": raise PermissionError("automation run is service-owned") _assert_automation_run_access(existing, claims) - task_name = str(payload.get("task") or payload.get("task_name") or "") existing_state = str(existing.get("task_state") or "") if isinstance(existing, dict) else "" task_state = str(payload.get("task_state") or payload.get("state") or existing_state or "running") existing_metadata = existing.get("metadata") if isinstance(existing, dict) and isinstance(existing.get("metadata"), dict) else {} @@ -1760,7 +1800,7 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st run_metadata["requested_mode"] = requested_mode pending_run = {"run_id": run_id, "task_name": task_name, "task_state": task_state, "metadata": run_metadata} control = _automation_control_snapshot(repo, task_name=task_name, requested_mode=requested_mode, pending_run=pending_run) - record = get_automation_run_ledger().record( + record = ledger.record( run_id, task_state, task_name=task_name, @@ -1772,7 +1812,8 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st owner_repository=repo, ) control = _automation_control_snapshot(repo, task_name=task_name, requested_mode=requested_mode, pending_run=record) - _json_response(self, HTTPStatus.OK, {"status": "ok", "run": record, "control": control}) + response = {"status": "ok", "run": record, "control": control} + _json_response(self, HTTPStatus.OK, response) def _handle_automation_authority(self, claims: dict[str, Any], payload: dict[str, Any]) -> None: source_repo = str(payload.get("source_repository") or "") diff --git a/service/briefing_dispatch.py b/service/briefing_dispatch.py index 2eb215c3..a91bb680 100644 --- a/service/briefing_dispatch.py +++ b/service/briefing_dispatch.py @@ -8,7 +8,7 @@ import subprocess import urllib.parse import urllib.request -from typing import Any +from typing import Any, Mapping from service.briefing_consumer import BriefingAction, BriefingConsumptionResult @@ -18,25 +18,25 @@ ) -def _telegram_token() -> str: +def _telegram_token(env: Mapping[str, str] = os.environ) -> str: for key in ( "TELEGRAM_TOKEN", "TG_TOKEN", "STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN", ): - value = str(os.environ.get(key) or "").strip() + value = str(env.get(key) or "").strip() if value: return value return "" -def _telegram_chat_ids() -> tuple[str, ...]: +def _telegram_chat_ids(env: Mapping[str, str] = os.environ) -> tuple[str, ...]: for key in ( "GLOBAL_TELEGRAM_CHAT_ID", "QSL_GLOBAL_TELEGRAM_CHAT_ID", "STRATEGY_PLUGIN_ALERT_TELEGRAM_CHAT_IDS", ): - raw = os.environ.get(key) + raw = env.get(key) if not raw: continue ids = [ @@ -49,6 +49,12 @@ def _telegram_chat_ids() -> tuple[str, ...]: return () +def telegram_delivery_config(env: Mapping[str, str] | None = None) -> tuple[str, tuple[str, ...]]: + """Return the existing Telegram delivery configuration without exposing values.""" + source = os.environ if env is None else env + return _telegram_token(source), _telegram_chat_ids(source) + + def _format_telegram_body(result: BriefingConsumptionResult) -> str: lines = [f"🚨 量化哨兵 daily briefing ({result.day})", ""] for finding in result.findings: diff --git a/service/review_event_notification.py b/service/review_event_notification.py new file mode 100644 index 00000000..0376b516 --- /dev/null +++ b/service/review_event_notification.py @@ -0,0 +1,147 @@ +"""Strict PR review completion event validation and sanitized notification.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import re +from typing import Any + +from service.briefing_dispatch import send_telegram_alert, telegram_delivery_config + + +EVENT_SCHEMA = "qsl.pr_review_event.v1" +EVENT_TASK = "pr_review_completed" +MAX_SAFE_INTEGER = 2**53 - 1 +_REPOSITORY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}/[A-Za-z0-9][A-Za-z0-9._-]{0,99}") +_HEAD_SHA_RE = re.compile(r"[0-9a-f]{40}") +_RUN_ID_RE = re.compile(r"[1-9][0-9]{0,19}") +_TRUSTED_WORKFLOW_RE = re.compile( + r"QuantStrategyLab/AIAuditBridge/\.github/workflows/codex_pr_review\.yml@" + r"(?:refs/heads/main|[0-9a-f]{40})" +) +_REVIEW_OUTCOMES = frozenset({"success", "failure", "cancelled", "skipped"}) +_EXACT_KEYS = frozenset( + { + "schema", + "repository", + "pr_number", + "head_sha", + "workflow_run_id", + "review_outcome", + "blocked", + "contract_conflict", + } +) + + +class ReviewEventError(ValueError): + """Raised when review event metadata violates the closed contract.""" + + +@dataclass(frozen=True, slots=True) +class ReviewEvent: + repository: str + pr_number: int + head_sha: str + workflow_run_id: str + review_outcome: str + blocked: bool | None + contract_conflict: bool | None + + @property + def workflow_run_url(self) -> str: + return f"https://github.com/{self.repository}/actions/runs/{self.workflow_run_id}" + + +def parse_review_event_metadata(owner_repository: str, metadata: Any) -> ReviewEvent: + if type(metadata) is not dict or frozenset(metadata) != _EXACT_KEYS: + raise ReviewEventError("review event metadata shape is invalid") + repository = metadata["repository"] + if ( + type(repository) is not str + or _REPOSITORY_RE.fullmatch(repository) is None + or repository != owner_repository + ): + raise ReviewEventError("review event repository is invalid") + if metadata["schema"] != EVENT_SCHEMA or type(metadata["schema"]) is not str: + raise ReviewEventError("review event schema is invalid") + pr_number = metadata["pr_number"] + if type(pr_number) is not int or not 0 < pr_number <= MAX_SAFE_INTEGER: + raise ReviewEventError("review event PR number is invalid") + head_sha = metadata["head_sha"] + if type(head_sha) is not str or _HEAD_SHA_RE.fullmatch(head_sha) is None: + raise ReviewEventError("review event head SHA is invalid") + workflow_run_id = metadata["workflow_run_id"] + if type(workflow_run_id) is not str or _RUN_ID_RE.fullmatch(workflow_run_id) is None: + raise ReviewEventError("review event workflow run id is invalid") + review_outcome = metadata["review_outcome"] + if type(review_outcome) is not str or review_outcome not in _REVIEW_OUTCOMES: + raise ReviewEventError("review event outcome is invalid") + blocked = metadata["blocked"] + contract_conflict = metadata["contract_conflict"] + if blocked is not None and type(blocked) is not bool: + raise ReviewEventError("review event blocked flag is invalid") + if contract_conflict is not None and type(contract_conflict) is not bool: + raise ReviewEventError("review event contract conflict flag is invalid") + return ReviewEvent( + repository=repository, + pr_number=pr_number, + head_sha=head_sha, + workflow_run_id=workflow_run_id, + review_outcome=review_outcome, + blocked=blocked, + contract_conflict=contract_conflict, + ) + + +def expected_review_event_run_id(event: ReviewEvent) -> str: + return ( + f"review:{event.repository}:{event.pr_number}:" + f"{event.head_sha}:{event.workflow_run_id}" + ) + + +def assert_review_event_provenance(claims: Any, event: ReviewEvent) -> None: + """Bind a notification to its GitHub run and the trusted review workflow.""" + if type(claims) is not dict: + raise PermissionError("review event OIDC claims are invalid") + auth_method = claims.get("auth_method") + if auth_method == "none" and claims.get("repository") == "local": + return + if auth_method != "github_oidc": + raise PermissionError("review events require GitHub OIDC") + if claims.get("repository") != event.repository: + raise PermissionError("review event OIDC repository does not match") + if claims.get("run_id") != event.workflow_run_id: + raise PermissionError("review event OIDC run id does not match") + workflow_ref = claims.get("workflow_ref") + job_workflow_ref = claims.get("job_workflow_ref") + trusted_ref = workflow_ref if job_workflow_ref is None or job_workflow_ref == "" else job_workflow_ref + if type(trusted_ref) is not str or _TRUSTED_WORKFLOW_RE.fullmatch(trusted_ref) is None: + raise PermissionError("review event workflow identity is not trusted") + + +def format_review_event_notification(event: ReviewEvent) -> str: + repository_name = event.repository.split("/", 1)[1] + blocked = "unknown" if event.blocked is None else str(event.blocked).lower() + conflict = "unknown" if event.contract_conflict is None else str(event.contract_conflict).lower() + return "\n".join( + ( + f"[QSL Review] {repository_name} PR #{event.pr_number}", + f"head={event.head_sha[:12]} review={event.review_outcome} blocked={blocked} contract_conflict={conflict}", + event.workflow_run_url, + ) + ) + + +def dispatch_review_event_notification(event: ReviewEvent) -> dict[str, str]: + token, chat_ids = telegram_delivery_config(os.environ) + if not token or not chat_ids: + return {"status": "skipped", "reason": "telegram_not_configured"} + sent = send_telegram_alert( + text=format_review_event_notification(event), + token=token, + chat_ids=chat_ids, + ) + return {"status": "sent" if sent else "failed"} diff --git a/service/review_event_store.py b/service/review_event_store.py new file mode 100644 index 00000000..672c70d9 --- /dev/null +++ b/service/review_event_store.py @@ -0,0 +1,152 @@ +"""Independent persisted notification state for PR review events.""" + +from __future__ import annotations + +import json +import math +import os +from pathlib import Path +import re +import tempfile +import threading +import time +from typing import Any + + +STORE_SCHEMA = "qsl.review_event_store.v1" +STORE_PATH_ENV = "CODEX_AUDIT_SERVICE_REVIEW_EVENT_STORE_PATH" +DEFAULT_MAX_EVENTS = 5_000 +MAX_STORE_BYTES = 2 * 1024 * 1024 +_EVENT_ID_RE = re.compile(r"[A-Za-z0-9._:/-]{1,512}") +_STATUSES = frozenset({"pending", "sent", "failed", "skipped"}) + + +class ReviewEventStoreError(RuntimeError): + """Raised when persisted review notification state is unsafe or unreadable.""" + + +class ReviewEventStore: + def __init__(self, *, storage_path: Path, max_events: int = DEFAULT_MAX_EVENTS) -> None: + self._storage_path = storage_path + self._max_events = max(1, int(max_events)) + self._lock = threading.Lock() + self._events = self._load() + + def _load(self) -> dict[str, dict[str, Any]]: + if not self._storage_path.exists(): + return {} + try: + raw = self._storage_path.read_bytes() + except OSError as exc: + raise ReviewEventStoreError("review event store is unreadable") from exc + if len(raw) > MAX_STORE_BYTES: + raise ReviewEventStoreError("review event store exceeds size limit") + try: + payload = json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ReviewEventStoreError("review event store is invalid") from exc + if type(payload) is not dict or set(payload) != {"schema", "events"}: + raise ReviewEventStoreError("review event store shape is invalid") + if payload["schema"] != STORE_SCHEMA or type(payload["schema"]) is not str: + raise ReviewEventStoreError("review event store schema is invalid") + events = payload["events"] + if type(events) is not dict or len(events) > self._max_events: + raise ReviewEventStoreError("review event store entries are invalid") + validated: dict[str, dict[str, Any]] = {} + for event_id, entry in events.items(): + if type(event_id) is not str or _EVENT_ID_RE.fullmatch(event_id) is None: + raise ReviewEventStoreError("review event id is invalid") + if type(entry) is not dict or set(entry) != {"status", "updated_at"}: + raise ReviewEventStoreError("review event entry shape is invalid") + status = entry["status"] + updated_at = entry["updated_at"] + if type(status) is not str or status not in _STATUSES: + raise ReviewEventStoreError("review event status is invalid") + if ( + type(updated_at) not in {int, float} + or isinstance(updated_at, bool) + or not math.isfinite(float(updated_at)) + or updated_at < 0 + ): + raise ReviewEventStoreError("review event timestamp is invalid") + validated[event_id] = {"status": status, "updated_at": float(updated_at)} + return validated + + def _persist_locked(self) -> None: + payload = {"schema": STORE_SCHEMA, "events": self._events} + body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + if len(body) > MAX_STORE_BYTES: + raise ReviewEventStoreError("review event store exceeds size limit") + temporary_path: Path | None = None + try: + self._storage_path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + dir=self._storage_path.parent, + prefix=f".{self._storage_path.name}.", + delete=False, + ) as handle: + temporary_path = Path(handle.name) + os.fchmod(handle.fileno(), 0o600) + handle.write(body) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, self._storage_path) + except OSError as exc: + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError: + pass + raise ReviewEventStoreError("review event store cannot be persisted") from exc + + def get_status(self, event_id: str) -> str | None: + if type(event_id) is not str or _EVENT_ID_RE.fullmatch(event_id) is None: + raise ReviewEventStoreError("review event id is invalid") + with self._lock: + entry = self._events.get(event_id) + return str(entry["status"]) if entry is not None else None + + def set_status(self, event_id: str, status: str) -> None: + if type(event_id) is not str or _EVENT_ID_RE.fullmatch(event_id) is None: + raise ReviewEventStoreError("review event id is invalid") + if type(status) is not str or status not in _STATUSES: + raise ReviewEventStoreError("review event status is invalid") + with self._lock: + previous = dict(self._events) + self._events[event_id] = {"status": status, "updated_at": time.time()} + if len(self._events) > self._max_events: + oldest = min( + self._events, + key=lambda key: (float(self._events[key]["updated_at"]), key), + ) + self._events.pop(oldest) + try: + self._persist_locked() + except ReviewEventStoreError: + self._events = previous + raise + + +def default_review_event_store_path() -> Path: + configured = os.environ.get(STORE_PATH_ENV, "").strip() + if configured: + return Path(configured).expanduser() + base_dir = Path( + os.environ.get("CODEX_AUDIT_SERVICE_JOB_DIR") + or os.environ.get("CODEX_AUDIT_SERVICE_STATE_DIR") + or (Path.home() / ".local/state/codex-audit-service") + ).expanduser() + return base_dir / "review_event_notifications.json" + + +_review_event_store: ReviewEventStore | None = None +_review_event_store_path: Path | None = None + + +def get_review_event_store() -> ReviewEventStore: + global _review_event_store, _review_event_store_path + path = default_review_event_store_path() + if _review_event_store is None or _review_event_store_path != path: + _review_event_store = ReviewEventStore(storage_path=path) + _review_event_store_path = path + return _review_event_store diff --git a/tests/test_ai_gateway_service_get_routes.py b/tests/test_ai_gateway_service_get_routes.py index a043e97f..93a2d033 100644 --- a/tests/test_ai_gateway_service_get_routes.py +++ b/tests/test_ai_gateway_service_get_routes.py @@ -255,6 +255,158 @@ def test_automation_run_record_rejects_blank_run_id_as_bad_request(self) -> None server.shutdown() server.server_close() + @patch("service.ai_gateway_service.dispatch_review_event_notification") + def test_review_event_is_strictly_bound_and_notified_once(self, dispatch) -> None: + dispatch.return_value = {"status": "sent"} + with tempfile.TemporaryDirectory() as tmp: + env = { + "CODEX_AUDIT_SERVICE_AUTH": "none", + "CODEX_AUDIT_SERVICE_ALLOW_NO_AUTH_FOR_LOCAL_TESTS": "true", + "CODEX_AUDIT_SERVICE_JOB_DIR": tmp, + "CODEX_AUDIT_SERVICE_AUTOMATION_OPERATOR_REPOSITORIES": "local", + } + metadata = { + "schema": "qsl.pr_review_event.v1", + "repository": "local/repo-a", + "pr_number": 17, + "head_sha": "a" * 40, + "workflow_run_id": "123456789", + "review_outcome": "failure", + "blocked": True, + "contract_conflict": False, + } + payload = { + "run_id": f"review:local/repo-a:17:{'a' * 40}:123456789", + "task_state": "completed", + "task": "pr_review_completed", + "mode": "review_only", + "source_repository": "local/repo-a", + "metadata": metadata, + } + with patch.dict(os.environ, env, clear=False): + server = ThreadingHTTPServer(("127.0.0.1", 0), AiGatewayRequestHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + base_url = f"http://127.0.0.1:{server.server_port}" + responses = [] + for _ in range(2): + request = urllib.request.Request( + f"{base_url}/v1/ai/automation/runs", + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=5) as response: + responses.append(json.loads(response.read().decode("utf-8"))) + finally: + server.shutdown() + server.server_close() + + self.assertEqual(responses[0]["notification"]["status"], "sent") + self.assertEqual(responses[1]["notification"]["status"], "deduplicated") + self.assertNotIn("control", responses[0]) + self.assertNotIn("run", responses[0]) + dispatch.assert_called_once() + self.assertIsNone(get_automation_run_ledger().get(payload["run_id"])) + + @patch("service.ai_gateway_service.dispatch_review_event_notification") + def test_review_event_rejects_unbound_run_id_before_notification(self, dispatch) -> None: + with tempfile.TemporaryDirectory() as tmp: + env = { + "CODEX_AUDIT_SERVICE_AUTH": "none", + "CODEX_AUDIT_SERVICE_ALLOW_NO_AUTH_FOR_LOCAL_TESTS": "true", + "CODEX_AUDIT_SERVICE_JOB_DIR": tmp, + "CODEX_AUDIT_SERVICE_AUTOMATION_OPERATOR_REPOSITORIES": "local", + } + payload = { + "run_id": "forged-run-id", + "task_state": "completed", + "task": "pr_review_completed", + "mode": "review_only", + "source_repository": "local/repo-a", + "metadata": { + "schema": "qsl.pr_review_event.v1", + "repository": "local/repo-a", + "pr_number": 17, + "head_sha": "a" * 40, + "workflow_run_id": "123456789", + "review_outcome": "failure", + "blocked": True, + "contract_conflict": False, + }, + } + with patch.dict(os.environ, env, clear=False): + server = ThreadingHTTPServer(("127.0.0.1", 0), AiGatewayRequestHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + request = urllib.request.Request( + f"http://127.0.0.1:{server.server_port}/v1/ai/automation/runs", + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with self.assertRaises(urllib.error.HTTPError) as ctx: + urllib.request.urlopen(request, timeout=5) + self.assertEqual(ctx.exception.code, 400) + finally: + server.shutdown() + server.server_close() + + dispatch.assert_not_called() + + @patch("service.ai_gateway_service.dispatch_review_event_notification") + def test_review_event_retries_after_notification_failure(self, dispatch) -> None: + dispatch.side_effect = [{"status": "failed"}, {"status": "sent"}] + with tempfile.TemporaryDirectory() as tmp: + env = { + "CODEX_AUDIT_SERVICE_AUTH": "none", + "CODEX_AUDIT_SERVICE_ALLOW_NO_AUTH_FOR_LOCAL_TESTS": "true", + "CODEX_AUDIT_SERVICE_JOB_DIR": tmp, + "CODEX_AUDIT_SERVICE_AUTOMATION_OPERATOR_REPOSITORIES": "local", + } + metadata = { + "schema": "qsl.pr_review_event.v1", + "repository": "local/repo-a", + "pr_number": 17, + "head_sha": "a" * 40, + "workflow_run_id": "123456789", + "review_outcome": "failure", + "blocked": True, + "contract_conflict": False, + } + payload = { + "run_id": f"review:local/repo-a:17:{'a' * 40}:123456789", + "task_state": "completed", + "task": "pr_review_completed", + "mode": "review_only", + "source_repository": "local/repo-a", + "metadata": metadata, + } + with patch.dict(os.environ, env, clear=False): + server = ThreadingHTTPServer(("127.0.0.1", 0), AiGatewayRequestHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + base_url = f"http://127.0.0.1:{server.server_port}" + statuses = [] + for _ in range(2): + request = urllib.request.Request( + f"{base_url}/v1/ai/automation/runs", + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(request, timeout=5) as response: + statuses.append(json.loads(response.read().decode("utf-8"))["notification"]["status"]) + finally: + server.shutdown() + server.server_close() + + self.assertEqual(statuses, ["failed", "sent"]) + self.assertEqual(dispatch.call_count, 2) + def test_external_automation_run_update_cannot_overwrite_service_job_run(self) -> None: with tempfile.TemporaryDirectory() as tmp: env = { diff --git a/tests/test_review_event_bridge.py b/tests/test_review_event_bridge.py new file mode 100644 index 00000000..a153cefb --- /dev/null +++ b/tests/test_review_event_bridge.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import MagicMock, patch + +from scripts import emit_pr_review_event +from service.review_event_notification import ( + ReviewEventError, + assert_review_event_provenance, + dispatch_review_event_notification, + expected_review_event_run_id, + parse_review_event_metadata, +) +from service.review_event_store import ReviewEventStore, ReviewEventStoreError + + +class ReviewEventEmitterTests(unittest.TestCase): + def test_build_payload_binds_current_pr_head_and_sanitized_decision(self) -> None: + with TemporaryDirectory() as tmp: + decision_path = Path(tmp) / "decision.json" + decision_path.write_text( + json.dumps({"blocked": True, "contract_conflict": False, "findings": ["secret body"]}), + encoding="utf-8", + ) + env = { + "GITHUB_REPOSITORY": "QuantStrategyLab/ExampleRepo", + "GITHUB_RUN_ID": "123456789", + "CODEX_REVIEW_STEP_OUTCOME": "failure", + "CODEX_REVIEW_DECISION_PATH": str(decision_path), + "CODEX_REVIEW_PR_NUMBER": "17", + "CODEX_REVIEW_HEAD_SHA": "a" * 40, + } + + payload = emit_pr_review_event.build_payload(env) + + self.assertEqual( + payload["run_id"], + f"review:QuantStrategyLab/ExampleRepo:17:{'a' * 40}:123456789", + ) + self.assertEqual(payload["task"], "pr_review_completed") + self.assertEqual(payload["task_state"], "completed") + self.assertEqual(payload["mode"], "review_only") + self.assertEqual( + payload["metadata"], + { + "schema": "qsl.pr_review_event.v1", + "repository": "QuantStrategyLab/ExampleRepo", + "pr_number": 17, + "head_sha": "a" * 40, + "workflow_run_id": "123456789", + "review_outcome": "failure", + "blocked": True, + "contract_conflict": False, + }, + ) + self.assertNotIn("findings", payload["metadata"]) + + def test_build_payload_uses_unknown_decision_fields_when_output_is_missing(self) -> None: + with TemporaryDirectory() as tmp: + payload = emit_pr_review_event.build_payload( + { + "GITHUB_REPOSITORY": "QuantStrategyLab/ExampleRepo", + "GITHUB_RUN_ID": "9", + "CODEX_REVIEW_STEP_OUTCOME": "failure", + "CODEX_REVIEW_DECISION_PATH": str(Path(tmp) / "missing.json"), + "CODEX_REVIEW_PR_NUMBER": "2", + "CODEX_REVIEW_HEAD_SHA": "b" * 40, + } + ) + + self.assertIsNone(payload["metadata"]["blocked"]) + self.assertIsNone(payload["metadata"]["contract_conflict"]) + + def test_normalize_service_url_requires_https_outside_loopback(self) -> None: + self.assertEqual( + emit_pr_review_event.normalize_service_url("https://audit.example/path/"), + "https://audit.example/path", + ) + self.assertEqual( + emit_pr_review_event.normalize_service_url("http://127.0.0.1:8080"), + "http://127.0.0.1:8080", + ) + for value in ( + "http://audit.example", + "https://user:pass@audit.example", + "https://audit.example/path?token=x", + "relative/path", + ): + with self.subTest(value=value), self.assertRaises(emit_pr_review_event.ReviewEventEmissionError): + emit_pr_review_event.normalize_service_url(value) + + def test_main_skips_cleanly_when_service_is_not_configured(self) -> None: + with patch.dict(os.environ, {}, clear=True): + self.assertEqual(emit_pr_review_event.main(), 0) + + def test_post_event_requires_delivered_or_deduplicated_notification(self) -> None: + response = MagicMock() + response.__enter__.return_value.read.return_value = json.dumps( + {"status": "ok", "notification": {"status": "failed"}} + ).encode("utf-8") + with patch("scripts.emit_pr_review_event.urllib.request.urlopen", return_value=response): + with self.assertRaisesRegex( + emit_pr_review_event.ReviewEventEmissionError, + "notification was not delivered", + ): + emit_pr_review_event.post_event("https://audit.example", "oidc", {}) + + +class ReviewEventStoreTests(unittest.TestCase): + def test_store_persists_status_without_automation_ledger_semantics(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "review_events.json" + store = ReviewEventStore(storage_path=path, max_events=2) + store.set_status("review:repo:1:a:1", "pending") + store.set_status("review:repo:1:a:1", "sent") + + reloaded = ReviewEventStore(storage_path=path, max_events=2) + + self.assertEqual(reloaded.get_status("review:repo:1:a:1"), "sent") + + def test_store_is_bounded_and_rejects_corrupt_state(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "review_events.json" + store = ReviewEventStore(storage_path=path, max_events=2) + with patch("service.review_event_store.time.time", side_effect=[1.0, 2.0, 3.0]): + store.set_status("review:repo:1:a:1", "sent") + store.set_status("review:repo:2:b:2", "sent") + store.set_status("review:repo:3:c:3", "sent") + self.assertIsNone(store.get_status("review:repo:1:a:1")) + self.assertEqual(store.get_status("review:repo:3:c:3"), "sent") + + path.write_text("not-json", encoding="utf-8") + with self.assertRaises(ReviewEventStoreError): + ReviewEventStore(storage_path=path) + + +class ReviewEventNotificationTests(unittest.TestCase): + def setUp(self) -> None: + self.metadata = { + "schema": "qsl.pr_review_event.v1", + "repository": "QuantStrategyLab/ExampleRepo", + "pr_number": 17, + "head_sha": "a" * 40, + "workflow_run_id": "123456789", + "review_outcome": "failure", + "blocked": True, + "contract_conflict": False, + } + + def test_parser_derives_public_url_and_run_id(self) -> None: + event = parse_review_event_metadata("QuantStrategyLab/ExampleRepo", self.metadata) + + self.assertEqual(event.workflow_run_url, "https://github.com/QuantStrategyLab/ExampleRepo/actions/runs/123456789") + self.assertEqual( + expected_review_event_run_id(event), + f"review:QuantStrategyLab/ExampleRepo:17:{'a' * 40}:123456789", + ) + + def test_parser_rejects_unknown_keys_and_type_confusion(self) -> None: + cases = [ + {**self.metadata, "review_body": "do not retain"}, + {**self.metadata, "pr_number": True}, + {**self.metadata, "blocked": 1}, + {**self.metadata, "head_sha": "A" * 40}, + {**self.metadata, "repository": "OtherOrg/ExampleRepo"}, + {**self.metadata, "workflow_run_id": "01"}, + ] + for metadata in cases: + with self.subTest(metadata=metadata), self.assertRaises(ReviewEventError): + parse_review_event_metadata("QuantStrategyLab/ExampleRepo", metadata) + + def test_provenance_binds_run_and_trusted_review_workflow(self) -> None: + event = parse_review_event_metadata("QuantStrategyLab/ExampleRepo", self.metadata) + direct_claims = { + "auth_method": "github_oidc", + "repository": event.repository, + "run_id": event.workflow_run_id, + "workflow_ref": "QuantStrategyLab/AIAuditBridge/.github/workflows/codex_pr_review.yml@refs/heads/main", + } + reusable_claims = { + **direct_claims, + "workflow_ref": "QuantStrategyLab/ExampleRepo/.github/workflows/codex_pr_review.yml@refs/heads/main", + "job_workflow_ref": "QuantStrategyLab/AIAuditBridge/.github/workflows/codex_pr_review.yml@refs/heads/main", + } + + assert_review_event_provenance(direct_claims, event) + assert_review_event_provenance(reusable_claims, event) + + invalid_claims = [ + {**direct_claims, "run_id": "987654321"}, + {**direct_claims, "repository": "QuantStrategyLab/OtherRepo"}, + {**direct_claims, "workflow_ref": "QuantStrategyLab/AIAuditBridge/.github/workflows/codex_audit.yml@refs/heads/main"}, + {**direct_claims, "auth_method": "static_token"}, + ] + for claims in invalid_claims: + with self.subTest(claims=claims), self.assertRaises(PermissionError): + assert_review_event_provenance(claims, event) + + @patch("service.review_event_notification.send_telegram_alert", return_value=True) + def test_dispatch_uses_sanitized_summary_only(self, send) -> None: + event = parse_review_event_metadata("QuantStrategyLab/ExampleRepo", self.metadata) + with patch.dict( + os.environ, + {"TELEGRAM_TOKEN": "token", "GLOBAL_TELEGRAM_CHAT_ID": "123"}, + clear=True, + ): + result = dispatch_review_event_notification(event) + + self.assertEqual(result["status"], "sent") + text = send.call_args.kwargs["text"] + self.assertIn("ExampleRepo PR #17", text) + self.assertIn("review=failure", text) + self.assertIn(event.workflow_run_url, text) + self.assertNotIn("token", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_run_codex_pr_review.py b/tests/test_run_codex_pr_review.py index 09ff08e5..730eb7a1 100644 --- a/tests/test_run_codex_pr_review.py +++ b/tests/test_run_codex_pr_review.py @@ -1517,6 +1517,7 @@ def test_reusable_workflow_runs_bridge_script_against_source_checkout(self) -> N self.assertIn("tr '[:upper:]' '[:lower:]'", workflow) self.assertIn('if ! api_fallback_enabled="$(resolve_boolean', workflow) self.assertIn('timeout --signal=TERM --kill-after=60s 25m python -I "${script_path}"', workflow) + self.assertIn('rm -rf "${GITHUB_WORKSPACE}/source/data/output/codex_pr_review"', workflow) self.assertIn("inputs.caller_concurrency_key || github.event.pull_request.number || github.run_id", workflow) self.assertNotIn("Validate bridge checkout token", workflow) self.assertIn("required: false", workflow) @@ -1536,6 +1537,14 @@ def test_reusable_workflow_runs_bridge_script_against_source_checkout(self) -> N self.assertIn("Trusted Codex review script not found", workflow) self.assertNotIn("source/scripts/run_codex_pr_review.py", workflow) self.assertIn("source/data/output/codex_pr_review/", workflow) + self.assertIn("Emit review completion event", workflow) + self.assertIn("if: always()", workflow) + self.assertIn("continue-on-error: true", workflow) + self.assertIn("bridge/scripts/emit_pr_review_event.py", workflow) + self.assertIn("CODEX_REVIEW_STEP_OUTCOME: ${{ steps.review.outcome }}", workflow) + self.assertIn("CODEX_REVIEW_PR_NUMBER: ${{ github.event.pull_request.number }}", workflow) + self.assertIn("CODEX_REVIEW_HEAD_SHA: ${{ github.event.pull_request.head.sha }}", workflow) + self.assertIn("CODEX_AUDIT_SERVICE_AUDIENCE", workflow) def test_repo_root_can_be_overridden_for_reusable_workflow(self) -> None: source = Path("scripts/run_codex_pr_review.py").read_text(encoding="utf-8") diff --git a/tests/test_run_monthly_codex_audit.py b/tests/test_run_monthly_codex_audit.py index 64693141..8f6edb45 100644 --- a/tests/test_run_monthly_codex_audit.py +++ b/tests/test_run_monthly_codex_audit.py @@ -2471,6 +2471,8 @@ def test_vps_deploy_adds_nginx_audit_route_without_router_service(self) -> None: self.assertIn("ANTHROPIC_USAGE_WINDOW_DAYS=", deploy_script) self.assertIn("CODEX_AUDIT_SERVICE_ADMIN_ENV_FILE", deploy_script) self.assertIn("EnvironmentFile=-${ADMIN_ENV_FILE}", deploy_script) + self.assertIn('TELEGRAM_ENV_FILE="${CODEX_AUDIT_SERVICE_TELEGRAM_ENV_FILE:-/etc/codex-audit-bridge/telegram.env}"', deploy_script) + self.assertIn("EnvironmentFile=-${TELEGRAM_ENV_FILE}", deploy_script) self.assertIn("sudo install -m 0600 -o root -g root", deploy_script) self.assertIn('sudo rm -f "$ADMIN_ENV_FILE"', deploy_script) self.assertNotIn("Environment=OPENAI_ADMIN_KEY=", deploy_script)