Qualify readonly terminal reports from protected workflow revision

scripts/read_binance_account_facts.py

@@ -209,7 +209,11 @@ def _ensure_no_active_runtime_run(*, token: str, api_url: str) -> None:
             raise _stop("runtime_activity_present")
 
 
-def _ensure_latest_runtime_run(*, run_id: str, token: str, api_url: str) -> None:
+def _ensure_latest_runtime_run(
+    *, run_id: str, token: str, api_url: str,
+    expected_workflow_sha: str = LEGACY_RUNTIME_WORKFLOW_SHA,
+    require_attempt_one: bool = False,
+) -> None:
     latest = _api_json(
         f"{api_url}/repos/{REPOSITORY}/actions/workflows/main.yml/runs"
         f"?per_page=1",
@@ -227,7 +231,10 @@ def _ensure_latest_runtime_run(*, run_id: str, token: str, api_url: str) -> None
         or row.get("status") != "completed"
         or row.get("conclusion") != "success"
         or row.get("head_branch") != RUNTIME_BRANCH
-        or row.get("head_sha") != LEGACY_RUNTIME_WORKFLOW_SHA
+        or row.get("head_sha") != expected_workflow_sha
+        or (require_attempt_one and (
+            type(row.get("run_attempt")) is not int or row.get("run_attempt") != 1
+        ))
         or not isinstance(row.get("repository"), Mapping)
         or row["repository"].get("full_name") != REPOSITORY
     ):
@@ -252,6 +259,54 @@ def _parse_github_time(value: Any) -> datetime:
     return parsed.astimezone(timezone.utc)
 
 
+def _validate_terminal_runtime_steps(
+    *, run_id: str, token: str, api_url: str
+) -> datetime:
+    jobs = _api_json(
+        f"{api_url}/repos/{REPOSITORY}/actions/runs/{quote(str(run_id))}/jobs?per_page=100",
+        token,
+    )
+    rows = jobs.get("jobs") if isinstance(jobs, Mapping) else None
+    if (
+        not isinstance(rows, list) or type(jobs.get("total_count")) is not int
+        or jobs["total_count"] > 100
+    ):
+        raise _stop("parent_run_jobs_unverified")
+    deploy_rows = [row for row in rows if isinstance(row, Mapping) and row.get("name") == "deploy"]
+    if len(deploy_rows) != 1:
+        raise _stop("parent_deploy_unverified")
+    return _validate_terminal_deploy(deploy_rows[0])
+
+
+def _validate_terminal_deploy(deploy: Mapping[str, Any]) -> datetime:
+    steps = deploy.get("steps")
+    if (
+        deploy.get("status") != "completed" or deploy.get("conclusion") != "success"
+        or not isinstance(steps, list)
+    ):
+        raise _stop("parent_deploy_unverified")
+    strategy = [step for step in steps if isinstance(step, Mapping) and step.get("name") == "4. Run trading strategy"]
+    report = [step for step in steps if isinstance(step, Mapping) and step.get("name") == "5. Stage execution report for isolated log publisher"]
+    if (
+        len(strategy) != 1 or strategy[0].get("status") != "completed"
+        or strategy[0].get("conclusion") != "success"
+        or len(report) != 1 or report[0].get("status") != "completed"
+        or report[0].get("conclusion") != "success"
+    ):
+        raise _stop("parent_strategy_or_report_unverified")
+    deploy_started = _parse_github_time(deploy.get("started_at"))
+    strategy_started = _parse_github_time(strategy[0].get("started_at"))
+    strategy_finished = _parse_github_time(strategy[0].get("completed_at"))
+    report_started = _parse_github_time(report[0].get("started_at"))
+    report_finished = _parse_github_time(report[0].get("completed_at"))
+    if (
+        strategy_started < deploy_started or strategy_finished < strategy_started
+        or report_started < strategy_finished or report_finished < report_started
+    ):
+        raise _stop("parent_run_jobs_unverified")
+    return strategy_finished
+
+
 def _ensure_no_other_active_parent_run(
     *, run_id: str, token: str, api_url: str
 ) -> None:
