diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 647686de674..ada56567528 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,6 +87,11 @@ jobs: name: Integration Tests needs: [invoke-build-java] uses: ./.github/workflows/test-integration-runtime.yml + # A called workflow gets no more than the calling job grants, and the + # runtime tests authenticate to the CI instances with an OIDC token. + permissions: + contents: read + id-token: write secrets: inherit invoke-tests-java: diff --git a/.github/workflows/test-integration-runtime.yml b/.github/workflows/test-integration-runtime.yml index 351f3abaf40..a90e68fa9cb 100644 --- a/.github/workflows/test-integration-runtime.yml +++ b/.github/workflows/test-integration-runtime.yml @@ -9,6 +9,12 @@ on: description: "ID of the workflow run that uploaded the artifact" required: true +# Reaching the CI instances needs an OIDC token. A caller must grant the same +# permission, which ci.yml does on the job that invokes this workflow. +permissions: + contents: read + id-token: write + jobs: runtime-tests: if: ${{ !contains(vars.CI_SKIP_JOBS, 'runtime-tests') }} @@ -18,14 +24,10 @@ jobs: include: - runner: [gke-runners-amd64] feldera_host: ${{ vars.FELDERA_HOST_CI_GKE_AMD64 }} - feldera_api_key: FELDERA_API_KEY_CI_GKE_AMD64 - runner: [gke-runners-arm64] feldera_host: ${{ vars.FELDERA_HOST_CI_GKE_ARM64 }} - feldera_api_key: FELDERA_API_KEY_CI_GKE_ARM64 runs-on: ${{ matrix.runner }} env: - FELDERA_HOST: ${{ matrix.feldera_host }} - FELDERA_API_KEY: ${{ secrets[matrix.feldera_api_key] }} FELDERA_RUNTIME_VERSION: ${{ github.sha }} # GCS test bucket over the S3-interop endpoint; the env names are the # standard ones the tests read (python/tests/utils.py). @@ -45,13 +47,20 @@ jobs: - name: Checkout repository uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 + # The instances identify this workflow by the audience it requests, so + # the value has to match the one their trust pins. + - name: Authenticate to Feldera + uses: feldera/oidc-auth-action@d1fce411d12fcda95a1f1a0bc441d2dc191e3f2c # v1.0.0 + with: + host: ${{ matrix.feldera_host }} + audience: feldera-ci-runtime-tests + - name: Python runtime tests if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-pytest') }} run: uv run --locked pytest -n ${{ vars.PYTEST_WORKERS }} tests/runtime --timeout=3600 --maxfail=1 -vv ${{ vars.PYTEST_EXTRA }} working-directory: python env: PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true - name: Python runtime_aggtest test if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-aggtest') }} @@ -60,7 +69,6 @@ jobs: env: RUNTIME_AGGTEST_JOBS: ${{ vars.RUNTIME_AGGTEST_JOBS }} PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true - name: Python workload tests if: ${{ vars.CI_DRY_RUN != 'true' && !contains(vars.CI_SKIP_JOBS, 'runtime-workload') }} @@ -68,7 +76,6 @@ jobs: working-directory: python env: PYTHONPATH: ${{ github.workspace }}/python - FELDERA_TLS_INSECURE: true KAFKA_BOOTSTRAP_SERVERS: ${{ vars.CI_KAFKA_BOOTSTRAP }} SCHEMA_REGISTRY_URL: ${{ vars.CI_SCHEMA_REGISTRY }} RUN_ID: 1 # we use this to run a single variant of the Kafka tests in test_kafka_avro.py diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index d3091a46ecf..010ffb373d1 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -1,8 +1,11 @@ "Utility functions for writing tests against a Feldera instance." +import base64 import logging import math import os +import urllib.parse +import urllib.request import platform import re import time @@ -44,6 +47,73 @@ def _get_effective_api_key(): return oidc_token if oidc_token else API_KEY +# Audience -> (token, seconds since the epoch after which it is re-minted). +_oidc_token_cache: dict[str, tuple[str, float]] = {} + +# Re-mint a token this many seconds before its own expiry, so a request issued +# just before the check cannot arrive after it. +_OIDC_REFRESH_MARGIN_SECONDS = 120.0 + + +def _github_oidc_token() -> str: + """A GitHub Actions ID token, re-minted shortly before it expires. + + The SDK resolves this before every request, so the implementation has to be + cheap. Minting per request adds a round trip to GitHub each time, and a + suite that polls in loops across parallel workers sends enough of them to be + throttled, which results in a connection timeout rather than an error. + """ + audience = os.environ.get("FELDERA_OIDC_AUDIENCE", "") + cached = _oidc_token_cache.get(audience) + if cached is not None and time.time() < cached[1]: + return cached[0] + + request_url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"] + if audience: + request_url += "&audience=" + urllib.parse.quote(audience, safe="") + request = urllib.request.Request(request_url) + request.add_header( + "Authorization", f"bearer {os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN']}" + ) + with urllib.request.urlopen(request, timeout=30) as response: + token = json.load(response)["value"] + + _oidc_token_cache[audience] = ( + token, + _token_expiry(token) - _OIDC_REFRESH_MARGIN_SECONDS, + ) + return token + + +def _token_expiry(token: str) -> float: + """`exp` from a JWT payload, in seconds since the epoch. + + Returns the current time if the payload cannot be read, which re-mints on + every call rather than serving a token past its expiry. + """ + try: + payload = token.split(".")[1] + payload += "=" * (-len(payload) % 4) + return float(json.loads(base64.urlsafe_b64decode(payload))["exp"]) + except Exception: + return time.time() + + +def _feldera_credential(): + """Bearer credential for the test client. + + Under GitHub Actions the credential is an ID token that expires well inside + a long test run, so hand the SDK the callable rather than a token: it + re-resolves per request and retries once on 401. A configured OIDC login + flow takes precedence over both. + """ + if os.environ.get("OIDC_TEST_ISSUER"): + return _get_effective_api_key() + if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"): + return _github_oidc_token + return API_KEY + + BASE_URL = os.environ.get("FELDERA_HOST") or "http://localhost:8080" FELDERA_REQUESTS_VERIFY = requests_verify_from_env() FELDERA_TEST_NUM_WORKERS = int(os.environ.get("FELDERA_TEST_NUM_WORKERS", "8")) @@ -87,7 +157,7 @@ def _ensure(self): if self._client is None: self._client = FelderaClient( connection_timeout=10, - api_key=_get_effective_api_key(), + api_key=_feldera_credential(), ) return self._client