ci: add opt-in research publisher identity verification

.github/workflows/theme_momentum_snapshot.yml

@@ -3,6 +3,11 @@ name: Theme Momentum Snapshot
 on:
   workflow_dispatch:
     inputs:
+      identity_only:
+        description: "Verify the configured publisher identity only; no collection, data changes, or PR. Main only."
+        required: false
+        default: false
+        type: boolean
       as_of:
         description: "Optional YYYY-MM-DD cutoff. Empty uses latest available/downloaded price date."
         required: false
@@ -45,13 +50,141 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:
+      # Manual diagnostic only. Normal generation and publication behavior is unchanged.
+      - name: Check trusted publisher identity context
+        id: publisher_identity_preflight
+        if: ${{ github.event_name == 'workflow_dispatch' && inputs.identity_only }}
+        env:
+          RESEARCH_PUBLISHER_APP_ID: ${{ vars.RESEARCH_PUBLISHER_APP_ID }}
+        run: |
+          python - <<'PY'
+          import os
+          from pathlib import Path
+
+          expected_repository = "QuantStrategyLab/ResearchSignalContextPipelines"
+          expected_workflow_ref = expected_repository + "/.github/workflows/theme_momentum_snapshot.yml@refs/heads/main"
+          app_id_matches = os.environ.get("RESEARCH_PUBLISHER_APP_ID") == "5214420"
+          repository_id = os.environ.get("GITHUB_REPOSITORY_ID", "")
+          context_matches = (
+              os.environ.get("GITHUB_EVENT_NAME") == "workflow_dispatch"
+              and os.environ.get("GITHUB_REF") == "refs/heads/main"
+              and os.environ.get("GITHUB_REPOSITORY") == expected_repository
+              and os.environ.get("GITHUB_WORKFLOW_REF") == expected_workflow_ref
+              and repository_id.isascii() and repository_id.isdecimal()
+              and int(repository_id) > 0
+          )
+          with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output:
+              output.write(f"app_id_matches={str(app_id_matches).lower()}\n")
+          if not (app_id_matches and context_matches):
+              print("PUBLISHER_IDENTITY_PREFLIGHT_FAILED")
+              raise SystemExit(1)
+          print("PUBLISHER_IDENTITY_PREFLIGHT_OK")
+          PY
+      - name: Create single-repository publisher identity token
+        id: publisher_identity_token
+        if: ${{ success() && github.event_name == 'workflow_dispatch' && inputs.identity_only }}
+        uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0
+        with:
+          app-id: ${{ vars.RESEARCH_PUBLISHER_APP_ID }}
+          private-key: ${{ secrets.RESEARCH_PUBLISHER_APP_PRIVATE_KEY }}
+          owner: QuantStrategyLab
+          repositories: ResearchSignalContextPipelines
+          permission-contents: write
+          permission-pull-requests: write
+      - name: Verify publisher installation and repository scope
+        id: publisher_identity_verify
+        if: ${{ success() && github.event_name == 'workflow_dispatch' && inputs.identity_only }}
+        env:
+          RESEARCH_PUBLISHER_INSTALLATION_ID: ${{ steps.publisher_identity_token.outputs.installation-id }}
+          RESEARCH_PUBLISHER_TOKEN: ${{ steps.publisher_identity_token.outputs.token }}
+        run: |
+          python - <<'PY'
+          import json
+          import os
+          import urllib.request
+          from pathlib import Path
+
+          class NoRedirect(urllib.request.HTTPRedirectHandler):
+              def redirect_request(self, req, fp, code, msg, headers, newurl):
+                  return None
+
+          installation_matches = os.environ.get("RESEARCH_PUBLISHER_INSTALLATION_ID") == "168603414"
+          scope_matches = False
+          verified = False
+          try:
+              token = os.environ.get("RESEARCH_PUBLISHER_TOKEN", "")
+              if installation_matches and token:
+                  request = urllib.request.Request(
+                      "https://api.github.com/installation/repositories?per_page=100",
+                      headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
+                      method="GET",
+                  )
+                  with urllib.request.build_opener(NoRedirect()).open(request, timeout=20) as response:
+                      raw = response.read(1024 * 1024 + 1)
+                      if response.status == 200 and not response.headers.get("Link") and len(raw) <= 1024 * 1024:
+                          payload = json.loads(raw)
+                          if type(payload) is dict and type(payload.get("total_count")) is int and payload["total_count"] == 1:
+                              repositories = payload.get("repositories")
+                              if type(repositories) is list and len(repositories) == 1 and type(repositories[0]) is dict:
+                                  repository = repositories[0]
+                                  scope_matches = (
+                                      repository.get("full_name") == "QuantStrategyLab/ResearchSignalContextPipelines"
+                                      and type(repository.get("id")) is int
+                                      and repository["id"] == int(os.environ["GITHUB_REPOSITORY_ID"])
+                                  )
+                  verified = installation_matches and scope_matches
+          except Exception:
+              # No API payload, exception text, request headers, token or JWT is logged.
+              verified = False
+          with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output:
+              for key, value in (
+                  ("installation_matches", installation_matches),
+                  ("single_repo_scope_matches", scope_matches),
+                  ("identity_verified", verified),
+              ):
+                  output.write(f"{key}={str(value).lower()}\n")
+          if not verified:
+              print("PUBLISHER_IDENTITY_VERIFICATION_FAILED")
+              raise SystemExit(1)
+          print("PUBLISHER_IDENTITY_VERIFIED")
+          PY
+      - name: Report publisher identity-only result
+        id: publisher_identity_summary
+        if: ${{ always() && github.event_name == 'workflow_dispatch' && inputs.identity_only }}
+        env:
+          APP_ID_MATCHES: ${{ steps.publisher_identity_preflight.outputs.app_id_matches }}
+          INSTALLATION_MATCHES: ${{ steps.publisher_identity_verify.outputs.installation_matches }}
+          SINGLE_REPO_SCOPE_MATCHES: ${{ steps.publisher_identity_verify.outputs.single_repo_scope_matches }}
+          IDENTITY_VERIFIED: ${{ steps.publisher_identity_verify.outputs.identity_verified }}
+        run: |
+          python - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+
+          result = {
+              "app_id_matches": os.environ.get("APP_ID_MATCHES") == "true",
+              "installation_matches": os.environ.get("INSTALLATION_MATCHES") == "true",
+              "single_repo_scope_matches": os.environ.get("SINGLE_REPO_SCOPE_MATCHES") == "true",
+              "identity_verified": os.environ.get("IDENTITY_VERIFIED") == "true",
+              "published": False,
+          }
+          serialized = json.dumps(result, sort_keys=True)
+          print(serialized)
+          with Path(os.environ["GITHUB_STEP_SUMMARY"]).open("a", encoding="utf-8") as output:
+              output.write("Publisher identity-only result (no data published):\n" + serialized + "\n")
+          PY
       - uses: actions/checkout@v6