@@ -355,31 +410,7 @@ def verify_parent_run(
     if len(deploy_jobs) != 1:
         raise _stop("parent_deploy_unverified")
     deploy = deploy_jobs[0]
-    steps = deploy.get("steps")
-    if (
-        deploy.get("status") != "completed" or deploy.get("conclusion") != "success"
-        or not isinstance(steps, list)
-    ):
-        raise _stop("parent_deploy_unverified")
-    strategy_steps = [step for step in steps if isinstance(step, Mapping) and step.get("name") == "4. Run trading strategy"]
-    report_steps = [step for step in steps if isinstance(step, Mapping) and step.get("name") == "5. Stage execution report for isolated log publisher"]
-    if (
-        len(strategy_steps) != 1 or strategy_steps[0].get("status") != "completed"
-        or strategy_steps[0].get("conclusion") != "success"
-        or len(report_steps) != 1 or report_steps[0].get("status") != "completed"
-        or report_steps[0].get("conclusion") != "success"
-    ):
-        raise _stop("parent_strategy_or_report_unverified")
-    deploy_started = _parse_github_time(deploy.get("started_at"))
-    strategy_started = _parse_github_time(strategy_steps[0].get("started_at"))
-    strategy_finished = _parse_github_time(strategy_steps[0].get("completed_at"))
-    report_started = _parse_github_time(report_steps[0].get("started_at"))
-    report_finished = _parse_github_time(report_steps[0].get("completed_at"))
-    if (
-        strategy_started < deploy_started or strategy_finished < strategy_started
-        or report_started < strategy_finished or report_finished < report_started
-    ):
-        raise _stop("parent_run_jobs_unverified")
+    strategy_finished = _validate_terminal_deploy(deploy)
 
     # The protected setting and report binding identify the intended app, but
     # only the Runtime release-selection log proves what this parent ran.
@@ -405,7 +436,6 @@ def verify_parent_run(
         artifact.get("expired") is not False
         or type(artifact.get("size_in_bytes")) is not int or artifact["size_in_bytes"] <= 0
         or _parse_github_time(artifact.get("created_at")) < strategy_finished
-        or _parse_github_time(artifact.get("created_at")) < deploy_started
     ):
         raise _stop("trigger_report_artifact_missing")
     return {"run_id": str(run_id), "report_artifact": artifact_name}
@@ -429,16 +459,18 @@ def verify_current_source_from_env(env: Mapping[str, str]) -> None:
             api_url=str(env.get("GITHUB_API_URL") or "https://api.github.com"),
         )
     elif mode == "legacy_terminal":
-        verify_source_is_current(
-            run_id=str(env.get("SOURCE_RUN_ID") or ""),
-            token=str(env.get("GITHUB_TOKEN") or ""),
-            api_url=str(env.get("GITHUB_API_URL") or "https://api.github.com"),
-        )
+        verify_terminal_source_from_env(env)
     else:
         raise _stop("source_mode_invalid")
 
 
-def verify_trigger_run(*, run_id: str, repository: str, token: str, api_url: str) -> dict[str, str]:
+def verify_trigger_run(
+    *, run_id: str, repository: str, token: str, api_url: str,
+    expected_workflow_sha: str = LEGACY_RUNTIME_WORKFLOW_SHA,
+    expected_ref: str | None = None,
+    require_attempt_one: bool = False,
+    require_strategy_steps: bool = False,
+) -> dict[str, str]:
     """Validate source run metadata, selected application SHA, and report artifact."""
     if repository != REPOSITORY or not re.fullmatch(r"[1-9][0-9]{0,19}", str(run_id)):
         raise _stop("trigger_identity_invalid")
