From f0a897e0889ddacaec127841254ed90b2f7ad223 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 23 Aug 2026 13:01:46 +0200 Subject: [PATCH] Harden ops run identity and leases Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f --- .../adr-005-ops-runs-vs-dev-work-records.md | 9 +- docs/deploy-ops-run-queue-railiance.md | 13 +- docs/ops-run-queue.md | 16 +- docs/runbook.md | 8 +- docs/task-emission-consumer-contract.md | 5 + k8s/railiance/20-runtime.yaml | 4 +- k8s/railiance/README.md | 6 +- k8s/railiance/bootstrap-secrets.sh | 4 +- src/activity_core/ops_run_queue.py | 36 ++++- src/activity_core/ops_runs_api.py | 109 ++++++++++---- tests/test_ops_run_queue.py | 85 +++++++++++ tests/test_ops_runs_api.py | 142 ++++++++++++++++++ ...0036-queue-identity-and-lease-integrity.md | 135 +++++++++++++++++ 13 files changed, 522 insertions(+), 50 deletions(-) create mode 100644 tests/test_ops_runs_api.py create mode 100644 workplans/ACTIVITY-WP-0036-queue-identity-and-lease-integrity.md diff --git a/docs/adr/adr-005-ops-runs-vs-dev-work-records.md b/docs/adr/adr-005-ops-runs-vs-dev-work-records.md index 92b59cb..dc577c5 100644 --- a/docs/adr/adr-005-ops-runs-vs-dev-work-records.md +++ b/docs/adr/adr-005-ops-runs-vs-dev-work-records.md @@ -96,13 +96,20 @@ When a definition emits a TaskSpec for internal fleet work: 3. Do **not** open a Forgejo issue. 4. Do **not** create a workplan task file for that day’s fire. -Claim API (sketch; implement in ACTIVITY-WP-0026): +Claim API (implemented in ACTIVITY-WP-0026 and hardened in +ACTIVITY-WP-0036): - `POST /ops-runs/claim` — lease next open run matching labels / worker id - `POST /ops-runs/{id}/complete` — succeeded + completion metadata - `POST /ops-runs/{id}/fail` — failed + retry policy - `GET /ops-runs?state=open` — operator visibility +Worker mutation credentials are bound to one configured queue `worker_id`; +the request body cannot assert a different claim owner. Heartbeat, completion, +and failure lock the row and require its lease deadline to remain strictly in +the future. Operator/SSO credentials provide visibility and explicit +administration, but do not act as a normal worker identity. + activity-core remains **when / what / where** only: it does **not** run domain LLM sessions or hold tenant git credentials. diff --git a/docs/deploy-ops-run-queue-railiance.md b/docs/deploy-ops-run-queue-railiance.md index c96ecf9..0768258 100644 --- a/docs/deploy-ops-run-queue-railiance.md +++ b/docs/deploy-ops-run-queue-railiance.md @@ -90,8 +90,9 @@ explicit is better for ops): | `OPS_RUN_LEASE_SECONDS` | `900` | Optional | | `OPS_RUN_MAX_ATTEMPTS` | `3` | Optional | | `OPS_RUN_SLA_HOURS` | `1` | Status stuck threshold | +| `ACTIVITY_CORE_WORKER_ID` | `rein-aharness@railiance01` | Exact non-secret identity bound to the worker token | -Optional worker auth (recommended before external claim): +Worker auth (required before external claim): ```bash # Generate once; store in secret — do not commit @@ -101,6 +102,10 @@ kubectl -n activity-core patch secret actcore-runtime-secret --type merge \ # Record token in operator secret store (OpenBao / password manager), not chat. ``` +Apply the ConfigMap identity and Secret token in the same rollout. If the token +is present without `ACTIVITY_CORE_WORKER_ID`, worker mutations fail with 503; +if a request body names another identity, they fail with 403. + Apply: ```bash @@ -223,11 +228,11 @@ curl -sS "http://127.0.0.1:8010/ops/automations/status?since=today" \ ### 6. Claim path smoke (manual, no harness yet) ```bash -WORKER_TOKEN=… # from secret if set; else local-dev open auth +WORKER_TOKEN=… # from secret curl -sS -X POST "http://127.0.0.1:8010/ops-runs/claim" \ -H "Content-Type: application/json" \ -H "X-Worker-Token: ${WORKER_TOKEN}" \ - -d '{"worker_id":"smoke@railiance01","labels":["automated"],"limit":1,"lease_seconds":120}' \ + -d '{"worker_id":"rein-aharness@railiance01","labels":["automated"],"limit":1,"lease_seconds":120}' \ | python3 -m json.tool ``` @@ -239,7 +244,7 @@ RUN_ID=… # from claim response curl -sS -X POST "http://127.0.0.1:8010/ops-runs/${RUN_ID}/fail" \ -H "Content-Type: application/json" \ -H "X-Worker-Token: ${WORKER_TOKEN}" \ - -d '{"worker_id":"smoke@railiance01","error":"T07 smoke only","reopen":true}' + -d '{"worker_id":"rein-aharness@railiance01","error":"T07 smoke only","reopen":true}' ``` - [ ] Claim returns the open run diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md index a35bb04..55d7810 100644 --- a/docs/ops-run-queue.md +++ b/docs/ops-run-queue.md @@ -59,7 +59,9 @@ workplan task file. Not an issue-core or Forgejo ticket. - `labels_mode`: `any` (default) — run must contain at least one listed label; `all` — run must contain every listed label; omit `labels` to claim any open run. - Claim uses `FOR UPDATE SKIP LOCKED` for concurrency safety. -- Stale claims (`state=claimed` and `lease_until < now()`) are reopened before select. +- Stale claims (`state=claimed` and `lease_until <= now()`) are reopened before select. +- Heartbeat, complete, and fail lock the row and require an active lease + (`lease_until > now()`). An expired worker cannot revive or close its claim. ### Complete / fail body @@ -150,9 +152,14 @@ model output are persisted or returned. ## Auth - **Worker:** `ACTIVITY_CORE_WORKER_TOKEN` via `X-Worker-Token` or - `Authorization: Bearer` (same value accepted on claim/complete/fail/heartbeat). -- **Operator:** existing ops SSO / `ACTIVITY_CORE_OPERATOR_TOKEN` for list/status. -- **Local dev:** `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS=1` when no tokens set. + `Authorization: Bearer`, bound to the exact non-secret + `ACTIVITY_CORE_WORKER_ID`. The body `worker_id` is a compatibility field and + must match that authenticated identity on claim/complete/fail/heartbeat. +- **Operator:** existing ops SSO / `ACTIVITY_CORE_OPERATOR_TOKEN` for + list/status and explicit lease expiry; operator credentials are not accepted + as worker mutation identities. +- **Local dev:** `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS=1` is required when + no tokens are set. There is no implicit open mode. ## Env @@ -162,6 +169,7 @@ model output are persisted or returned. | `OPS_RUN_LEASE_SECONDS` | `900` | Default claim lease | | `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims | | `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim credential | +| `ACTIVITY_CORE_WORKER_ID` | unset | Exact queue identity bound to the worker credential; required when the token is set | ## Consumer (rein-aharness) diff --git a/docs/runbook.md b/docs/runbook.md index 9906101..03815ca 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -456,7 +456,7 @@ curl -sS -X POST "http://localhost:8010/ops-runs/claim" \ # Reopen stale leases curl -sS -X POST "http://localhost:8010/ops-runs/expire-leases" \ - -H "X-Worker-Token: $ACTIVITY_CORE_WORKER_TOKEN" + -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN" ``` | Env | Default | Meaning | @@ -466,6 +466,12 @@ curl -sS -X POST "http://localhost:8010/ops-runs/expire-leases" \ | `OPS_RUN_MAX_ATTEMPTS` | `3` | Fail permanently after N claims | | `OPS_RUN_SLA_HOURS` | `1` | Stuck threshold in status | | `ACTIVITY_CORE_WORKER_TOKEN` | unset | Harness claim auth | +| `ACTIVITY_CORE_WORKER_ID` | unset | Exact worker identity bound to claim auth | + +Worker mutations reject a body `worker_id` that differs from +`ACTIVITY_CORE_WORKER_ID`, and heartbeat/complete/fail reject a missing or +expired lease. Set the token and identity together before starting the claim +consumer. Glas-backed workers may return the Glas 1.0 `GatewayResult` on complete or fail. Activity Core persists only `result.execution_evidence` plus the existing diff --git a/docs/task-emission-consumer-contract.md b/docs/task-emission-consumer-contract.md index 9ae0522..62f114f 100644 --- a/docs/task-emission-consumer-contract.md +++ b/docs/task-emission-consumer-contract.md @@ -22,6 +22,11 @@ issue-core and not a Forgejo ticket. | Complete | `POST /ops-runs/{id}/complete` | `{ worker_id, result }` | | Fail | `POST /ops-runs/{id}/fail` | `{ worker_id, error, reopen? }` | +The worker token is bound by Activity Core to one configured `worker_id`. +Every worker mutation must name that exact identity and must still hold a lease +whose deadline is strictly in the future. Operator/SSO authentication does not +substitute for worker authentication on these calls. + Full field list, auth, and env: **`docs/ops-run-queue.md`**. Consumer implementation (rein-aharness): **REIN-A-0002**. diff --git a/k8s/railiance/20-runtime.yaml b/k8s/railiance/20-runtime.yaml index fc2d819..924c164 100644 --- a/k8s/railiance/20-runtime.yaml +++ b/k8s/railiance/20-runtime.yaml @@ -24,7 +24,9 @@ data: OPS_RUN_LEASE_SECONDS: "900" OPS_RUN_MAX_ATTEMPTS: "3" OPS_RUN_SLA_HOURS: "1" - # ACTIVITY_CORE_WORKER_TOKEN lives in actcore-runtime-secret (optional until REIN-A-0002) + # Non-secret identity bound to ACTIVITY_CORE_WORKER_TOKEN at the API boundary. + ACTIVITY_CORE_WORKER_ID: rein-aharness@railiance01 + # ACTIVITY_CORE_WORKER_TOKEN lives in actcore-runtime-secret. ACTIVITY_DEFINITION_DIRS: /etc/activity-core/external-definitions CUSTODIAN_REPO_ROOT: /var/custodian ACTIVITY_CORE_ROOT: /etc/activity-core diff --git a/k8s/railiance/README.md b/k8s/railiance/README.md index 5f3d23d..1f7294b 100644 --- a/k8s/railiance/README.md +++ b/k8s/railiance/README.md @@ -41,8 +41,10 @@ host-network bridge and workstation tunnel are not part of this deployment. `OPS_RUN_QUEUE_ENABLED=true`. After image + migrate job (alembic **0007**), workers insert claimable `ops_runs` on emit. Full railiance checklist: `docs/deploy-ops-run-queue-railiance.md`. Keep host timers until REIN-A-0002. -Optional `ACTIVITY_CORE_WORKER_TOKEN` in `actcore-runtime-secret` for harness -claim auth. +`ACTIVITY_CORE_WORKER_TOKEN` in `actcore-runtime-secret` authenticates the +harness claim client. The non-secret `ACTIVITY_CORE_WORKER_ID` in the runtime +ConfigMap binds that credential to `rein-aharness@railiance01`; deploy both +settings together. | ExternalSecret | OpenBao path | Secret key | | --- | --- | --- | diff --git a/k8s/railiance/bootstrap-secrets.sh b/k8s/railiance/bootstrap-secrets.sh index 905a8da..3e58750 100644 --- a/k8s/railiance/bootstrap-secrets.sh +++ b/k8s/railiance/bootstrap-secrets.sh @@ -43,6 +43,6 @@ fi # actcore-issue-core-runtime (k8s/railiance/15-externalsecret-issue-core.yaml). # Apply that manifest after ClusterSecretStore openbao-activity-core is Ready. # -# Optional (ACTIVITY-WP-0026 / REIN-A-0002): patch ACTIVITY_CORE_WORKER_TOKEN and -# ACTIVITY_CORE_OPERATOR_TOKEN into actcore-runtime-secret — see +# Patch ACTIVITY_CORE_WORKER_TOKEN and ACTIVITY_CORE_OPERATOR_TOKEN into +# actcore-runtime-secret before enabling their API paths — see # docs/deploy-ops-run-queue-railiance.md. Not auto-generated here. diff --git a/src/activity_core/ops_run_queue.py b/src/activity_core/ops_run_queue.py index f199e07..8b65c97 100644 --- a/src/activity_core/ops_run_queue.py +++ b/src/activity_core/ops_run_queue.py @@ -36,6 +36,20 @@ def default_lease_seconds() -> int: return 900 +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _has_active_lease(row: OpsRun, *, now: datetime) -> bool: + """Return whether ``row`` has a lease strictly after ``now``.""" + lease_until = row.lease_until + if not isinstance(lease_until, datetime): + return False + if lease_until.tzinfo is None: + lease_until = lease_until.replace(tzinfo=timezone.utc) + return lease_until > now + + def max_attempts() -> int: try: return max(1, int(os.environ.get("OPS_RUN_MAX_ATTEMPTS", "3"))) @@ -163,14 +177,14 @@ async def create_ops_run_from_spec( async def reopen_stale_claims(session: AsyncSession) -> int: """Return claimed rows with expired leases to open.""" - now = datetime.now(timezone.utc) + now = _utcnow() stmt = ( update(OpsRun) .where( and_( OpsRun.state == "claimed", OpsRun.lease_until.is_not(None), - OpsRun.lease_until < now, + OpsRun.lease_until <= now, ) ) .values( @@ -252,13 +266,15 @@ async def heartbeat_ops_run( worker_id: str, lease_seconds: int | None = None, ) -> OpsRun | None: - row = await session.get(OpsRun, run_id) + row = await session.get(OpsRun, run_id, with_for_update=True) if row is None: return None if row.state != "claimed" or row.claim_owner != worker_id: return None + now = _utcnow() + if not _has_active_lease(row, now=now): + return None lease_seconds = lease_seconds or default_lease_seconds() - now = datetime.now(timezone.utc) row.lease_until = now + timedelta(seconds=lease_seconds) row.updated_at = now return row @@ -271,12 +287,14 @@ async def complete_ops_run( worker_id: str, result: dict[str, Any] | None = None, ) -> OpsRun | None: - row = await session.get(OpsRun, run_id) + row = await session.get(OpsRun, run_id, with_for_update=True) if row is None: return None if row.state != "claimed" or row.claim_owner != worker_id: return None - now = datetime.now(timezone.utc) + now = _utcnow() + if not _has_active_lease(row, now=now): + return None row.state = "succeeded" row.lease_until = None row.result = normalise_ops_result(result) @@ -293,12 +311,14 @@ async def fail_ops_run( reopen: bool = False, result: dict[str, Any] | None = None, ) -> OpsRun | None: - row = await session.get(OpsRun, run_id) + row = await session.get(OpsRun, run_id, with_for_update=True) if row is None: return None if row.state != "claimed" or row.claim_owner != worker_id: return None - now = datetime.now(timezone.utc) + now = _utcnow() + if not _has_active_lease(row, now=now): + return None payload = normalise_ops_result(result) if error: payload["error"] = error[:2000] diff --git a/src/activity_core/ops_runs_api.py b/src/activity_core/ops_runs_api.py index 8a1e8c5..ba401fa 100644 --- a/src/activity_core/ops_runs_api.py +++ b/src/activity_core/ops_runs_api.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hmac import os import uuid from datetime import datetime @@ -16,6 +17,7 @@ from activity_core.ops_auth import ( extract_operator_token, extract_sso_principal, operator_token_configured, + require_operator, ) from activity_core.ops_run_queue import ( claim_ops_runs, @@ -34,6 +36,7 @@ router = APIRouter(prefix="/ops-runs", tags=["ops-runs"]) _get_db: Callable[[], async_sessionmaker[AsyncSession]] | None = None WORKER_TOKEN_ENV = "ACTIVITY_CORE_WORKER_TOKEN" +WORKER_ID_ENV = "ACTIVITY_CORE_WORKER_ID" def bind_ops_runs_deps( @@ -53,6 +56,11 @@ def _worker_token_configured() -> bool: return bool((os.environ.get(WORKER_TOKEN_ENV) or "").strip()) +def configured_worker_id() -> str | None: + value = (os.environ.get(WORKER_ID_ENV) or "").strip() + return value or None + + def _extract_worker_token( *, x_worker_token: str | None, @@ -77,8 +85,18 @@ def require_worker_or_operator( x_worker_token=x_worker_token, authorization=authorization ) expected_worker = (os.environ.get(WORKER_TOKEN_ENV) or "").strip() - if expected_worker and worker_tok and worker_tok == expected_worker: - return f"worker:{worker_tok[:8]}" + if ( + expected_worker + and worker_tok + and hmac.compare_digest(worker_tok, expected_worker) + ): + worker_id = configured_worker_id() + if not worker_id: + raise HTTPException( + status_code=503, + detail=f"worker identity not configured ({WORKER_ID_ENV})", + ) + return f"worker:{worker_id}" sso = extract_sso_principal(request) if sso: @@ -88,18 +106,61 @@ def require_worker_or_operator( x_operator_token=x_operator_token, authorization=authorization ) expected_op = (os.environ.get("ACTIVITY_CORE_OPERATOR_TOKEN") or "").strip() - if expected_op and op_tok and op_tok == expected_op: + if expected_op and op_tok and hmac.compare_digest(op_tok, expected_op): return "operator:token" if not _worker_token_configured() and not operator_token_configured(): if allow_unauth_mutations(): return "dev:unauth" - # Dev convenience when no tokens configured at all - return "dev:open" + + raise HTTPException(status_code=503, detail="ops run auth not configured") raise HTTPException(status_code=401, detail="worker or operator auth required") +def require_worker( + request: Request, + *, + x_worker_token: str | None = None, + authorization: str | None = None, +) -> str: + """Return the configured worker identity for a valid worker credential.""" + del request # reserved for future mTLS/proxy-bound worker principals + expected_worker = (os.environ.get(WORKER_TOKEN_ENV) or "").strip() + worker_tok = _extract_worker_token( + x_worker_token=x_worker_token, authorization=authorization + ) + + if expected_worker: + if not worker_tok or not hmac.compare_digest(worker_tok, expected_worker): + raise HTTPException(status_code=401, detail="worker auth required") + worker_id = configured_worker_id() + if not worker_id: + raise HTTPException( + status_code=503, + detail=f"worker identity not configured ({WORKER_ID_ENV})", + ) + return worker_id + + if allow_unauth_mutations() and not operator_token_configured(): + return "dev:unauth" + + raise HTTPException(status_code=503, detail="worker auth not configured") + + +def bind_worker_id(requested_worker_id: str, authenticated_worker_id: str) -> str: + """Bind a body compatibility field to its authenticated queue principal.""" + requested = requested_worker_id.strip() + if authenticated_worker_id == "dev:unauth": + return requested + if requested != authenticated_worker_id: + raise HTTPException( + status_code=403, + detail="worker_id does not match authenticated worker", + ) + return authenticated_worker_id + + class ClaimBody(BaseModel): worker_id: str = Field(..., min_length=1, max_length=256) labels: list[str] | None = None @@ -195,21 +256,20 @@ async def post_claim( body: ClaimBody, request: Request, x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"), - x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"), authorization: str | None = Header(default=None), ) -> dict[str, Any]: - require_worker_or_operator( + authenticated_worker_id = require_worker( request, x_worker_token=x_worker_token, - x_operator_token=x_operator_token, authorization=authorization, ) + worker_id = bind_worker_id(body.worker_id, authenticated_worker_id) Session = _db() async with Session() as session: async with session.begin(): claimed = await claim_ops_runs( session, - worker_id=body.worker_id, + worker_id=worker_id, labels=body.labels, labels_mode=body.labels_mode, limit=body.limit, @@ -227,28 +287,27 @@ async def post_heartbeat( body: HeartbeatBody, request: Request, x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"), - x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"), authorization: str | None = Header(default=None), ) -> dict[str, Any]: - require_worker_or_operator( + authenticated_worker_id = require_worker( request, x_worker_token=x_worker_token, - x_operator_token=x_operator_token, authorization=authorization, ) + worker_id = bind_worker_id(body.worker_id, authenticated_worker_id) Session = _db() async with Session() as session: async with session.begin(): row = await heartbeat_ops_run( session, run_id, - worker_id=body.worker_id, + worker_id=worker_id, lease_seconds=body.lease_seconds, ) if row is None: raise HTTPException( status_code=409, - detail="not claimed by this worker or not found", + detail="not actively leased by this worker or not found", ) return ops_run_to_dict(row) @@ -259,28 +318,27 @@ async def post_complete( body: CompleteBody, request: Request, x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"), - x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"), authorization: str | None = Header(default=None), ) -> dict[str, Any]: - require_worker_or_operator( + authenticated_worker_id = require_worker( request, x_worker_token=x_worker_token, - x_operator_token=x_operator_token, authorization=authorization, ) + worker_id = bind_worker_id(body.worker_id, authenticated_worker_id) Session = _db() async with Session() as session: async with session.begin(): row = await complete_ops_run( session, run_id, - worker_id=body.worker_id, + worker_id=worker_id, result=body.result, ) if row is None: raise HTTPException( status_code=409, - detail="not claimed by this worker or not found", + detail="not actively leased by this worker or not found", ) return ops_run_to_dict(row) @@ -291,22 +349,21 @@ async def post_fail( body: FailBody, request: Request, x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"), - x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"), authorization: str | None = Header(default=None), ) -> dict[str, Any]: - require_worker_or_operator( + authenticated_worker_id = require_worker( request, x_worker_token=x_worker_token, - x_operator_token=x_operator_token, authorization=authorization, ) + worker_id = bind_worker_id(body.worker_id, authenticated_worker_id) Session = _db() async with Session() as session: async with session.begin(): row = await fail_ops_run( session, run_id, - worker_id=body.worker_id, + worker_id=worker_id, error=body.error, reopen=body.reopen, result=body.result, @@ -314,7 +371,7 @@ async def post_fail( if row is None: raise HTTPException( status_code=409, - detail="not claimed by this worker or not found", + detail="not actively leased by this worker or not found", ) return ops_run_to_dict(row) @@ -322,13 +379,11 @@ async def post_fail( @router.post("/expire-leases") async def post_expire_leases( request: Request, - x_worker_token: str | None = Header(default=None, alias="X-Worker-Token"), x_operator_token: str | None = Header(default=None, alias="X-Operator-Token"), authorization: str | None = Header(default=None), ) -> dict[str, Any]: - require_worker_or_operator( + await require_operator( request, - x_worker_token=x_worker_token, x_operator_token=x_operator_token, authorization=authorization, ) diff --git a/tests/test_ops_run_queue.py b/tests/test_ops_run_queue.py index 3fe79d8..04d3140 100644 --- a/tests/test_ops_run_queue.py +++ b/tests/test_ops_run_queue.py @@ -232,6 +232,7 @@ async def test_complete_persists_only_normalized_glas_evidence() -> None: row = MagicMock() row.state = "claimed" row.claim_owner = "worker-1" + row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1) row.result = {} session = AsyncMock() session.get = AsyncMock(return_value=row) @@ -275,6 +276,7 @@ async def test_fail_reopen_under_max_attempts(monkeypatch: pytest.MonkeyPatch) - row.id = uuid.uuid4() row.state = "claimed" row.claim_owner = "w1" + row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1) row.attempt = 1 row.result = {} session = AsyncMock() @@ -297,6 +299,7 @@ async def test_fail_permanent_at_max_attempts(monkeypatch: pytest.MonkeyPatch) - row.id = uuid.uuid4() row.state = "claimed" row.claim_owner = "w1" + row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1) row.attempt = 3 row.result = {} session = AsyncMock() @@ -316,6 +319,7 @@ async def test_fail_persists_redacted_failure_evidence() -> None: row = MagicMock() row.state = "claimed" row.claim_owner = "worker-1" + row.lease_until = datetime.now(timezone.utc) + timedelta(minutes=1) row.attempt = 1 row.result = {} session = AsyncMock() @@ -347,6 +351,87 @@ async def test_fail_persists_redacted_failure_evidence() -> None: assert "tool_error" not in row.result +@pytest.mark.asyncio +@pytest.mark.parametrize("lease_offset", [None, timedelta(0), timedelta(seconds=-1)]) +async def test_heartbeat_rejects_missing_or_expired_lease( + lease_offset: timedelta | None, +) -> None: + from activity_core import ops_run_queue as oq + + now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc) + row = MagicMock() + row.state = "claimed" + row.claim_owner = "worker-1" + row.lease_until = now + lease_offset if lease_offset is not None else None + session = AsyncMock() + session.get = AsyncMock(return_value=row) + + with patch.object(oq, "_utcnow", return_value=now): + heartbeat = await oq.heartbeat_ops_run( + session, + uuid.uuid4(), + worker_id="worker-1", + lease_seconds=60, + ) + + assert heartbeat is None + assert row.lease_until == (now + lease_offset if lease_offset is not None else None) + session.get.assert_awaited_once() + assert session.get.await_args.kwargs == {"with_for_update": True} + + +@pytest.mark.asyncio +async def test_complete_rejects_expired_lease_without_mutation() -> None: + from activity_core import ops_run_queue as oq + + now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc) + row = MagicMock() + row.state = "claimed" + row.claim_owner = "worker-1" + row.lease_until = now - timedelta(microseconds=1) + row.result = {"before": True} + session = AsyncMock() + session.get = AsyncMock(return_value=row) + + with patch.object(oq, "_utcnow", return_value=now): + completed = await oq.complete_ops_run( + session, + uuid.uuid4(), + worker_id="worker-1", + result={"ok": True}, + ) + + assert completed is None + assert row.state == "claimed" + assert row.result == {"before": True} + + +@pytest.mark.asyncio +async def test_fail_rejects_wrong_owner_with_active_lease() -> None: + from activity_core import ops_run_queue as oq + + now = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc) + row = MagicMock() + row.state = "claimed" + row.claim_owner = "worker-1" + row.lease_until = now + timedelta(minutes=1) + row.result = {"before": True} + session = AsyncMock() + session.get = AsyncMock(return_value=row) + + with patch.object(oq, "_utcnow", return_value=now): + failed = await oq.fail_ops_run( + session, + uuid.uuid4(), + worker_id="worker-2", + error="must not persist", + ) + + assert failed is None + assert row.state == "claimed" + assert row.result == {"before": True} + + def test_label_filter_any_vs_all() -> None: """Document labels_mode semantics used by claim_ops_runs.""" row_labels = {"automated", "research-brief"} diff --git a/tests/test_ops_runs_api.py b/tests/test_ops_runs_api.py new file mode 100644 index 0000000..68de209 --- /dev/null +++ b/tests/test_ops_runs_api.py @@ -0,0 +1,142 @@ +"""Authentication boundary tests for the ops_run worker API.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient + +from activity_core.ops_runs_api import ( + bind_worker_id, + require_worker, + require_worker_or_operator, + router, +) + + +def _request(headers: dict[str, str] | None = None) -> MagicMock: + request = MagicMock() + request.headers = headers or {} + return request + + +def _client() -> TestClient: + app = FastAPI() + app.include_router(router) + return TestClient(app) + + +def test_worker_token_binds_configured_identity(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret") + monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01") + + authenticated = require_worker( + _request(), + x_worker_token="worker-secret", + ) + + assert authenticated == "rein-aharness@railiance01" + assert bind_worker_id("rein-aharness@railiance01", authenticated) == authenticated + + +def test_spoofed_worker_id_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret") + monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01") + authenticated = require_worker(_request(), x_worker_token="worker-secret") + + with pytest.raises(HTTPException) as exc: + bind_worker_id("another-worker", authenticated) + + assert exc.value.status_code == 403 + + +def test_worker_token_without_identity_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret") + monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False) + + with pytest.raises(HTTPException) as exc: + require_worker(_request(), x_worker_token="worker-secret") + + assert exc.value.status_code == 503 + assert "ACTIVITY_CORE_WORKER_ID" in exc.value.detail + + +def test_invalid_worker_token_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret") + monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01") + + with pytest.raises(HTTPException) as exc: + require_worker(_request(), x_worker_token="wrong") + + assert exc.value.status_code == 401 + + +def test_operator_or_sso_is_not_worker_identity(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False) + monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret") + + for request, kwargs in ( + (_request({"Remote-User": "alice"}), {}), + (_request(), {"authorization": "Bearer operator-secret"}), + ): + with pytest.raises(HTTPException) as exc: + require_worker(request, **kwargs) + assert exc.value.status_code == 503 + + +def test_claim_endpoint_rejects_operator_and_spoofed_identity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret") + monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01") + monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret") + client = _client() + body = {"worker_id": "rein-aharness@railiance01", "limit": 1} + + operator = client.post( + "/ops-runs/claim", + json=body, + headers={"X-Operator-Token": "operator-secret"}, + ) + spoofed = client.post( + "/ops-runs/claim", + json={"worker_id": "another-worker", "limit": 1}, + headers={"X-Worker-Token": "worker-secret"}, + ) + + assert operator.status_code == 401 + assert spoofed.status_code == 403 + + +def test_unauthenticated_dev_worker_requires_explicit_opt_in( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False) + + with pytest.raises(HTTPException) as exc: + require_worker(_request()) + assert exc.value.status_code == 503 + + monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "true") + assert require_worker(_request()) == "dev:unauth" + assert bind_worker_id("local-worker", "dev:unauth") == "local-worker" + + +def test_read_auth_no_longer_defaults_open(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False) + + with pytest.raises(HTTPException) as exc: + require_worker_or_operator(_request()) + + assert exc.value.status_code == 503 diff --git a/workplans/ACTIVITY-WP-0036-queue-identity-and-lease-integrity.md b/workplans/ACTIVITY-WP-0036-queue-identity-and-lease-integrity.md new file mode 100644 index 0000000..a31b2e4 --- /dev/null +++ b/workplans/ACTIVITY-WP-0036-queue-identity-and-lease-integrity.md @@ -0,0 +1,135 @@ +--- +id: ACTIVITY-WP-0036 +type: workplan +title: "Bind queue mutations to worker identity and active leases" +domain: infotech +repo: activity-core +status: active +owner: codex +topic_slug: activity-core +priority: high +created: "2026-08-23" +updated: "2026-08-23" +related: + - ACT-ADR-005 + - ACTIVITY-WP-0026 + - ACTIVITY-WP-0032 + - HARNESS-WP-0003 +--- + +# Bind Queue Mutations to Worker Identity and Active Leases + +## Origin + +Activity Core's review of rein-aharness ADR-002 found two enforcement gaps at +the `ops_run` boundary: + +1. `ACTIVITY_CORE_WORKER_TOKEN` authenticates a shared caller class, but the + authenticated principal is discarded and the caller supplies any + `worker_id` in the request body. +2. heartbeat, completion, and failure accept an owned `claimed` row even when + its lease has expired, until another claim or explicit expiry request happens + to reopen it. + +These gaps make queue ownership advisory at precisely the point where a +repository executor needs a durable acceptance decision. Rein-side +cancellation remains necessary, but cannot replace server-side enforcement. + +## Boundary and rollout rules + +- Activity Core authenticates queue callers and owns lease acceptance. +- The queue `worker_id` is separate from the governed actor (`agt`) used by + Glas and sand-boxer. +- Operator SSO and break-glass credentials are not worker identities and must + not claim, heartbeat, complete, or fail runs through the normal worker API. +- Existing result normalization and retry ceilings remain unchanged. +- Production must receive the configured worker identity in the same rollout + that activates strict binding; do not strand the live consumer between API + and configuration revisions. + +## Confirm the cross-repo responsibility contract + +```task +id: ACTIVITY-WP-0036-T01 +status: done +priority: high +``` + +Review rein-aharness ADR-002 against the implemented Activity Core boundary. +Report any distinction between current state and target guarantees, and require +the responsibility matrix to assign both worker authentication and lease +acceptance to an enforceable owner. + +Activity Core sent required edits in State Hub message +`428abb02-75dc-450d-a7f6-56fc715409ab`: the ADR must disclose the current +self-asserted identity, require credential-to-worker binding, require +server-side rejection after lease expiry, and keep operator identity out of the +normal worker mutation path. + +## Reject mutations without an active lease + +```task +id: ACTIVITY-WP-0036-T02 +status: done +priority: high +``` + +Lock the target row while deciding heartbeat, completion, or failure. Require +the row to be claimed by the caller and `lease_until` to be strictly later than +the server's current UTC time. A missing lease, or a deadline equal to or before +the decision time, is a conflict and must not update result or state. Stale +claim reopening must use the same deadline boundary. + +Done when unit/API tests cover missing, equal, expired, active, and wrong-owner +leases and prove no late completion or heartbeat can revive an expired claim. + +Implemented with row-level locking and one strict server-time predicate shared +by heartbeat, completion, and failure. Stale reopening now uses the same +inclusive expiry boundary (`lease_until <= now`). Tests prove missing, equal, +expired, active, and wrong-owner behavior without mutating late rows. + +## Bind authenticated worker credentials to the claim owner + +```task +id: ACTIVITY-WP-0036-T03 +status: done +priority: high +``` + +Add an explicit configured queue worker identity and bind a valid worker token +to that identity. Compare the body `worker_id` to the authenticated identity +before every worker mutation and persist only the authenticated identity as +`claim_owner`. Refuse production worker mutations when token or identity +configuration is incomplete. Preserve an explicitly enabled local-development +path without presenting it as authenticated production behavior. + +Done when tests prove spoofed worker ids, operator/SSO credentials on worker +mutations, missing production identity configuration, and invalid worker +tokens fail closed. + +Implemented `ACTIVITY_CORE_WORKER_ID` as the non-secret identity bound to the +worker token. Claim and close paths validate the compatibility body field and +persist only the authenticated identity. Operator/SSO credentials are removed +from normal worker mutations; unauthenticated development requires the +existing explicit opt-in, and the former implicit open mode is gone. + +## Document, deploy, and prove the boundary + +```task +id: ACTIVITY-WP-0036-T04 +status: progress +priority: high +``` + +Update the queue contract, runbook, and Railiance manifest with the non-secret +configured worker identity. Run focused and full tests, deploy API and consumer +configuration without a compatibility gap, and prove one authenticated claim +plus active heartbeat and a rejected mismatched/expired mutation. Return the +revision and bounded evidence to rein-aharness for its ADR acknowledgement +cycle. + +Source verification: 468 tests passed with one live integration test skipped; +Python compilation, whitespace checks, and parsing all 16 Railiance Kubernetes +documents passed. The manifest binds the existing production worker token to +`rein-aharness@railiance01`. Production rollout and live rejection evidence +remain before this task and workplan can finish.