+        if: ${{ !inputs.identity_only }}
       - uses: actions/setup-python@v6
+        if: ${{ !inputs.identity_only }}
         with:
           python-version: "3.11"
       - name: Install package
+        if: ${{ !inputs.identity_only }}
         run: python -m pip install -e .
       - name: Build theme momentum snapshot
+        if: ${{ !inputs.identity_only }}
         env:
           INPUT_AS_OF: ${{ github.event.inputs.as_of || '' }}
           PRICES_PATH: ${{ github.event.inputs.prices_path || '' }}
@@ -78,6 +211,7 @@ jobs:
             echo '```'
           } >> "$GITHUB_STEP_SUMMARY"
       - name: Commit snapshot
+        if: ${{ !inputs.identity_only }}
         env:
           COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }}
         run: |
@@ -109,6 +243,7 @@ jobs:
             fi
           fi
       - name: Upload theme momentum artifact
+        if: ${{ !inputs.identity_only }}
         uses: actions/upload-artifact@v7
         with:
           name: theme-momentum-snapshot

tests/test_theme_momentum_snapshot_workflow.py

@@ -1,6 +1,14 @@
 from __future__ import annotations
 
 from pathlib import Path
+import copy
+import io
+import re
+import textwrap
+import urllib.request
+
+import json
+import pytest
 
 
 def test_theme_momentum_snapshot_workflow_tolerates_repo_rule_push_rejection() -> None:
