diff --git a/.gitignore b/.gitignore
index bcb6b89f6ff..9eff6289235 100644
--- a/.gitignore
+++ b/.gitignore
@@ -30,4 +30,5 @@ env/
 .idea
 .env*
 **/venv
-**/noxfile.py
\ No newline at end of file
+**/noxfile.py\n# Local secrets file\nauth/custom-credentials/okta/custom-credentials-okta-secrets.json
+auth/custom-credentials/aws/custom-credentials-aws-secrets.json
diff --git a/auth/custom-credentials/aws/Dockerfile b/auth/custom-credentials/aws/Dockerfile
new file mode 100644
index 00000000000..1cf23d3ac50
--- /dev/null
+++ b/auth/custom-credentials/aws/Dockerfile
@@ -0,0 +1,20 @@
+FROM python:3.9-slim
+
+# Create a non-root user
+RUN useradd -m appuser
+
+# Create a working directory
+WORKDIR /app
+
+# Copy files and install dependencies
+COPY --chown=appuser:appuser requirements.txt .
+COPY --chown=appuser:appuser snippets.py .
+
+# Switch to the non-root user
+USER appuser
+
+# Install dependencies for the user
+RUN pip install --no-cache-dir --user -r requirements.txt
+
+# Set the entrypoint
+CMD ["python3", "snippets.py"]
diff --git a/auth/custom-credentials/aws/README.md b/auth/custom-credentials/aws/README.md
new file mode 100644
index 00000000000..767906682a2
--- /dev/null
+++ b/auth/custom-credentials/aws/README.md
@@ -0,0 +1,108 @@
+# Running the Custom Credential Supplier Sample
+
+If you want to use AWS security credentials that cannot be retrieved using methods supported natively by the [google-auth](https://github.com/googleapis/google-auth-library-python) library, a custom `AwsSecurityCredentialsSupplier` implementation may be specified. The supplier must return valid, unexpired AWS security credentials when called by the Google Cloud Auth library.
+
+This sample demonstrates how to use **Boto3** (the AWS SDK for Python) as a custom supplier to bridge AWS credentials—from sources like EKS IRSA, ECS, or Fargate—to Google Cloud Workload Identity.
+
+## Running Locally
+
+For local development, you can provide credentials and configuration in a JSON file. For containerized environments like EKS, the script can fall back to environment variables.
+
+### 1. Install Dependencies
+
+Ensure you have Python installed, then install the required libraries:
+
+```bash
+pip install -r requirements.txt
+```
+
+### 2. Configure Credentials for Local Development
+
+1.  Copy the example secrets file to a new file named `custom-credentials-aws-secrets.json`:
+    ```bash
+    cp custom-credentials-aws-secrets.json.example custom-credentials-aws-secrets.json
+    ```
+2.  Open `custom-credentials-aws-secrets.json` and fill in the required values for your AWS and GCP configuration. The `custom-credentials-aws-secrets.json` file is ignored by Git, so your credentials will not be checked into version control.
+
+### 3. Run the Script
+
+```bash
+python3 snippets.py
+```
+
+When run locally, the script will detect the `custom-credentials-aws-secrets.json` file and use it to configure the necessary environment variables for the Boto3 client.
+
+## Running in a Containerized Environment (EKS)
+
+This section provides a brief overview of how to run the sample in an Amazon EKS cluster.
+
+### 1. EKS Cluster Setup
+
+First, you need an EKS cluster. You can create one using `eksctl` or the AWS Management Console. For detailed instructions, refer to the [Amazon EKS documentation](https://docs.aws.amazon.com/eks/latest/userguide/create-cluster.html).
+
+### 2. Configure IAM Roles for Service Accounts (IRSA)
+
+IRSA allows you to associate an IAM role with a Kubernetes service account. This provides a secure way for your pods to access AWS services without hardcoding long-lived credentials.
+
+You can essentially complete the OIDC setup, IAM role creation, and Service Account association in one step using `eksctl`.
+
+Run the following command to create the IAM role and bind it to a Kubernetes Service Account:
+
+```bash
+eksctl create iamserviceaccount \
+  --name your-k8s-service-account \
+  --namespace default \
+  --cluster your-cluster-name \
+  --region your-aws-region \
+  --role-name your-role-name \
+  --attach-policy-arn arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess \
+  --approve
+```
+
+> **Note**: The `--attach-policy-arn` flag is used here to demonstrate attaching permissions. Update this with the specific AWS policy ARN your application requires (e.g., if your Boto3 client needs to read from S3 or DynamoDB).
+
+For a deep dive into how this works manually, refer to the [IAM Roles for Service Accounts](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html) documentation.
+
+### 3. Configure GCP to Trust the AWS Role
+
+To allow your AWS role to authenticate as a Google Cloud service account, you need to configure Workload Identity Federation. This process involves these key steps:
+
+1.  **Create a Workload Identity Pool and an AWS Provider:** The pool holds the configuration, and the provider is set up to trust your AWS account.
+
+2.  **Create or select a GCP Service Account:** This service account will be impersonated by your AWS role. Grant this service account the necessary GCP permissions for your application (e.g., access to GCS or BigQuery).
+
+3.  **Bind the AWS Role to the GCP Service Account:** Create an IAM policy binding that gives your AWS role the `Workload Identity User` (`roles/iam.workloadIdentityUser`) role on the GCP service account. This allows the AWS role to impersonate the service account.
+
+**Alternative: Direct Access**
+
+> For supported resources, you can grant roles directly to the AWS identity, bypassing service account impersonation. To do this, grant a role (like `roles/storage.objectViewer`) to the workload identity principal (`principalSet://...`) directly on the resource's IAM policy.
+
+For more detailed information, see the documentation on [Configuring Workload Identity Federation](https://cloud.google.com/iam/docs/workload-identity-federation-with-other-clouds).
+
+### 4. Containerize and Package the Application
+
+Create a `Dockerfile` for the Python application and push the image to a container registry (e.g., Amazon ECR) that your EKS cluster can access. Refer to the [`Dockerfile`](Dockerfile) for the container image definition.
+
+Build and push the image:
+```bash
+docker build -t your-container-image:latest .
+docker push your-container-image:latest
+```
+
+### 5. Deploy to EKS
+
+Create a Kubernetes deployment manifest to deploy your application to the EKS cluster. See the [`pod.yaml`](pod.yaml) file for an example.
+
+Deploy the pod:
+
+```bash
+kubectl apply -f pod.yaml
+```
+
+### 6. Clean Up
+
+To clean up the resources, delete the EKS cluster and any other AWS and GCP resources you created.
+
+```bash
+eksctl delete cluster --name your-cluster-name
+```
\ No newline at end of file
diff --git a/auth/custom-credentials/aws/custom-credentials-aws-secrets.json.example b/auth/custom-credentials/aws/custom-credentials-aws-secrets.json.example
new file mode 100644
index 00000000000..300dc70c138
--- /dev/null
+++ b/auth/custom-credentials/aws/custom-credentials-aws-secrets.json.example
@@ -0,0 +1,8 @@
+{
+  "aws_access_key_id": "YOUR_AWS_ACCESS_KEY_ID",
+  "aws_secret_access_key": "YOUR_AWS_SECRET_ACCESS_KEY",
+  "aws_region": "YOUR_AWS_REGION",
+  "gcp_workload_audience": "YOUR_GCP_WORKLOAD_AUDIENCE",
+  "gcs_bucket_name": "YOUR_GCS_BUCKET_NAME",
+  "gcp_service_account_impersonation_url": "YOUR_GCP_SERVICE_ACCOUNT_IMPERSONATION_URL"
+}
diff --git a/auth/custom-credentials/aws/noxfile_config.py b/auth/custom-credentials/aws/noxfile_config.py
new file mode 100644
index 00000000000..834faa2b2f6
--- /dev/null
+++ b/auth/custom-credentials/aws/noxfile_config.py
@@ -0,0 +1,18 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+TEST_CONFIG_OVERRIDE = {
+    # Ignore all versions except 3.9, which is the version available.
+    "ignored_versions": ["2.7", "3.6", "3.7", "3.8", "3.10", "3.11", "3.12", "3.13"],
+}
diff --git a/auth/custom-credentials/aws/pod.yaml b/auth/custom-credentials/aws/pod.yaml
new file mode 100644
index 00000000000..4ab6d7997e3
--- /dev/null
+++ b/auth/custom-credentials/aws/pod.yaml
@@ -0,0 +1,33 @@
+# Copyright 2025 Google LLC
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: Pod
+metadata:
+  name: custom-credential-pod
+spec:
+  serviceAccountName: your-k8s-service-account # The service account associated with the AWS IAM role
+  containers:
+  - name: gcp-auth-sample
+    image: your-container-image:latest # Your image from ECR
+    env:
+    # AWS_REGION is often required for Boto3 to initialize correctly in containers
+    - name: AWS_REGION
+      value: "your-aws-region"
+    - name: GCP_WORKLOAD_AUDIENCE
+      value: "your-gcp-workload-audience"
+    # Optional: If you want to use service account impersonation
+    # - name: GCP_SERVICE_ACCOUNT_IMPERSONATION_URL
+    #   value: "your-gcp-service-account-impersonation-url"
+    - name: GCS_BUCKET_NAME
+      value: "your-gcs-bucket-name"
diff --git a/auth/custom-credentials/aws/requirements-test.txt b/auth/custom-credentials/aws/requirements-test.txt
new file mode 100644
index 00000000000..43b24059d3e
--- /dev/null
+++ b/auth/custom-credentials/aws/requirements-test.txt
@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest==8.2.0
diff --git a/auth/custom-credentials/aws/requirements.txt b/auth/custom-credentials/aws/requirements.txt
new file mode 100644
index 00000000000..4a091fb39fb
--- /dev/null
+++ b/auth/custom-credentials/aws/requirements.txt
@@ -0,0 +1,4 @@
+boto3==1.40.53
+google-auth==2.43.0
+python-dotenv==1.1.1
+requests==2.32.3
diff --git a/auth/custom-credentials/aws/snippets.py b/auth/custom-credentials/aws/snippets.py
new file mode 100644
index 00000000000..67625e1246d
--- /dev/null
+++ b/auth/custom-credentials/aws/snippets.py
@@ -0,0 +1,154 @@
+# Copyright 2025 Google LLC
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# [START auth_custom_credential_supplier_aws]
+import json
+import os
+
+import boto3
+from google.auth import aws
+from google.auth import exceptions
+from google.auth.transport import requests as auth_requests
+
+
+class CustomAwsSupplier(aws.AwsSecurityCredentialsSupplier):
+    """Custom AWS Security Credentials Supplier using Boto3."""
+
+    def __init__(self):
+        """Initializes the Boto3 session, prioritizing environment variables for region."""
+        # Explicitly read the region from the environment first.
+        region = os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION")
+
+        # If region is None, Boto3's discovery chain will be used when needed.
+        self.session = boto3.Session(region_name=region)
+        self._cached_region = None
+
+    def get_aws_region(self, context, request) -> str:
+        """Returns the AWS region using Boto3's default provider chain."""
+        if self._cached_region:
+            return self._cached_region
+
+        self._cached_region = self.session.region_name
+
+        if not self._cached_region:
+            raise exceptions.GoogleAuthError(
+                "Boto3 was unable to resolve an AWS region."
+            )
+
+        return self._cached_region
+
+    def get_aws_security_credentials(
+        self, context, request=None
+    ) -> aws.AwsSecurityCredentials:
+        """Retrieves AWS security credentials using Boto3's default provider chain."""
+        creds = self.session.get_credentials()
+        if not creds:
+            raise exceptions.GoogleAuthError(
+                "Unable to resolve AWS credentials from Boto3."
+            )
+
+        return aws.AwsSecurityCredentials(
+            access_key_id=creds.access_key,
+            secret_access_key=creds.secret_key,
+            session_token=creds.token,
+        )
+
+
+def authenticate_with_aws_credentials(bucket_name, audience, impersonation_url=None):
+    """Authenticates using the custom AWS supplier and gets bucket metadata.
+
+    Returns:
+        dict: The bucket metadata response from the Google Cloud Storage API.
+    """
+
+    # 1. Instantiate the custom supplier.
+    custom_supplier = CustomAwsSupplier()
+
+    # 2. Instantiate the AWS Credentials object.
+    credentials = aws.Credentials(
+        audience=audience,
+        subject_token_type="urn:ietf:params:aws:token-type:aws4_request",
+        service_account_impersonation_url=impersonation_url,
+        aws_security_credentials_supplier=custom_supplier,
+        scopes=["https://www.googleapis.com/auth/devstorage.read_write"],
+    )
+
+    # 3. Create an authenticated session.
+    authed_session = auth_requests.AuthorizedSession(credentials)
+
+    # 4. Make the API Request.
+    bucket_url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}"
+
+    response = authed_session.get(bucket_url)
+    response.raise_for_status()
+
+    return response.json()
+
+
+# [END auth_custom_credential_supplier_aws]
+
+
+def _load_config_from_file():
+    """
+    If a local secrets file is present, load it into the environment.
+    This is a "just-in-time" configuration for local development. These
+    variables are only set for the current process and are not exposed to the
+    shell.
+    """
+    if os.path.exists("custom-credentials-aws-secrets.json"):
+        with open("custom-credentials-aws-secrets.json", "r") as f:
+            secrets = json.load(f)
+
+        os.environ["AWS_ACCESS_KEY_ID"] = secrets.get("aws_access_key_id", "")
+        os.environ["AWS_SECRET_ACCESS_KEY"] = secrets.get("aws_secret_access_key", "")
+        os.environ["AWS_REGION"] = secrets.get("aws_region", "")
+        os.environ["GCP_WORKLOAD_AUDIENCE"] = secrets.get("gcp_workload_audience", "")
+        os.environ["GCS_BUCKET_NAME"] = secrets.get("gcs_bucket_name", "")
+        os.environ["GCP_SERVICE_ACCOUNT_IMPERSONATION_URL"] = secrets.get(
+            "gcp_service_account_impersonation_url", ""
+        )
+
+
+def main():
+
+    # Reads the custom-credentials-aws-secrets.json if running locally.
+    _load_config_from_file()
+
+    # Now, read the configuration from the environment. In a local run, these
+    # will be the values we just set. In a containerized run, they will be
+    # the values provided by the environment.
+    gcp_audience = os.getenv("GCP_WORKLOAD_AUDIENCE")
+    sa_impersonation_url = os.getenv("GCP_SERVICE_ACCOUNT_IMPERSONATION_URL")
+    gcs_bucket_name = os.getenv("GCS_BUCKET_NAME")
+
+    if not all([gcp_audience, gcs_bucket_name]):
+        print(
+            "Required configuration missing. Please provide it in a "
+            "custom-credentials-aws-secrets.json file or as environment variables: "
+            "GCP_WORKLOAD_AUDIENCE, GCS_BUCKET_NAME"
+        )
+        return
+
+    try:
+        print(f"Retrieving metadata for bucket: {gcs_bucket_name}...")
+        metadata = authenticate_with_aws_credentials(
+            gcs_bucket_name, gcp_audience, sa_impersonation_url
+        )
+        print("--- SUCCESS! ---")
+        print(json.dumps(metadata, indent=2))
+    except Exception as e:
+        print(f"Authentication or Request failed: {e}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/auth/custom-credentials/aws/snippets_test.py b/auth/custom-credentials/aws/snippets_test.py
new file mode 100644
index 00000000000..779a492c08f
--- /dev/null
+++ b/auth/custom-credentials/aws/snippets_test.py
@@ -0,0 +1,131 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import os
+from unittest import mock
+
+import pytest
+
+import snippets
+
+# --- Unit Tests ---
+
+
+@mock.patch.dict(os.environ, {"AWS_REGION": "us-west-2"})
+@mock.patch("boto3.Session")
+def test_init_priority_env_var(mock_boto_session):
+    """Test that AWS_REGION env var takes priority during init."""
+    snippets.CustomAwsSupplier()
+    mock_boto_session.assert_called_with(region_name="us-west-2")
+
+
+@mock.patch.dict(os.environ, {}, clear=True)
+@mock.patch("boto3.Session")
+def test_get_aws_region_caching(mock_boto_session):
+    """Test that get_aws_region caches the result from Boto3."""
+    mock_session_instance = mock_boto_session.return_value
+    mock_session_instance.region_name = "us-east-1"
+
+    supplier = snippets.CustomAwsSupplier()
+
+    # First call should hit the session
+    region = supplier.get_aws_region(None, None)
+    assert region == "us-east-1"
+
+    # Change the mock to ensure we aren't calling it again
+    mock_session_instance.region_name = "us-west-2"
+
+    # Second call should return the cached value
+    region2 = supplier.get_aws_region(None, None)
+    assert region2 == "us-east-1"
+
+
+@mock.patch("boto3.Session")
+def test_get_aws_security_credentials_success(mock_boto_session):
+    """Test successful retrieval of AWS credentials."""
+    mock_session_instance = mock_boto_session.return_value
+
+    mock_creds = mock.MagicMock()
+    mock_creds.access_key = "test-key"
+    mock_creds.secret_key = "test-secret"
+    mock_creds.token = "test-token"
+    mock_session_instance.get_credentials.return_value = mock_creds
+
+    supplier = snippets.CustomAwsSupplier()
+    creds = supplier.get_aws_security_credentials(None)
+
+    assert creds.access_key_id == "test-key"
+    assert creds.secret_access_key == "test-secret"
+    assert creds.session_token == "test-token"
+
+
+@mock.patch("snippets.auth_requests.AuthorizedSession")
+@mock.patch("snippets.aws.Credentials")
+@mock.patch("snippets.CustomAwsSupplier")
+def test_authenticate_unit_success(MockSupplier, MockAwsCreds, MockSession):
+    """Unit test for the main flow using mocks."""
+    mock_response = mock.MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"name": "my-bucket"}
+
+    mock_session_instance = MockSession.return_value
+    mock_session_instance.get.return_value = mock_response
+
+    result = snippets.authenticate_with_aws_credentials(
+        bucket_name="my-bucket",
+        audience="//iam.googleapis.com/...",
+        impersonation_url=None,
+    )
+
+    assert result == {"name": "my-bucket"}
+    MockSupplier.assert_called_once()
+    MockAwsCreds.assert_called_once()
+
+
+# --- System Test (Integration) ---
+
+def test_authenticate_system():
+    """
+    System test that runs against the real API.
+    Skips automatically if custom-credentials-aws-secrets.json is missing or incomplete.
+    """
+    if not os.path.exists("custom-credentials-aws-secrets.json"):
+        pytest.skip(
+            "Skipping system test: custom-credentials-aws-secrets.json not found."
+        )
+
+    with open("custom-credentials-aws-secrets.json", "r") as f:
+        secrets = json.load(f)
+
+    required_keys = [
+        "gcp_workload_audience",
+        "gcs_bucket_name",
+        "aws_access_key_id",
+        "aws_secret_access_key",
+        "aws_region",
+    ]
+    if not all(key in secrets for key in required_keys):
+        pytest.skip(
+            "Skipping system test: custom-credentials-aws-secrets.json is missing required keys."
+        )
+
+    # The main() function handles the auth flow and printing.
+    # We mock the print function to verify the output.
+    with mock.patch("builtins.print") as mock_print:
+        snippets.main()
+
+        # Check for the success message in the print output.
+        output = "\n".join([call.args[0] for call in mock_print.call_args_list])
+        assert "--- SUCCESS! ---" in output
diff --git a/auth/custom-credentials/okta/README.md b/auth/custom-credentials/okta/README.md
new file mode 100644
index 00000000000..dd40f20df11
--- /dev/null
+++ b/auth/custom-credentials/okta/README.md
@@ -0,0 +1,79 @@
+Here is the adapted `README.md` for the Python version of the Okta Custom Credential Supplier sample.
+
+# Running the Custom Okta Credential Supplier Sample (Python)
+
+If you want to use OIDC or SAML 2.0 tokens that cannot be retrieved using methods supported natively by the [google-auth](https://github.com/googleapis/google-auth-library-python) library, a custom `SubjectTokenSupplier` implementation may be specified when creating an identity pool client. The supplier must return a valid, unexpired subject token when called by the GCP credential.
+
+This document provides instructions on how to run the custom Okta credential supplier sample using Python.
+
+## 1. Okta Configuration
+
+Before running the sample, you need to configure an Okta application for Machine-to-Machine (M2M) communication.
+
+### Create an M2M Application in Okta
+
+1.  Log in to your Okta developer console.
+2.  Navigate to **Applications** > **Applications** and click **Create App Integration**.
+3.  Select **API Services** as the sign-on method and click **Next**.
+4.  Give your application a name and click **Save**.
+
+### Obtain Okta Credentials
+
+Once the application is created, you will find the following information in the **General** tab:
+
+*   **Okta Domain**: Your Okta developer domain (e.g., `https://dev-123456.okta.com`).
+*   **Client ID**: The client ID for your application.
+*   **Client Secret**: The client secret for your application.
+
+You will need these values to configure the sample.
+
+## 2. GCP Configuration
+
+You need to configure a Workload Identity Pool in GCP to trust the Okta application.
+
+### Set up Workload Identity Federation
+
+1.  In the Google Cloud Console, navigate to **IAM & Admin** > **Workload Identity Federation**.
+2.  Click **Create Pool** to create a new Workload Identity Pool.
+3.  Add a new **OIDC provider** to the pool.
+4.  Configure the provider with your Okta domain as the issuer URL.
+5.  Map the Okta `sub` (subject) assertion to a GCP principal.
+
+For detailed instructions, refer to the [Workload Identity Federation documentation](https://cloud.google.com/iam/docs/workload-identity-federation).
+
+### GCS Bucket
+
+Ensure you have a GCS bucket that the authenticated user will have access to. You will need the name of this bucket to run the sample.
+
+## 3. Running the Script
+
+To run the sample on your local system, you need to install the dependencies and configure your credentials.
+
+### Install Dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### Configure Credentials
+
+1.  Copy the example secrets file to a new file named `custom-credentials-okta-secrets.json`:
+    ```bash
+    cp custom-credentials-okta-secrets.json.example custom-credentials-okta-secrets.json
+    ```
+2.  Open `custom-credentials-okta-secrets.json` and fill in the following values:
+
+    *   `okta_domain`: Your Okta developer domain (e.g., `https://dev-123456.okta.com`).
+    *   `okta_client_id`: The client ID for your application.
+    *   `okta_client_secret`: The client secret for your application.
+    *   `gcp_workload_audience`: The audience for the GCP Workload Identity Pool. This typically follows the format: `//iam.googleapis.com/projects/YOUR_PROJECT_NUMBER/locations/global/workloadIdentityPools/YOUR_POOL_ID/providers/YOUR_PROVIDER_ID`.
+    *   `gcs_bucket_name`: The name of the GCS bucket to access.
+    *   `gcp_service_account_impersonation_url`: (Optional) The URL for service account impersonation. If set, the script will exchange the federated token for a Google Service Account token. Example: `https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/my-service-account@my-project.iam.gserviceaccount.com:generateAccessToken`.
+
+### Run the Application
+
+```bash
+python3 snippets.py
+```
+
+The script will then authenticate with Okta to get an OIDC token, exchange that token for a GCP federated token (and optionally a Service Account token), and use it to list metadata for the specified GCS bucket.
\ No newline at end of file
diff --git a/auth/custom-credentials/okta/custom-credentials-okta-secrets.json.example b/auth/custom-credentials/okta/custom-credentials-okta-secrets.json.example
new file mode 100644
index 00000000000..fa04fda7cb2
--- /dev/null
+++ b/auth/custom-credentials/okta/custom-credentials-okta-secrets.json.example
@@ -0,0 +1,8 @@
+{
+  "okta_domain": "https://your-okta-domain.okta.com",
+  "okta_client_id": "your-okta-client-id",
+  "okta_client_secret": "your-okta-client-secret",
+  "gcp_workload_audience": "//iam.googleapis.com/projects/123456789/locations/global/workloadIdentityPools/my-pool/providers/my-provider",
+  "gcs_bucket_name": "your-gcs-bucket-name",
+  "gcp_service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/my-service-account@my-project.iam.gserviceaccount.com:generateAccessToken"
+}
diff --git a/auth/custom-credentials/okta/noxfile_config.py b/auth/custom-credentials/okta/noxfile_config.py
new file mode 100644
index 00000000000..834faa2b2f6
--- /dev/null
+++ b/auth/custom-credentials/okta/noxfile_config.py
@@ -0,0 +1,18 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+TEST_CONFIG_OVERRIDE = {
+    # Ignore all versions except 3.9, which is the version available.
+    "ignored_versions": ["2.7", "3.6", "3.7", "3.8", "3.10", "3.11", "3.12", "3.13"],
+}
diff --git a/auth/custom-credentials/okta/requirements-test.txt b/auth/custom-credentials/okta/requirements-test.txt
new file mode 100644
index 00000000000..f47609d2651
--- /dev/null
+++ b/auth/custom-credentials/okta/requirements-test.txt
@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest==7.1.2
diff --git a/auth/custom-credentials/okta/requirements.txt b/auth/custom-credentials/okta/requirements.txt
new file mode 100644
index 00000000000..ca2a46a5a41
--- /dev/null
+++ b/auth/custom-credentials/okta/requirements.txt
@@ -0,0 +1,3 @@
+requests==2.32.3
+google-auth==2.43.0
+python-dotenv==1.1.1
\ No newline at end of file
diff --git a/auth/custom-credentials/okta/snippets.py b/auth/custom-credentials/okta/snippets.py
new file mode 100644
index 00000000000..c1b0718d779
--- /dev/null
+++ b/auth/custom-credentials/okta/snippets.py
@@ -0,0 +1,148 @@
+# Copyright 2025 Google LLC
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# [START auth_custom_credential_supplier_okta]
+import json
+import time
+import urllib.parse
+
+from google.auth import identity_pool
+from google.auth.transport import requests as auth_requests
+import requests
+
+
+class OktaClientCredentialsSupplier:
+    """A custom SubjectTokenSupplier that authenticates with Okta.
+
+    This supplier uses the Client Credentials grant flow for machine-to-machine
+    (M2M) authentication with Okta.
+    """
+
+    def __init__(self, domain, client_id, client_secret):
+        self.okta_token_url = f"{domain.rstrip('/')}/oauth2/default/v1/token"
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.access_token = None
+        self.expiry_time = 0
+
+    def get_subject_token(self, context, request=None) -> str:
+        """Fetches a new token if the current one is expired or missing."""
+        if self.access_token and time.time() < self.expiry_time - 60:
+            return self.access_token
+
+        self._fetch_okta_access_token()
+        return self.access_token
+
+    def _fetch_okta_access_token(self):
+        """Performs the Client Credentials grant flow with Okta."""
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+        }
+        data = {
+            "grant_type": "client_credentials",
+            "scope": "gcp.test.read",  # Set scope as per Okta app config.
+        }
+
+        response = requests.post(
+            self.okta_token_url,
+            headers=headers,
+            data=urllib.parse.urlencode(data),
+            auth=(self.client_id, self.client_secret),
+        )
+        response.raise_for_status()
+
+        token_data = response.json()
+        self.access_token = token_data["access_token"]
+        self.expiry_time = time.time() + token_data["expires_in"]
+
+
+def authenticate_with_okta_credentials(
+    bucket_name, audience, domain, client_id, client_secret, impersonation_url=None
+):
+    """Authenticates using the custom Okta supplier and gets bucket metadata.
+
+    Returns:
+        dict: The bucket metadata response from the Google Cloud Storage API.
+    """
+
+    # 1. Instantiate the custom supplier.
+    okta_supplier = OktaClientCredentialsSupplier(domain, client_id, client_secret)
+
+    # 2. Instantiate the IdentityPoolClient.
+    client = identity_pool.Credentials(
+        audience=audience,
+        subject_token_type="urn:ietf:params:oauth:token-type:jwt",
+        token_url="https://sts.googleapis.com/v1/token",
+        subject_token_supplier=okta_supplier,
+        default_scopes=["https://www.googleapis.com/auth/devstorage.read_write"],
+        service_account_impersonation_url=impersonation_url,
+    )
+
+    # 3. Create an authenticated session.
+    authed_session = auth_requests.AuthorizedSession(client)
+
+    # 4. Make the API Request.
+    bucket_url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}"
+
+    response = authed_session.get(bucket_url)
+    response.raise_for_status()
+
+    return response.json()
+
+
+# [END auth_custom_credential_supplier_okta]
+
+
+def main():
+    try:
+        with open("custom-credentials-okta-secrets.json") as f:
+            secrets = json.load(f)
+    except FileNotFoundError:
+        print(
+            "Could not find custom-credentials-okta-secrets.json."
+        )
+        return
+
+    gcp_audience = secrets.get("gcp_workload_audience")
+    gcs_bucket_name = secrets.get("gcs_bucket_name")
+    sa_impersonation_url = secrets.get("gcp_service_account_impersonation_url")
+
+    okta_domain = secrets.get("okta_domain")
+    okta_client_id = secrets.get("okta_client_id")
+    okta_client_secret = secrets.get("okta_client_secret")
+
+    if not all(
+        [gcp_audience, gcs_bucket_name, okta_domain, okta_client_id, okta_client_secret]
+    ):
+        print("Missing required values in secrets.json.")
+        return
+
+    try:
+        print(f"Retrieving metadata for bucket: {gcs_bucket_name}...")
+        metadata = authenticate_with_okta_credentials(
+            bucket_name=gcs_bucket_name,
+            audience=gcp_audience,
+            domain=okta_domain,
+            client_id=okta_client_id,
+            client_secret=okta_client_secret,
+            impersonation_url=sa_impersonation_url,
+        )
+        print("--- SUCCESS! ---")
+        print(json.dumps(metadata, indent=2))
+    except Exception as e:
+        print(f"Authentication or Request failed: {e}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/auth/custom-credentials/okta/snippets_test.py b/auth/custom-credentials/okta/snippets_test.py
new file mode 100644
index 00000000000..539f33150cc
--- /dev/null
+++ b/auth/custom-credentials/okta/snippets_test.py
@@ -0,0 +1,133 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import os
+import time
+from unittest import mock
+import urllib.parse
+
+import pytest
+
+import snippets
+
+# --- Unit Tests ---
+
+
+def test_init_url_cleaning():
+    """Test that the token URL strips trailing slashes."""
+    s1 = snippets.OktaClientCredentialsSupplier("https://okta.com/", "id", "sec")
+    assert s1.okta_token_url == "https://okta.com/oauth2/default/v1/token"
+
+    s2 = snippets.OktaClientCredentialsSupplier("https://okta.com", "id", "sec")
+    assert s2.okta_token_url == "https://okta.com/oauth2/default/v1/token"
+
+
+@mock.patch("requests.post")
+def test_get_subject_token_fetch(mock_post):
+    """Test fetching a new token from Okta."""
+    supplier = snippets.OktaClientCredentialsSupplier("https://okta.com", "id", "sec")
+
+    mock_response = mock.MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"access_token": "new-token", "expires_in": 3600}
+    mock_post.return_value = mock_response
+
+    token = supplier.get_subject_token(None, None)
+
+    assert token == "new-token"
+    mock_post.assert_called_once()
+
+    # Verify args
+    _, kwargs = mock_post.call_args
+    assert kwargs["auth"] == ("id", "sec")
+
+    sent_data = urllib.parse.parse_qs(kwargs["data"])
+    assert sent_data["grant_type"][0] == "client_credentials"
+
+
+@mock.patch("requests.post")
+def test_get_subject_token_cached(mock_post):
+    """Test that cached token is returned if valid."""
+    supplier = snippets.OktaClientCredentialsSupplier("https://okta.com", "id", "sec")
+    supplier.access_token = "cached-token"
+    supplier.expiry_time = time.time() + 3600
+
+    token = supplier.get_subject_token(None, None)
+
+    assert token == "cached-token"
+    mock_post.assert_not_called()
+
+
+@mock.patch("snippets.auth_requests.AuthorizedSession")
+@mock.patch("snippets.identity_pool.Credentials")
+@mock.patch("snippets.OktaClientCredentialsSupplier")
+def test_authenticate_unit_success(MockSupplier, MockCreds, MockSession):
+    """Unit test for the main Okta auth flow."""
+    mock_response = mock.MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"name": "test-bucket"}
+
+    mock_session_instance = MockSession.return_value
+    mock_session_instance.get.return_value = mock_response
+
+    metadata = snippets.authenticate_with_okta_credentials(
+        bucket_name="test-bucket",
+        audience="test-aud",
+        domain="https://okta.com",
+        client_id="id",
+        client_secret="sec",
+        impersonation_url=None,
+    )
+
+    assert metadata == {"name": "test-bucket"}
+    MockSupplier.assert_called_once()
+    MockCreds.assert_called_once()
+
+
+# --- System Test ---
+
+def test_authenticate_system():
+    """
+    System test that runs against the real API.
+    Skips automatically if custom-credentials-okta-secrets.json is missing or incomplete.
+    """
+    if not os.path.exists("custom-credentials-okta-secrets.json"):
+        pytest.skip(
+            "Skipping system test: custom-credentials-okta-secrets.json not found."
+        )
+
+    with open("custom-credentials-okta-secrets.json", "r") as f:
+        secrets = json.load(f)
+
+    required_keys = [
+        "gcp_workload_audience",
+        "gcs_bucket_name",
+        "okta_domain",
+        "okta_client_id",
+        "okta_client_secret",
+    ]
+    if not all(key in secrets for key in required_keys):
+        pytest.skip(
+            "Skipping system test: custom-credentials-okta-secrets.json is missing required keys."
+        )
+
+    # The main() function handles the auth flow and printing.
+    # We mock the print function to verify the output.
+    with mock.patch("builtins.print") as mock_print:
+        snippets.main()
+
+        # Check for the success message in the print output.
+        output = "\n".join([call.args[0] for call in mock_print.call_args_list])
+        assert "--- SUCCESS! ---" in output