From 4a70dc025c92e0df73baa61f586725211276ed3b Mon Sep 17 00:00:00 2001 From: vladmesh <16962535+vladmesh@users.noreply.github.com> Date: Fri, 17 Jul 2026 00:01:15 +0300 Subject: [PATCH] test: observe the real deploy run in the mega harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2026-07-16 mega waited out its 420s deploy-run timeout while the deploy it was waiting for, deploy-poll-ea0bed35, had already succeeded. list_runs narrows its result to Run.user_id == caller for every non-admin X-Telegram-ID it sees, and a valid internal key does not lift that narrowing. pr_poller creates deploy runs with no user_id, so the harness user — which is not admin — was answered [] for the whole wait, whatever filter it passed. The contract suite faked /api/runs/ without that rule, so it stayed green. Observe deploy runs as a plain internal service with no user header, and pin the rule in an API service test against the real router. Look the run up by story: pr_poller stamps story_id on the run it creates, and a project can carry deploy runs of other stories. The returned run is checked against this mega's story and must carry a merged head_sha, so neither a foreign run nor an engineering-triggered one without a SHA can be mistaken for it. A user-scoped client is now rejected on the spot instead of polling a blind endpoint for seven minutes, and the fake runs API mirrors the ownership narrowing so this cannot pass unnoticed again. --- .../api/tests/service/test_runs_story_id.py | 48 +++++ tests/live/pipeline_helpers.py | 65 ++++-- tests/live/test_full_pipeline.py | 15 +- tests/live/test_harness_contract.py | 200 ++++++++++++++++-- 4 files changed, 287 insertions(+), 41 deletions(-) diff --git a/services/api/tests/service/test_runs_story_id.py b/services/api/tests/service/test_runs_story_id.py index 0b3c44f7..4ee8edbe 100644 --- a/services/api/tests/service/test_runs_story_id.py +++ b/services/api/tests/service/test_runs_story_id.py @@ -75,6 +75,54 @@ async def test_filter_runs_by_story_id(async_client: AsyncClient, _tasks_project assert not any(r["id"] == run_b for r in runs) +@pytest.mark.asyncio +async def test_list_runs_hides_unowned_runs_from_a_non_admin_caller( + async_client: AsyncClient, _tasks_project +): + """A non-admin X-Telegram-ID narrows the result even with a valid internal key. + + pr_poller creates deploy runs with no user_id, so this rule answers `[]` for + them to any user-scoped caller. The live mega harness relies on it: it must + observe deploy runs as a plain internal service, never as its own non-admin + user, or it waits out a deploy that already succeeded (2026-07-16). + """ + telegram_id = 999000998 + await async_client.post( + "/api/users/upsert", + json={"telegram_id": telegram_id, "username": "non_admin", "first_name": "Non"}, + ) + story_id = await _create_story(async_client, "Ownership narrowing test") + + run_id = f"deploy-poll-{uuid.uuid4().hex[:8]}" + resp = await async_client.post( + "/api/runs/", + json={ + "id": run_id, + "type": "deploy", + "project_id": TASK_TEST_PROJECT_ID, + "story_id": story_id, + "run_metadata": {"triggered_by": "pr_poll", "head_sha": "abc123"}, + }, + ) + assert resp.status_code == HTTPStatus.CREATED + assert resp.json()["user_id"] is None + + as_user = await async_client.get( + "/api/runs/", + params={"story_id": story_id, "run_type": "deploy"}, + headers={"X-Telegram-ID": str(telegram_id)}, + ) + assert as_user.status_code == HTTPStatus.OK + assert as_user.json() == [] + + as_service = await async_client.get( + "/api/runs/", + params={"story_id": story_id, "run_type": "deploy"}, + ) + assert as_service.status_code == HTTPStatus.OK + assert [run["id"] for run in as_service.json()] == [run_id] + + @pytest.mark.asyncio async def test_create_run_without_story_id(async_client: AsyncClient, _tasks_project): """Run without story_id has null story_id (standalone deploy).""" diff --git a/tests/live/pipeline_helpers.py b/tests/live/pipeline_helpers.py index e3e4f4b3..34bb08df 100644 --- a/tests/live/pipeline_helpers.py +++ b/tests/live/pipeline_helpers.py @@ -31,7 +31,8 @@ # ── Constants ──────────────────────────────────────────────────────────── API_URL = "http://localhost:8000" TEST_TELEGRAM_ID = 999_000_001 -AUTH_HEADERS = {"X-Telegram-ID": str(TEST_TELEGRAM_ID)} +USER_AUTH_HEADER = "X-Telegram-ID" +AUTH_HEADERS = {USER_AUTH_HEADER: str(TEST_TELEGRAM_ID)} GITHUB_ORG = "project-factory-organization" TEMPLATE_REPO = "gh:vladmesh/service-template" @@ -82,10 +83,13 @@ def internal_headers() -> dict[str, str]: """Auth headers for internal-service endpoints, as the real consumers send them. - /api/servers/* is gated by require_internal_or_admin, and /api/runs/ hides - other owners' runs from a non-admin caller. The harness user carries only - X-Telegram-ID and is not admin, so without this header those endpoints answer - 401/403 or silently return nothing. + /api/servers/* is gated by require_internal_or_admin, and the harness user is + not admin, so without this header those endpoints answer 401/403. + + This key alone does not make /api/runs/ show every run: list_runs still + narrows its result to the caller's own runs whenever it sees a non-admin + X-Telegram-ID. Unowned runs are only visible to a client that sends no user + header at all — see ``require_unscoped_run_observer``. """ return {"X-Internal-Key": os.environ["INTERNAL_API_KEY"]} @@ -573,51 +577,82 @@ def record_env_contract( # ── Deploy run outcome ─────────────────────────────────────────────────── +def require_unscoped_run_observer(api_internal: httpx.AsyncClient) -> None: + """Reject a run-observing client that authenticates as a user. + + list_runs narrows its result to ``Run.user_id == caller`` for every non-admin + ``X-Telegram-ID`` it sees, and a valid internal key does not lift that + narrowing. pr_poller creates deploy runs with no user_id, so a user-scoped + client is answered `[]` for them no matter which filter it passes. + + On 2026-07-16 that silently cost the mega a 420s wait for the already + successful deploy `deploy-poll-ea0bed35`, so this is a loud crash rather than + a blind poll. + """ + if USER_AUTH_HEADER in api_internal.headers: + raise RuntimeError( + f"deploy runs must be observed without {USER_AUTH_HEADER}: list_runs " + "narrows its result to runs the non-admin harness user owns, and " + "pr_poller's deploy run has no user_id" + ) + + async def wait_deploy_run( - api: httpx.AsyncClient, + api_internal: httpx.AsyncClient, ctx: dict, *, timeout: int = DEPLOY_RUN_TIMEOUT, poll_interval: float = DEPLOY_RUN_POLL_INTERVAL, ) -> dict | None: - """Wait for this project's deploy run that carries the merged head SHA. + """Wait for this story's deploy run that carries the merged head SHA. pr_poller creates it only once the story PR reports merged_at, and records the merged head SHA in run_metadata — the exact ref deploy resolves the environment contract at. Engineering-triggered deploy runs carry no head_sha, so a run without one is not the run this mega deploys. + + The story is the link the API really supports for this: pr_poller stamps + story_id on the run it creates, and a project can carry deploy runs of other + stories. Both the filter and the returned run are checked against this mega's + story, so a foreign run cannot be mistaken for it. + + Reads /api/runs/ as an internal service with no user header — see + ``require_unscoped_run_observer``. """ + require_unscoped_run_observer(api_internal) + story_id = ctx["story_id"] deadline = time.monotonic() + timeout while time.monotonic() < deadline: - resp = await api.get( + resp = await api_internal.get( "/api/runs/", - params={"project_id": ctx["project_id"], "run_type": RunType.DEPLOY.value}, + params={"story_id": story_id, "run_type": RunType.DEPLOY.value}, headers=internal_headers(), ) resp.raise_for_status() # The API orders runs newest first. for run in resp.json(): - head_sha = (run.get("run_metadata") or {}).get("head_sha") + if run["story_id"] != story_id: + continue + head_sha = (run["run_metadata"] or {}).get("head_sha") if head_sha: ctx["deploy_run_id"] = run["id"] ctx["deploy_head_sha"] = head_sha return run await asyncio.sleep(poll_interval) ctx["deploy_run_error"] = ( - f"no deploy run with a merged head_sha appeared for project " - f"{ctx['project_id']} within {timeout}s" + f"no deploy run with a merged head_sha appeared for story {story_id} within {timeout}s" ) return None async def wait_deploy_outcome( - api: httpx.AsyncClient, + api_internal: httpx.AsyncClient, ctx: dict, *, timeout: int = DEPLOY_OUTCOME_TIMEOUT, poll_interval: float = DEPLOY_OUTCOME_POLL_INTERVAL, ) -> DeployRunResult | None: - """Type this project's own deploy run result and record its outcome. + """Type this story's own deploy run result and record its outcome. A running application only proves some container answers; the deploy run result is what the pipeline itself concluded about the deploy, so the mega @@ -626,7 +661,7 @@ async def wait_deploy_outcome( run_id = ctx["deploy_run_id"] deadline = time.monotonic() + timeout while time.monotonic() < deadline: - resp = await api.get(f"/api/runs/{run_id}", headers=internal_headers()) + resp = await api_internal.get(f"/api/runs/{run_id}", headers=internal_headers()) resp.raise_for_status() run = resp.json() if run["status"] in _TERMINAL_RUN_STATUSES: diff --git a/tests/live/test_full_pipeline.py b/tests/live/test_full_pipeline.py index 4f9d28c7..e598d84e 100644 --- a/tests/live/test_full_pipeline.py +++ b/tests/live/test_full_pipeline.py @@ -32,6 +32,7 @@ create_story_and_task, dump_debug, ensure_test_user, + internal_headers, record_env_contract, trigger_scaffold, wait_deploy, @@ -56,7 +57,15 @@ async def pipeline(): """Full pipeline: scaffold → engineering → deploy. Yields context for assertions.""" async with httpx.AsyncClient(base_url=API_URL, timeout=10, headers=AUTH_HEADERS) as api: await ensure_test_user(api) - async with httpx.AsyncClient(base_url=API_URL, timeout=10) as api_no_auth: + # Deploy runs belong to no user, and list_runs hides unowned runs from the + # non-admin harness user, so they are observed through a client that + # authenticates only as an internal service. + async with ( + httpx.AsyncClient(base_url=API_URL, timeout=10) as api_no_auth, + httpx.AsyncClient( + base_url=API_URL, timeout=10, headers=internal_headers() + ) as api_internal, + ): ctx = await create_noop_project(api) async with cleanup_guard( lambda: cleanup_all(api, api_no_auth, ctx), manifest=ctx["manifest"] @@ -88,7 +97,7 @@ async def pipeline(): # does a deploy run appear carrying the merged head SHA — the ref # deploy reads the contract at. Re-check the contract there: the # scaffolded tree proves nothing about what engineering merged. - deploy_run = await wait_deploy_run(api, ctx, timeout=DEPLOY_RUN_TIMEOUT) + deploy_run = await wait_deploy_run(api_internal, ctx, timeout=DEPLOY_RUN_TIMEOUT) if deploy_run is None: yield ctx dump_debug(ctx, "full-deploy-run") @@ -104,7 +113,7 @@ async def pipeline(): return await wait_deploy(api, api_no_auth, ctx, timeout=DEPLOY_TIMEOUT) - await wait_deploy_outcome(api, ctx, timeout=DEPLOY_OUTCOME_TIMEOUT) + await wait_deploy_outcome(api_internal, ctx, timeout=DEPLOY_OUTCOME_TIMEOUT) if ( ctx.get("final_app_status") == ApplicationStatus.RUNNING.value and ctx.get("deploy_outcome") == DeployOutcome.SUCCESS.value diff --git a/tests/live/test_harness_contract.py b/tests/live/test_harness_contract.py index 44bde5be..73f9969d 100644 --- a/tests/live/test_harness_contract.py +++ b/tests/live/test_harness_contract.py @@ -1147,47 +1147,201 @@ def test_record_env_contract_rejects_sha_absent_from_main(monkeypatch): # ── Typed deploy outcome ───────────────────────────────────────────────── +HARNESS_USER_ID = 7 + + +def _deploy_run( + run_id: str, + *, + story_id: str = "story-1", + head_sha: str | None = "abc123", + user_id: int | None = None, +) -> dict: + """One /api/runs/ record shaped as the API returns it. + + ``head_sha=None`` is the engineering-triggered deploy run: pr_poller records + the merged SHA, engineering does not. ``user_id=None`` is how every deploy + run is really stored — neither producer attributes it to a user. + """ + metadata = {"triggered_by": "pr_poll", "head_sha": head_sha} if head_sha else {} + return { + "id": run_id, + "type": "deploy", + "project_id": "project-1", + "story_id": story_id, + "user_id": user_id, + "status": "completed", + "run_metadata": metadata, + } + + +def _runs_api(runs: list[dict]): + """A GET /api/runs/ that answers as services/api/src/routers/runs.py does. + + The ownership narrowing is the part that matters: list_runs restricts the + result to the caller's own runs for any non-admin X-Telegram-ID, and the + internal key does not lift it. Faking the endpoint without that rule is what + let the 2026-07-16 blindness through a green contract suite. + """ + + def handler(request: httpx.Request) -> httpx.Response: + params = request.url.params + selected = [ + run + for run in runs + if ("story_id" not in params or run["story_id"] == params["story_id"]) + and ("project_id" not in params or run["project_id"] == params["project_id"]) + and ("run_type" not in params or run["type"] == params["run_type"]) + ] + if pipeline_helpers.USER_AUTH_HEADER in request.headers: + selected = [run for run in selected if run["user_id"] == HARNESS_USER_ID] + return httpx.Response(200, json=selected) + + return handler + + +def _runs_client(runs: list[dict], *, as_user: bool = False) -> httpx.AsyncClient: + """Client for the fake runs API, authenticated the way the harness would.""" + headers = dict(pipeline_helpers.internal_headers()) + if as_user: + headers.update(pipeline_helpers.AUTH_HEADERS) + return httpx.AsyncClient( + base_url="http://test", + transport=httpx.MockTransport(_runs_api(runs)), + headers=headers, + ) + @pytest.mark.asyncio async def test_wait_deploy_run_selects_the_run_carrying_the_merged_sha(monkeypatch): """Only the pr_poller run records head_sha; the engineering-triggered deploy run does not, so it is not the run this mega's contract check must follow.""" monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") - seen = {} - - async def get(url, params=None, headers=None): - seen["params"] = params - seen["headers"] = headers - return httpx.Response( - 200, - json=[ - {"id": "deploy-eng-1", "run_metadata": {}}, - {"id": "deploy-poll-1", "run_metadata": {"head_sha": "abc123"}}, - ], - request=httpx.Request("GET", "http://test/api/runs/"), - ) + runs = [ + _deploy_run("deploy-eng-1", head_sha=None), + _deploy_run("deploy-poll-1", head_sha="abc123"), + ] + ctx = {"project_id": "project-1", "story_id": "story-1"} - ctx = {"project_id": "project-1"} - run = await pipeline_helpers.wait_deploy_run(SimpleNamespace(get=get), ctx, timeout=1) + async with _runs_client(runs) as api_internal: + run = await pipeline_helpers.wait_deploy_run(api_internal, ctx, timeout=1) assert run["id"] == "deploy-poll-1" assert ctx["deploy_run_id"] == "deploy-poll-1" assert ctx["deploy_head_sha"] == "abc123" - assert seen["params"] == {"project_id": "project-1", "run_type": "deploy"} + + +@pytest.mark.asyncio +async def test_wait_deploy_run_finds_the_unowned_run_the_user_filter_hid(monkeypatch): + """Regression, 2026-07-16: deploy `deploy-poll-ea0bed35` succeeded while the + mega waited 420s for a run it could not see. + + list_runs narrows to `Run.user_id == caller` for any non-admin X-Telegram-ID, + internal key or not, and pr_poller's deploy run has no user_id — so the + harness user was answered `[]` for the whole wait. Observed as a plain + internal service, the same run is found, and the request must reach the API + with the internal key and without the user header. + """ + monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") + seen = {} + runs = [_deploy_run("deploy-poll-ea0bed35", head_sha="ea0bed35c0ffee", user_id=None)] + + def handler(request: httpx.Request) -> httpx.Response: + seen["params"] = dict(request.url.params) + seen["headers"] = request.headers + return _runs_api(runs)(request) + + async with httpx.AsyncClient( + base_url="http://test", + transport=httpx.MockTransport(handler), + headers=pipeline_helpers.internal_headers(), + ) as api_internal: + ctx = {"project_id": "project-1", "story_id": "story-1"} + run = await pipeline_helpers.wait_deploy_run(api_internal, ctx, timeout=1) + + assert run["id"] == "deploy-poll-ea0bed35" + assert ctx["deploy_head_sha"] == "ea0bed35c0ffee" + assert seen["params"] == {"story_id": "story-1", "run_type": "deploy"} assert seen["headers"]["X-Internal-Key"] == "test-internal-key" + assert pipeline_helpers.USER_AUTH_HEADER not in seen["headers"] @pytest.mark.asyncio -async def test_wait_deploy_run_times_out_without_a_merged_deploy_run(monkeypatch): +async def test_wait_deploy_run_rejects_a_user_scoped_client(monkeypatch): + """The 2026-07-16 client must fail loudly instead of polling a blind endpoint. + + Proven against the same fake API: as a user it sees nothing, so a wait would + only ever end in a false timeout. + """ monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") + runs = [_deploy_run("deploy-poll-ea0bed35")] + ctx = {"project_id": "project-1", "story_id": "story-1"} - async def get(url, params=None, headers=None): - return httpx.Response(200, json=[], request=httpx.Request("GET", "http://test/api/runs/")) + async with _runs_client(runs, as_user=True) as api_as_user: + blind = await api_as_user.get( + "/api/runs/", params={"story_id": "story-1", "run_type": "deploy"} + ) + assert blind.json() == [] - ctx = {"project_id": "project-1"} - run = await pipeline_helpers.wait_deploy_run( - SimpleNamespace(get=get), ctx, timeout=0.001, poll_interval=0 - ) + with pytest.raises(RuntimeError, match=pipeline_helpers.USER_AUTH_HEADER): + await pipeline_helpers.wait_deploy_run(api_as_user, ctx, timeout=1) + + assert "deploy_run_id" not in ctx + + +@pytest.mark.asyncio +async def test_wait_deploy_run_ignores_another_storys_deploy_run(monkeypatch): + """A project outlives one story: an earlier story's deploy run also carries a + merged head_sha, and deploying that SHA is not what this mega asserts about. + + The run is checked against this story even if the API were to widen its + filter, so the contract check cannot silently follow a foreign deploy. + """ + monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") + runs = [_deploy_run("deploy-poll-other", story_id="story-earlier", head_sha="dead00")] + ctx = {"project_id": "project-1", "story_id": "story-1"} + + async with httpx.AsyncClient( + base_url="http://test", + # A deliberately over-wide API: it ignores the story_id filter entirely. + transport=httpx.MockTransport(lambda request: httpx.Response(200, json=runs)), + headers=pipeline_helpers.internal_headers(), + ) as api_internal: + run = await pipeline_helpers.wait_deploy_run( + api_internal, ctx, timeout=0.001, poll_interval=0 + ) + + assert run is None + assert "deploy_run_id" not in ctx + assert "story-1" in ctx["deploy_run_error"] + + +@pytest.mark.asyncio +async def test_wait_deploy_run_rejects_a_deploy_run_without_a_merged_sha(monkeypatch): + """An engineering-triggered deploy run records no head_sha, so the mega has no + ref to check the environment contract at and must not accept it.""" + monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") + ctx = {"project_id": "project-1", "story_id": "story-1"} + + async with _runs_client([_deploy_run("deploy-eng-1", head_sha=None)]) as api_internal: + run = await pipeline_helpers.wait_deploy_run( + api_internal, ctx, timeout=0.001, poll_interval=0 + ) + + assert run is None + assert "deploy_run_id" not in ctx + assert "no deploy run with a merged head_sha" in ctx["deploy_run_error"] + + +@pytest.mark.asyncio +async def test_wait_deploy_run_times_out_without_a_merged_deploy_run(monkeypatch): + monkeypatch.setenv("INTERNAL_API_KEY", "test-internal-key") + ctx = {"project_id": "project-1", "story_id": "story-1"} + + async with _runs_client([]) as api_internal: + run = await pipeline_helpers.wait_deploy_run( + api_internal, ctx, timeout=0.001, poll_interval=0 + ) assert run is None assert "no deploy run with a merged head_sha" in ctx["deploy_run_error"]