@@ -11,3 +19,214 @@ def test_theme_momentum_snapshot_workflow_tolerates_repo_rule_push_rejection() -
     assert "grep -Eqi 'required status check' \"${push_log}\"" in workflow
     assert "GH013|repository rule violations|required status check" not in workflow
     assert "git push 2>\"${push_log}\"" in workflow
+
+
+IDENTITY_REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines"
+IDENTITY_WORKFLOW = "theme_momentum_snapshot.yml"
+
+
+# Identity-only tests execute the reviewed inline Python with mocked HTTP only.
+
+
+IDENTITY_ACTION = "actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1"
+IDENTITY_URL = "https://api.github.com/installation/repositories?per_page=100"
+
+
+def identity_workflow():
+    return Path(__file__).parents[1] / ".github" / "workflows" / IDENTITY_WORKFLOW
+
+
+def identity_steps():
+    return ["      - " + block for block in re.split(r"(?m)^      - ", identity_workflow().read_text())[1:]]
+
+
+def identity_step(step_id):
+    return next(block for block in identity_steps() if f"        id: {step_id}\n" in block)
+
+
+def identity_python(step_id):
+    block = identity_step(step_id).split("        run: |\n", 1)[1]
+    script = textwrap.dedent(block)
+    assert script.startswith("python - <<'PY'\n")
+    return script.removeprefix("python - <<'PY'\n").rsplit("\nPY", 1)[0]
+
+
+def identity_env(monkeypatch, tmp_path):
+    values = {
+        "GITHUB_EVENT_NAME": "workflow_dispatch",
+        "GITHUB_REF": "refs/heads/main",
+        "GITHUB_REPOSITORY": IDENTITY_REPOSITORY,
+        "GITHUB_REPOSITORY_ID": "123456",
+        "GITHUB_WORKFLOW_REF": f"{IDENTITY_REPOSITORY}/.github/workflows/{IDENTITY_WORKFLOW}@refs/heads/main",
+        "RESEARCH_PUBLISHER_APP_ID": "5214420",
+        "RESEARCH_PUBLISHER_INSTALLATION_ID": "168603414",
+        "RESEARCH_PUBLISHER_TOKEN": "synthetic-test-token-never-real",
+        "GITHUB_OUTPUT": str(tmp_path / "outputs"),
+        "GITHUB_STEP_SUMMARY": str(tmp_path / "summary"),
+    }
+    for key, value in values.items():
+        monkeypatch.setenv(key, value)
+    return values
+
+
+def run_identity_python(step_id):
+    exec(compile(identity_python(step_id), f"<mocked-{step_id}>", "exec"), {"__name__": "__main__"})
+
+
+def identity_payload():
+    return {"total_count": 1, "repositories": [{"id": 123456, "full_name": IDENTITY_REPOSITORY}]}
+
+
+def mock_identity_http(monkeypatch, payload=None, *, body=None, status=200, headers=None, error=None):
+    calls = []
+    raw = body if body is not None else json.dumps(payload or identity_payload()).encode()
+
+    class Response(io.BytesIO):
+        def __init__(self):
+            super().__init__(raw)
+            self.status = status
+            self.headers = headers or {}
+
+    class Opener:
+        def open(self, request, timeout):
+            calls.append(request)
+            assert request.full_url == IDENTITY_URL
+            assert request.get_method() == "GET"
+            assert request.data is None
+            assert request.get_header("Authorization") == "Bearer synthetic-test-token-never-real"
+            assert timeout == 20
+            if error:
+                raise error
+            return Response()
+
+    def build_opener(handler):
+        assert isinstance(handler, urllib.request.HTTPRedirectHandler)
+        assert handler.redirect_request(None, None, 302, "", {}, "https://untrusted.example/") is None
+        return Opener()
+
+    monkeypatch.setattr(urllib.request, "build_opener", build_opener)
+    return calls
+
+
+def test_identity_only_is_opt_in_and_isolates_all_normal_steps():
+    text = identity_workflow().read_text()
+    assert "      identity_only:\n" in text
+    assert "        default: false\n        type: boolean" in text
+    steps = identity_steps()
+    identity = [step for step in steps if "        id: publisher_identity_" in step]
+    assert len(identity) == 4
+    for step in steps:
+        if step not in identity:
+            assert "if: ${{ !inputs.identity_only }}" in step or "if: ${{ always() && !inputs.identity_only }}" in step
+    for step in identity:
+        assert "github.event_name == 'workflow_dispatch' && inputs.identity_only" in step
+    assert "always()" in identity_step("publisher_identity_summary")
+    mint = identity_step("publisher_identity_token")
+    assert IDENTITY_ACTION in mint
+    assert f"          repositories: {IDENTITY_REPOSITORY.split('/')[1]}\n" in mint
+    assert "          owner: QuantStrategyLab\n" in mint
+    assert "          permission-contents: write\n" in mint
+    assert "          permission-pull-requests: write\n" in mint
+    assert mint.count("          permission-") == 2
+    assert "skip-token-revoke" not in mint
+    assert "CROSS_REPO" not in "".join(identity)
+    assert "persist-credentials" not in "".join(identity)
+    assert "steps.publisher_identity_token.outputs.token" in identity_step("publisher_identity_verify")
+
+
+def test_identity_preflight_succeeds_without_network(monkeypatch, tmp_path):
+    values = identity_env(monkeypatch, tmp_path)
+    monkeypatch.setattr(urllib.request, "build_opener", lambda *args: pytest.fail("preflight must not use HTTP"))
+    run_identity_python("publisher_identity_preflight")
+    assert Path(values["GITHUB_OUTPUT"]).read_text() == "app_id_matches=true\n"
+
+
+@pytest.mark.parametrize(("key", "value"), [
+    ("GITHUB_EVENT_NAME", "schedule"),
+    ("GITHUB_REF", "refs/heads/unreviewed"),
+    ("GITHUB_REPOSITORY", "other/other"),
+    ("GITHUB_REPOSITORY_ID", "not-an-id"),
+    ("GITHUB_WORKFLOW_REF", "other/other/.github/workflows/test.yml@refs/heads/main"),
+    ("RESEARCH_PUBLISHER_APP_ID", "123"),
+])
+def test_identity_preflight_fails_closed(monkeypatch, tmp_path, capsys, key, value):
+    identity_env(monkeypatch, tmp_path)
+    monkeypatch.setenv(key, value)
+    monkeypatch.setattr(urllib.request, "build_opener", lambda *args: pytest.fail("preflight must not use HTTP"))
+    with pytest.raises(SystemExit, match="1"):
+        run_identity_python("publisher_identity_preflight")
+    assert "PUBLISHER_IDENTITY_PREFLIGHT_FAILED" in capsys.readouterr().out
+
+
+def test_identity_checks_single_repository_without_writing(monkeypatch, tmp_path, capsys):
+    values = identity_env(monkeypatch, tmp_path)
+    calls = mock_identity_http(monkeypatch)
+    run_identity_python("publisher_identity_verify")
+    assert len(calls) == 1
+    output = Path(values["GITHUB_OUTPUT"]).read_text()
+    assert "installation_matches=true\n" in output
+    assert "single_repo_scope_matches=true\n" in output
+    assert "identity_verified=true\n" in output
+    assert "synthetic-test-token" not in capsys.readouterr().out + output
+
+
+@pytest.mark.parametrize("mutation", ["extra_repo", "wrong_name", "wrong_id", "bool_id", "bool_count", "wrong_count", "no_repo"])
+def test_identity_rejects_wrong_repository_scope(monkeypatch, tmp_path, capsys, mutation):
+    values = identity_env(monkeypatch, tmp_path)
+    payload = copy.deepcopy(identity_payload())
+    if mutation == "extra_repo":
+        payload["repositories"].append({"id": 44, "full_name": "other/other"})
+    elif mutation == "wrong_name":
+        payload["repositories"][0]["full_name"] = "other/other"
+    elif mutation == "wrong_id":
+        payload["repositories"][0]["id"] = 44
+    elif mutation == "bool_id":
+        payload["repositories"][0]["id"] = True
+    elif mutation == "bool_count":
+        payload["total_count"] = True
+    elif mutation == "wrong_count":
+        payload["total_count"] = 2
+    else:
+        payload["repositories"] = []
+    mock_identity_http(monkeypatch, payload)
+    with pytest.raises(SystemExit, match="1"):
+        run_identity_python("publisher_identity_verify")
+    assert "identity_verified=false" in Path(values["GITHUB_OUTPUT"]).read_text()
+    assert "synthetic-test-token" not in capsys.readouterr().out
+
+
+@pytest.mark.parametrize("kwargs", [
+    {"headers": {"Link": '<https://api.github.com/installation/repositories?page=2>; rel="next"'}},
+    {"status": 302},
+    {"body": b"invalid JSON synthetic-test-token-never-real"},
+    {"body": b"x" * (1024 * 1024 + 1)},
+    {"error": OSError("sensitive-response synthetic-test-token-never-real")},
+])
+def test_identity_failures_do_not_leak_responses(monkeypatch, tmp_path, capsys, kwargs):
+    values = identity_env(monkeypatch, tmp_path)
+    mock_identity_http(monkeypatch, **kwargs)
+    with pytest.raises(SystemExit, match="1"):
+        run_identity_python("publisher_identity_verify")
+    output = capsys.readouterr()
+    assert output.err == ""
+    assert output.out == "PUBLISHER_IDENTITY_VERIFICATION_FAILED\n"
+    assert "identity_verified=false" in Path(values["GITHUB_OUTPUT"]).read_text()
+
+
+@pytest.mark.parametrize(("key", "value"), [("RESEARCH_PUBLISHER_INSTALLATION_ID", "1"), ("RESEARCH_PUBLISHER_TOKEN", "")])
+def test_identity_rejects_bad_installation_or_missing_token_before_http(monkeypatch, tmp_path, key, value):
+    identity_env(monkeypatch, tmp_path)
+    monkeypatch.setenv(key, value)
+    monkeypatch.setattr(urllib.request, "build_opener", lambda *args: pytest.fail("HTTP must not run"))
+    with pytest.raises(SystemExit, match="1"):
+        run_identity_python("publisher_identity_verify")
+
+
+def test_identity_summary_defaults_to_unpublished_and_unverified(monkeypatch, tmp_path, capsys):
+    values = identity_env(monkeypatch, tmp_path)
+    for key in ["APP_ID_MATCHES", "INSTALLATION_MATCHES", "SINGLE_REPO_SCOPE_MATCHES", "IDENTITY_VERIFIED"]:
+        monkeypatch.delenv(key, raising=False)
+    run_identity_python("publisher_identity_summary")
+    summary = json.loads(capsys.readouterr().out)
+    assert summary == {"app_id_matches": False, "installation_matches": False, "single_repo_scope_matches": False, "identity_verified": False, "published": False}
+    assert "synthetic-test-token" not in Path(values["GITHUB_STEP_SUMMARY"]).read_text()