feldera
diff --git a/‎CLAUDE.md‎
Lines changed: 41 additions & 50 deletions b/‎CLAUDE.md‎
Lines changed: 41 additions & 50 deletions
diff --git a/‎demo/all-packaged/run.py‎
Lines changed: 22 additions & 3 deletions b/‎demo/all-packaged/run.py‎
Lines changed: 22 additions & 3 deletions
diff --git a/‎python/feldera/testutils_oidc.py‎
Lines changed: 186 additions & 19 deletions b/‎python/feldera/testutils_oidc.py‎
Lines changed: 186 additions & 19 deletions
@@ -8279,6 +8279,7 @@ make clean
 - **Pipeline**: Main interface for pipeline operations and data I/O
 - **PipelineBuilder**: Declarative pipeline construction
 - **Output Handlers**: Stream processors for pipeline outputs
+- **OIDC Authentication**: `testutils_oidc.py` - OIDC token management and authentication utilities (see `tests/CLAUDE.md` for details)
 
 ## Important Implementation Details
 
@@ -8375,10 +8376,12 @@ pipeline = client.create_pipeline("my_pipeline")
 - `pandas>=2.1.2` - DataFrame operations and data manipulation
 - `numpy>=2.2.4` - Numerical computing support
 - `typing-extensions` - Enhanced type annotations
+- `PyJWT>=2.8.0` - JWT token handling for OIDC authentication
+- `pretty-errors` - Enhanced error formatting
+- `ruff>=0.6.9` - Fast Python linter and formatter
 
 #### Development Dependencies
 - `pytest>=8.3.5` - Testing framework with timeout support
-- `ruff>=0.6.9` - Fast Python linter and formatter
 - `sphinx==7.3.7` - Documentation generation
 - `kafka-python-ng==2.2.2` - Kafka integration for testing
 
@@ -8448,51 +8451,26 @@ This completely eliminates auth server rate limiting issues and cross-process to
 
 ##### Implementation Details
 
-**Master-Only Token Fetching**:
-```python
-def pytest_configure(config):
-    """Fetch OIDC token on master node only."""
-    if is_master(config):
-        token_data = _fetch_oidc_token()
-        # Store in environment variable for cross-process access
-        import base64
-        token_json = json.dumps(token_data)
-        token_b64 = base64.b64encode(token_json.encode()).decode()
-        os.environ['FELDERA_PYTEST_OIDC_TOKEN'] = token_b64
-```
-
-**Cross-Process Token Access**:
-```python
-def obtain_access_token(self):
-    """Get token from environment variable set by master node."""
-    env_token = os.getenv('FELDERA_PYTEST_OIDC_TOKEN')
-    if env_token:
-        import base64
-        token_json = base64.b64decode(env_token.encode()).decode()
-        token_data = json.loads(token_json)
-        return token_data['access_token']
-```
-
-**Session Fixture Verification**:
-- Session-scoped `oidc_token_fixture` verifies token is available and valid
-- Validates token expiration with 30-second buffer
-- Serves as early validation that authentication is properly set up
-
-**Header Merging for Custom Requests**:
-```python
-def http_request(method: str, path: str, **kwargs):
-    """Merge authentication headers with custom headers."""
-    base_headers = _base_headers()  # Contains Authorization token
-    custom_headers = kwargs.pop("headers", None) or {}
-    headers = {**base_headers, **custom_headers}  # Merge both
-```
+The OIDC token caching system uses pytest-xdist hooks and environment variables for cross-process token sharing:
+
+- **Master-Only Fetching**: `pytest_configure()` hook fetches OIDC token once on master node
+- **Environment Storage**: Token cached in `FELDERA_PYTEST_OIDC_TOKEN` with base64 encoding
+- **Cross-Process Access**: Worker processes inherit environment variable automatically
+- **Session Validation**: `oidc_token_fixture` verifies token setup with 30-second expiration buffer
+- **Header Integration**: `http_request()` merges authentication headers with custom headers
 
 **Key Components**:
 - **`pytest_configure()` (conftest.py)**: Master-only hook that fetches OIDC token once and stores in environment
 - **`pytest_configure_node()` (conftest.py)**: xdist hook that passes token data via workerinput (backup mechanism)
 - **`oidc_token_fixture()` (conftest.py)**: Session fixture that verifies token setup
