fix(claims): authenticate catalog ownership lookups

.github/workflows/claim-notice.yml

@@ -66,8 +66,11 @@ jobs:
           PR_TITLE: ${{ github.event.inputs.pr_title || github.event.pull_request.title }}
           PR_AUTHOR: ${{ github.event.inputs.pr_author || github.event.pull_request.user.login }}
           VERIFY_MERGED_PR: ${{ github.event_name == 'workflow_dispatch' }}
-          REGISTRY_API: https://hol.org/registry/api/v1
-        run: python3 scripts/post-claim-notice.py
+          REGISTRY_API: https://hol.org/api/v1
+          REGISTRY_BROKER_API_KEY: ${{ secrets.REGISTRY_BROKER_API_KEY }}
+        run: |
+          test -n "$REGISTRY_BROKER_API_KEY" || { echo "Missing catalog API key secret"; exit 1; }
+          python3 scripts/post-claim-notice.py
 
       - name: Clear delivered notice from retry queue
         if: success() && github.event_name != 'schedule'
@@ -80,5 +83,8 @@ jobs:
         if: github.event_name == 'schedule'
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          REGISTRY_API: https://hol.org/registry/api/v1
-        run: python3 scripts/backfill-claim-notices.py
+          REGISTRY_API: https://hol.org/api/v1
+          REGISTRY_BROKER_API_KEY: ${{ secrets.REGISTRY_BROKER_API_KEY }}
+        run: |
+          test -n "$REGISTRY_BROKER_API_KEY" || { echo "Missing catalog API key secret"; exit 1; }
+          python3 scripts/backfill-claim-notices.py

.github/workflows/validate-contribution.yml

@@ -16,6 +16,7 @@ on:
       - "scripts/post-claim-notice.py"
       - "scripts/backfill-claim-notices.py"
       - "tests/test_post_claim_notice.py"
+      - "tests/test_claim_notice_api_key.py"
       - "tests/test_backfill_claim_notices.py"
       - ".github/workflows/claim-notice.yml"
       - ".github/workflows/sweep-open-prs.yml"
@@ -50,7 +51,7 @@ jobs:
           python3 -m unittest scripts/test_validate_contribution_branch.py
           python3 -m unittest scripts/test_publish_open_pr_checks.py
           python3 -m unittest scripts/test_check_alphabetical.py
-          python3 -m unittest tests/test_post_claim_notice.py tests/test_backfill_claim_notices.py
+          python3 -m unittest tests/test_post_claim_notice.py tests/test_backfill_claim_notices.py tests/test_claim_notice_api_key.py
           python3 scripts/check-alphabetical.py README.md --base-ref "$BASE_REF"
           python3 scripts/validate-contribution-centralized.py \
             --base-ref "$BASE_REF" \

scripts/post-claim-notice.py

@@ -18,10 +18,11 @@
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
 from pathlib import Path
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlsplit
 
 # --- Config ---
-REGISTRY_API = os.environ.get("REGISTRY_API", "https://hol.org/registry/api/v1")
+REGISTRY_API = os.environ.get("REGISTRY_API", "https://hol.org/api/v1")
+REGISTRY_BROKER_API_KEY = os.environ.get("REGISTRY_BROKER_API_KEY", "")
 GH_TOKEN = os.environ.get("GH_TOKEN", "")
 PR_NUMBER = os.environ.get("PR_NUMBER", "")
 PR_TITLE = os.environ.get("PR_TITLE", "")
@@ -149,6 +150,13 @@ def api_request(url, headers=None, method="GET", data=None, retry_registry=False
         data = json.dumps(data).encode("utf-8")
         req_headers["Content-Type"] = "application/json"
     req = urllib.request.Request(url, headers=req_headers, method=method, data=data)
+    if retry_registry and REGISTRY_BROKER_API_KEY:
+        target = urlsplit(url)
+        if (method != "GET" or target.scheme != "https" or target.netloc != "hol.org"
+                or target.path != "/api/v1/plugins/catalog"):
+            raise RegistryCatalogFetchError("Refusing to send the catalog key outside its read endpoint")
+        # urllib does not copy unredirected headers into redirected requests.
+        req.add_unredirected_header("x-api-key", REGISTRY_BROKER_API_KEY)
     attempts = REGISTRY_MAX_ATTEMPTS if retry_registry and method == "GET" else 1
     for attempt in range(attempts):
         retry_after = None

tests/test_claim_notice_api_key.py

@@ -0,0 +1,45 @@
+import importlib.util
+import io
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+from urllib.request import HTTPRedirectHandler
+
+SPEC = importlib.util.spec_from_file_location('claim_api_key', Path(__file__).resolve().parents[1] / 'scripts/post-claim-notice.py')
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+
+
+class CatalogKeyTests(unittest.TestCase):
+    def test_key_is_sent_only_to_the_catalog_read_and_not_forwarded_on_redirect(self):
+        with patch.object(MODULE, 'REGISTRY_BROKER_API_KEY', 'fixture-key'), patch.object(
+            MODULE.urllib.request, 'urlopen', return_value=io.BytesIO(b'{"items":[]}')
+        ) as fetch:
+            MODULE.api_request('https://hol.org/api/v1/plugins/catalog?q=owner%2Frepo', retry_registry=True)
+            request = fetch.call_args.args[0]
+            self.assertEqual(request.get_header('X-api-key'), 'fixture-key')
+            redirected = HTTPRedirectHandler().redirect_request(request, None, 302, 'found', {}, 'https://other.example/catalog')
+            self.assertIsNone(redirected.get_header('X-api-key'))
+
+    def test_github_calls_never_receive_the_registry_key(self):
+        with patch.object(MODULE, 'REGISTRY_BROKER_API_KEY', 'fixture-key'), patch.object(
+            MODULE.urllib.request, 'urlopen', return_value=io.BytesIO(b'{}')
+        ) as fetch:
+            MODULE.api_request('https://api.github.com/repos/owner/catalog')
+            self.assertIsNone(fetch.call_args.args[0].get_header('X-api-key'))
+
+    def test_bad_origin_path_or_method_cannot_receive_the_key(self):
+        with patch.object(MODULE, 'REGISTRY_BROKER_API_KEY', 'fixture-key'), patch.object(MODULE.urllib.request, 'urlopen') as fetch:
+            for url, method in [
+                ('https://other.example/api/v1/plugins/catalog', 'GET'),
+                ('http://hol.org/api/v1/plugins/catalog', 'GET'),
+                ('https://hol.org/api/v1/plugins', 'GET'),
+                ('https://hol.org/api/v1/plugins/catalog', 'POST'),
+            ]:
+                with self.subTest(url=url, method=method), self.assertRaises(MODULE.RegistryCatalogFetchError):
+                    MODULE.api_request(url, method=method, retry_registry=True)
+            fetch.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()