@@ -447,14 +479,18 @@ def verify_trigger_run(*, run_id: str, repository: str, token: str, api_url: str
     head_repo = run.get("head_repository") if isinstance(run, dict) else None
     base_repo = run.get("repository") if isinstance(run, dict) else None
     if (
-        run.get("id") != int(run_id)
+        type(run.get("id")) is not int or run.get("id") != int(run_id)
         or run.get("name") != RUNTIME_WORKFLOW_NAME
         or str(run.get("path") or "").split("@", 1)[0] != ".github/workflows/main.yml"
         or run.get("event") != "workflow_dispatch"
         or run.get("status") != "completed"
         or run.get("conclusion") != "success"
         or run.get("head_branch") != RUNTIME_BRANCH
-        or run.get("head_sha") != LEGACY_RUNTIME_WORKFLOW_SHA
+        or run.get("head_sha") != expected_workflow_sha
+        or (require_attempt_one and (
+            type(run.get("run_attempt")) is not int or run.get("run_attempt") != 1
+        ))
+        or (expected_ref is not None and expected_ref != f"refs/heads/{RUNTIME_BRANCH}")
         or not isinstance(head_repo, Mapping)
         or head_repo.get("full_name") != REPOSITORY
         or not isinstance(base_repo, Mapping)
@@ -463,19 +499,68 @@ def verify_trigger_run(*, run_id: str, repository: str, token: str, api_url: str
         raise _stop("trigger_identity_mismatch")
     _validate_release_log(str(run_id), token, api_url)
     _ensure_no_active_runtime_run(token=token, api_url=api_url)
+    _ensure_latest_runtime_run(
+        run_id=str(run_id), token=token, api_url=api_url,
+        expected_workflow_sha=expected_workflow_sha,
+        require_attempt_one=require_attempt_one,
+    )
+    strategy_finished = (
+        _validate_terminal_runtime_steps(run_id=str(run_id), token=token, api_url=api_url)
+        if require_strategy_steps else None
+    )
     artifacts = _api_json(
         f"{api_url}/repos/{REPOSITORY}/actions/runs/{quote(str(run_id))}/artifacts?per_page=100",
         token,
     )
     rows = artifacts.get("artifacts") if isinstance(artifacts, dict) else None
     expected_name = f"{EXPECTED_REPORT_ARTIFACT_PREFIX}{run_id}"
     matches = [row for row in rows if isinstance(row, Mapping) and row.get("name") == expected_name] if isinstance(rows, list) else []
-    if len(matches) != 1 or matches[0].get("expired") is not False or matches[0].get("size_in_bytes", 0) <= 0:
+    if (
+        len(matches) != 1 or matches[0].get("expired") is not False
+        or type(matches[0].get("size_in_bytes")) is not int or matches[0]["size_in_bytes"] <= 0
+        or (
+            strategy_finished is not None
+            and _parse_github_time(matches[0].get("created_at")) < strategy_finished
+        )
+    ):
         raise _stop("trigger_report_artifact_missing")
-    verify_source_is_current(run_id=str(run_id), token=token, api_url=api_url)
+    _ensure_no_active_runtime_run(token=token, api_url=api_url)
+    _ensure_latest_runtime_run(
+        run_id=str(run_id), token=token, api_url=api_url,
+        expected_workflow_sha=expected_workflow_sha,
+        require_attempt_one=require_attempt_one,
+    )
     return {"run_id": str(run_id), "report_artifact": expected_name}
 
 
+def verify_terminal_source_from_env(env: Mapping[str, str]) -> dict[str, str]:
+    run_id = str(env.get("SOURCE_RUN_ID") or "")
+    repository = str(env.get("GITHUB_REPOSITORY") or "")
+    token = str(env.get("GITHUB_TOKEN") or "")
+    api_url = str(env.get("GITHUB_API_URL") or "https://api.github.com")
+    pinned_workflow_sha = str(env.get("BINANCE_RUNTIME_WORKFLOW_SHA") or "")
+    if repository != REPOSITORY or not re.fullmatch(r"[1-9][0-9]{0,19}", run_id):
+        raise _stop("trigger_identity_invalid")
+    if _GIT_SHA.fullmatch(pinned_workflow_sha):
+        run = _api_json(
+            f"{api_url}/repos/{REPOSITORY}/actions/runs/{quote(run_id)}", token
+        )
+        if isinstance(run, Mapping) and run.get("head_sha") == pinned_workflow_sha:
+            return verify_trigger_run(
+                run_id=run_id,
+                repository=repository,
+                token=token,
+                api_url=api_url,
+                expected_workflow_sha=pinned_workflow_sha,
+                expected_ref=str(env.get("GITHUB_REF") or ""),
+                require_attempt_one=True,
+                require_strategy_steps=True,
+            )
+    return verify_trigger_run(
+        run_id=run_id, repository=repository, token=token, api_url=api_url
+    )
+
+
 def _valid_order_result_tree(value: Any, parent_key: str = "") -> bool:
     if isinstance(value, Mapping):
         for key, child in value.items():
@@ -850,12 +935,7 @@ def preflight(*, env: Mapping[str, str], output: Path) -> None:
             api_url=str(env.get("GITHUB_API_URL") or "https://api.github.com"),
         )
     elif mode == "legacy_terminal":
-        result = verify_trigger_run(
-            run_id=str(env.get("SOURCE_RUN_ID") or ""),
-            repository=str(env.get("GITHUB_REPOSITORY") or ""),
-            token=str(env.get("GITHUB_TOKEN") or ""),
-            api_url=str(env.get("GITHUB_API_URL") or "https://api.github.com"),
-        )
+        result = verify_terminal_source_from_env(env)
     else:
         raise _stop("source_mode_invalid")
     output.parent.mkdir(parents=True, exist_ok=True)

tests/test_binance_account_facts.py

@@ -17,6 +17,7 @@
 
 
 APP_SHA = "8cb56617115fa45028e34d788e71884b6a303d77"
+PROTECTED_RUNTIME_SHA = "d" * 40
 READER_SHA = "a" * 40
 UID_DIGEST = "b" * 64
 START = datetime(2026, 10, 2, 10, 0, tzinfo=timezone.utc)
@@ -168,6 +169,114 @@ def _install_parent_release_log(monkeypatch, reader, selected_sha=APP_SHA):
     monkeypatch.setattr(reader, "_read_job_log_text", lambda *_args: log_text.encode())
 
 
+def _current_terminal_run_api(url, _token, mutation=None):
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    source_sha = "9cfcf0531d1ea176e6f26590cf15edbd31bd6567" if mutation == "legacy_old" else PROTECTED_RUNTIME_SHA
+    if "/actions/workflows/main.yml/runs?status=" in url:
+        return {"total_count": 0, "workflow_runs": []}
+    if parsed.path.endswith("/actions/workflows/main.yml/runs"):
+        run_id = 602 if mutation == "newer_terminal" else 601
+        return {"total_count": 2, "workflow_runs": [{
+            "id": run_id, "name": "Runtime",
+            "path": ".github/workflows/main.yml@refs/heads/runtime-production",
+            "event": "workflow_dispatch", "status": "completed", "conclusion": "success",
+            "head_branch": "runtime-production", "head_sha": source_sha,
+            "run_attempt": 1, "repository": {"full_name": "QuantStrategyLab/BinancePlatform"},
+        }]}
+    if parsed.path.endswith("/actions/runs/601"):
+        attempt = 2 if mutation == "attempt_two" else 1
+        return {
+            "id": 601, "run_attempt": attempt, "name": "Runtime",
+            "path": ".github/workflows/main.yml@refs/heads/runtime-production",
+            "event": "workflow_dispatch", "status": "completed", "conclusion": "success",
+            "head_branch": "runtime-production", "head_sha": source_sha,
+            "repository": {"full_name": "QuantStrategyLab/BinancePlatform"},
+            "head_repository": {"full_name": "QuantStrategyLab/BinancePlatform"},
+        }
+    if parsed.path.endswith("/actions/runs/601/jobs"):
+        upload_conclusion = "failure" if mutation == "failed_report" else "success"
+        return {"total_count": 1, "jobs": [{
+            "id": 701, "name": "deploy", "status": "completed", "conclusion": "success",
+            "started_at": "2026-10-02T11:00:00Z",
+            "steps": [
+                {"name": "Resolve approved runtime release SHA", "status": "completed", "conclusion": "success"},
+                {"name": "4. Run trading strategy", "status": "completed", "conclusion": "success", "started_at": "2026-10-02T11:01:00Z", "completed_at": "2026-10-02T11:05:00Z"},
+                {"name": "5. Stage execution report for isolated log publisher", "status": "completed", "conclusion": upload_conclusion, "started_at": "2026-10-02T11:05:01Z", "completed_at": "2026-10-02T11:06:00Z"},
+            ],
+        }]}
+    if parsed.path.endswith("/actions/runs/601/artifacts"):
+        created_at = "2026-10-02T11:04:59Z" if mutation == "early_artifact" else "2026-10-02T11:05:10Z"
+        return {"artifacts": [{
+            "name": "binance-execution-report-601", "expired": False,
+            "size_in_bytes": 123, "created_at": created_at,
+        }]}
+    raise AssertionError("unexpected synthetic GitHub API path")
+
+
+@pytest.mark.parametrize(
+    "mutation,reason",
+    [
+        ("wrong_application", "trigger_release_mismatch"),
+        ("attempt_two", "trigger_identity_mismatch"),
+        ("failed_report", "parent_strategy_or_report_unverified"),
+        ("early_artifact", "trigger_report_artifact_missing"),
+        ("newer_terminal", "runtime_run_not_latest_success"),
+    ],
+)
+def test_manual_terminal_accepts_only_current_pinned_runtime_and_8cb_app(monkeypatch, mutation, reason):
+    from scripts import read_binance_account_facts as reader
+
+    monkeypatch.setattr(reader, "_api_json", lambda url, token: _current_terminal_run_api(url, token, mutation))
+    _install_parent_release_log(
+        monkeypatch, reader, selected_sha="e" * 40 if mutation == "wrong_application" else APP_SHA
+    )
+    env = {
+        "SOURCE_RUN_ID": "601",
+        "GITHUB_REPOSITORY": "QuantStrategyLab/BinancePlatform",
+        "GITHUB_REF": "refs/heads/runtime-production",
+        "BINANCE_RUNTIME_WORKFLOW_SHA": PROTECTED_RUNTIME_SHA,
+        "GITHUB_TOKEN": "synthetic-token",
+        "GITHUB_API_URL": "https://api.example",
+    }
+    if reason:
+        with pytest.raises(reader.ReaderError, match=reason):
+            reader.verify_terminal_source_from_env(env)
+
+
+def test_manual_terminal_accepts_protected_sha_when_runtime_selected_8cb(monkeypatch):
+    from scripts import read_binance_account_facts as reader
+
+    monkeypatch.setattr(reader, "_api_json", _current_terminal_run_api)
+    _install_parent_release_log(monkeypatch, reader)
+    result = reader.verify_terminal_source_from_env({
+        "SOURCE_RUN_ID": "601",
+        "GITHUB_REPOSITORY": "QuantStrategyLab/BinancePlatform",
+        "GITHUB_REF": "refs/heads/runtime-production",
+        "BINANCE_RUNTIME_WORKFLOW_SHA": PROTECTED_RUNTIME_SHA,
+        "GITHUB_TOKEN": "synthetic-token",
+        "GITHUB_API_URL": "https://api.example",
+    })
+    assert result == {"run_id": "601", "report_artifact": "binance-execution-report-601"}
+
+
+def test_manual_terminal_keeps_fixed_legacy_9cfc_path(monkeypatch):
+    from scripts import read_binance_account_facts as reader
+
+    monkeypatch.setattr(reader, "_api_json", lambda url, token: _current_terminal_run_api(url, token, "legacy_old"))
+    _install_parent_release_log(monkeypatch, reader)
+    result = reader.verify_terminal_source_from_env({
+        "SOURCE_RUN_ID": "601",
+        "GITHUB_REPOSITORY": "QuantStrategyLab/BinancePlatform",
+        "GITHUB_REF": "refs/heads/runtime-production",
+        "BINANCE_RUNTIME_WORKFLOW_SHA": PROTECTED_RUNTIME_SHA,
+        "GITHUB_TOKEN": "synthetic-token",
+        "GITHUB_API_URL": "https://api.example",
+    })
+    assert result == {"run_id": "601", "report_artifact": "binance-execution-report-601"}
+
+
 def test_same_parent_runtime_read_is_allowed_only_for_exact_in_progress_run(monkeypatch):
     from scripts import read_binance_account_facts as reader