-- **`obtain_access_token()` (oidc_test_helper.py)**: Returns token from environment variable or fails fast
+- **`obtain_access_token()` (testutils_oidc.py)**: Returns token from environment variable or fails fast
 - **`http_request()` (helper.py)**: Merges authentication headers with custom headers for ingress/egress requests
+- **Reusable Token Functions (testutils_oidc.py)**:
+  - `setup_token_cache()`: High-level function that sets up token caching (used by both pytest and demo runners)
+  - `get_cached_token_from_env()`: Retrieves and validates cached tokens
+  - `parse_cached_token()`: Parses base64-encoded token data
+  - `is_token_valid()`: Checks token expiration with configurable buffer
+  - `encode_token_for_env()`: Encodes tokens for environment storage
 
 ### API Key Authentication (Fallback)
 
@@ -8504,17 +8482,9 @@ export FELDERA_API_KEY="your-api-key"
 
 ### Usage in Tests
 
-Platform tests automatically detect and use the appropriate authentication method:
+Platform tests automatically detect and use the appropriate authentication method via `_base_headers()` helper function.
 
-```python
-from tests.platform.helper import _base_headers
-
-# Headers automatically include OIDC token or API key
-headers = _base_headers()
-response = requests.get(url, headers=headers)
-```
-
-The authentication selection follows this priority:
+Authentication priority:
 1. **OIDC** (if `OIDC_TEST_ISSUER` and related vars are set)
 2. **API Key** (if `FELDERA_API_KEY` is set)
 3. **No Auth** (for local testing without authentication)
@@ -8533,6 +8503,27 @@ env:
 ```
 
 This ensures consistent authentication across both "Runtime Integration Tests" and "Platform Integration Tests (OSS Docker Image)" workflows that run in parallel.
+
+### OIDC Usage Beyond Testing
+
+The OIDC infrastructure is designed to be reusable outside of the test suite:
+
+#### Demo Runners and External Tools
+
+External tools can reuse the OIDC infrastructure by calling `setup_token_cache()` followed by `_get_effective_api_key()` to get cached tokens or fallback API keys.
+
+#### Token Caching for Multiple Processes
+
+The token caching system is designed to work across multiple processes:
+
+1. **First Process**: Calls `setup_token_cache()` → fetches and caches token in environment
+2. **Subsequent Processes**: Call `setup_token_cache()` → reuses cached token if still valid
+3. **Automatic Refresh**: Fetches new token only when cached token expires
+
+This pattern is used by:
+- **Pytest Test Runs**: Master node fetches token, workers reuse it
+- **Demo Runners**: `demo/all-packaged/run.py` uses the same caching mechanism
+- **CI Workflows**: Multiple demos in sequence reuse the same token
 <!-- SECTION:python/tests/CLAUDE.md END -->
 
 ---
 
@@ -16,23 +16,42 @@
 import argparse
 from feldera import FelderaClient, PipelineBuilder
 from feldera.enums import PipelineStatus
+from feldera.testutils import _get_effective_api_key
+from feldera.testutils_oidc import setup_token_cache
 
 
 def main():
+    # Initialize OIDC token cache before any API calls (reuses pytest logic)
+    setup_token_cache()
+    
     # Parse command-line arguments
     parser = argparse.ArgumentParser()
     parser.add_argument(
         "--api-url", required=True, help="Feldera API URL (e.g., http://localhost:8080)"
     )
     args = parser.parse_args()
 
-    # Create Feldera client using the API URL
+    # Create Feldera client using the API URL with OIDC token caching support
     api_url = args.api_url
     print(f"Feldera API URL: {api_url}")
-    client = FelderaClient(api_url)
+    
+    # Get effective API key (OIDC token takes precedence over static API key)
+    # This reuses the same token caching infrastructure as the Python test suite
+    effective_api_key = _get_effective_api_key()
+    if effective_api_key:
+        print("🔐 AUTH: Using cached OIDC authentication")
+        client = FelderaClient(api_url, api_key=effective_api_key)
+    else:
+        print("🔐 AUTH: Using no authentication")
+        client = FelderaClient(api_url)
 
     # Retrieve and run all packaged demos
-    demos = requests.get(f"{api_url}/v0/config/demos").json()
+    # Use the same authentication headers for the demos request
+    headers = {"Accept": "application/json"}
+    if effective_api_key:
+        headers["Authorization"] = f"Bearer {effective_api_key}"
+    
+    demos = requests.get(f"{api_url}/v0/config/demos", headers=headers).json()
     print(f"Total {len(demos)} packages demos were found and will be run")
     for demo in demos:
         print(f"Running packaged demo: {demo['name']}...")
 
@@ -101,25 +101,13 @@ def obtain_access_token(self, pytest_cache=None) -> str:
         current_time = time.time()
 
         # Check environment variable for cross-process token sharing
-        env_token = os.getenv("FELDERA_PYTEST_OIDC_TOKEN")
-        if env_token:
-            try:
-                import base64
-
-                token_json = base64.b64decode(env_token.encode()).decode()
-                token_data = json.loads(token_json)
-
-                if (
-                    token_data.get("access_token")
-                    and current_time < token_data.get("expires_at", 0) - 30
-                ):
-                    logger.info("Using environment variable cached access token")
-                    # Cache in instance for future calls to avoid repeated parsing
-                    self._access_token = token_data["access_token"]
-                    self._token_expires_at = token_data["expires_at"]
-                    return token_data["access_token"]
-            except Exception as e:
-                logger.warning(f"Failed to parse environment token: {e}")
+        token_data = get_cached_token_from_env()
+        if token_data:
+            logger.info("Using environment variable cached access token")
+            # Cache in instance for future calls to avoid repeated parsing
+            self._access_token = token_data["access_token"]
+            self._token_expires_at = token_data["expires_at"]
+            return token_data["access_token"]
 
         # Fallback: Check instance cache
         if self._access_token and current_time < self._token_expires_at - 30:
@@ -195,3 +183,182 @@ def get_oidc_test_helper() -> Optional[OidcTestHelper]:
         if config:
             _test_helper = OidcTestHelper(config)
     return _test_helper
+
+
+def parse_cached_token(env_token: str) -> Optional[Dict[str, Any]]:
+    """
+    Parse and validate a base64-encoded token from environment variable.
+
+    Args:
+        env_token: Base64-encoded JSON token data from environment variable
+
+    Returns:
+        Parsed token data dict if valid, None if invalid or expired
+    """
+    try:
+        import base64
+        token_json = base64.b64decode(env_token.encode()).decode()
+        token_data = json.loads(token_json)
+        return token_data
+    except Exception as e:
+        logging.getLogger(__name__).warning(f"Failed to parse cached token: {e}")
+        return None
+
+
+def is_token_valid(token_data: Dict[str, Any], buffer_seconds: int = 30) -> bool:
+    """
+    Check if token data is valid and not expired.
+
+    Args:
+        token_data: Dictionary containing token information
+        buffer_seconds: Safety buffer before expiration (default 30 seconds)
+
+    Returns:
+        True if token is valid and not expired, False otherwise
+    """
+    if not token_data:
+        return False
+
+    access_token = token_data.get("access_token")
+    expires_at = token_data.get("expires_at", 0)
+
+    if not access_token:
+        return False
+
+    current_time = time.time()
+    return current_time < expires_at - buffer_seconds
+
+
+def encode_token_for_env(token_data: Dict[str, Any]) -> str:
+    """
+    Encode token data as base64 for storage in environment variables.
+
+    Args:
+        token_data: Dictionary containing token information
+
+    Returns:
+        Base64-encoded JSON string suitable for environment variable storage
+    """
+    import base64
+    token_json = json.dumps(token_data)
+    return base64.b64encode(token_json.encode()).decode()
+
+
+def get_cached_token_from_env(env_var_name: str = "FELDERA_PYTEST_OIDC_TOKEN") -> Optional[Dict[str, Any]]:
+    """
+    Retrieve and validate cached token from environment variable.
+
+    Args:
+        env_var_name: Name of environment variable containing cached token
+
+    Returns:
+        Valid token data if available and not expired, None otherwise
+    """
+    import os
+
+    env_token = os.getenv(env_var_name)
+    if not env_token:
+        return None
+
+    token_data = parse_cached_token(env_token)
+    if token_data and is_token_valid(token_data):
+        return token_data
+
+    return None
+
+
+def setup_token_cache() -> Optional[Dict[str, Any]]:
+    """
+    Set up OIDC token cache in environment variable if not already present.
+
+    This function:
+    1. Checks if a valid token is already cached
+    2. If not, fetches a new token
+    3. Stores the token in environment variable for cross-process access
+
+    Used by both pytest hooks and demo runners to ensure consistent token caching.
+
+    Returns:
+        Token data if successfully cached, None if OIDC not configured
+    """
+    import os
+
+    # Check if token is already cached and still valid
+    cached_token = get_cached_token_from_env()
+    if cached_token:
+        print("🔐 AUTH: Using existing cached OIDC token")
+        return cached_token
+
+    # Fetch new token if needed
+    token_data = fetch_oidc_token()
+    if token_data:
+        # Store in environment variable for reuse by subsequent processes
+        token_b64 = encode_token_for_env(token_data)
+        os.environ["FELDERA_PYTEST_OIDC_TOKEN"] = token_b64
+        print("🔐 AUTH: Token cached in environment for cross-process access")
+        return token_data
+    else:
+        print("🔐 AUTH: No OIDC configuration found, using fallback authentication")
+        return None
+
+
+def fetch_oidc_token() -> Optional[Dict[str, Any]]:
+    """
+    Fetch OIDC token using Resource Owner Password Grant flow.
+
+    This function is used by both pytest hooks and demo runners to ensure
+    consistent token fetching behavior across the entire test infrastructure.
+
+    Returns:
+        Dict containing access_token, expires_at, and cached_at if successful,
+        None if OIDC is not configured.
+    """
+    oidc_helper = get_oidc_test_helper()
+    if oidc_helper is None:
+        return None
+
+    print("🔐 AUTH: Fetching OIDC token")
+
+    try:
+        token_endpoint = oidc_helper.get_token_endpoint()
+        data = {
+            "grant_type": "password",
+            "username": oidc_helper.config.username,
+            "password": oidc_helper.config.password,
+            "client_id": oidc_helper.config.client_id,
+            "client_secret": oidc_helper.config.client_secret,
+            "scope": oidc_helper.config.scope,
+            "audience": "feldera-api",
+        }
+
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+        }
+
+        response = requests.post(token_endpoint, data=data, headers=headers, timeout=30)
+
+        if not response.ok:
+            print(f"🔐 AUTH: ❌ Token request FAILED: {response.status_code}")
+            raise Exception(
+                f"Token request failed: {response.status_code} - {response.text}"
+            )
+
+        token_response = response.json()
+        print("🔐 AUTH: ✅ Token request SUCCESS!")
+
+        access_token = token_response["access_token"]
+        expires_in = token_response.get("expires_in", 3600)
+        expires_at = time.time() + expires_in
+
+        return {
+            "access_token": access_token,
+            "expires_at": expires_at,
+            "cached_at": time.time(),
+        }
+
+    except Exception as e:
+        print(f"🔐 AUTH: CRITICAL FAILURE - Failed to fetch OIDC token: {e}")
+        raise RuntimeError(
+            f"OIDC authentication is configured but token retrieval failed: {e}"
+        ) from e