diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index 0ad0580e05..20a248bc4f 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -36,7 +36,7 @@
 /mediacdn/                             @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers
 
 # Other Owners
-/auth/                                 @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/googleapis-auth
+/auth/                                 @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/googleapis-auth @GoogleCloudPlatform/aion-sdk
 /asset/                                @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/cloud-asset-analysis-team
 /billing/                              @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/billing-samples-maintainers
 /dlp/                                  @GoogleCloudPlatform/go-samples-reviewers @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/googleapis-dlp
diff --git a/auth/custom_credential_supplier/aws/README.md b/auth/custom_credential_supplier/aws/README.md
new file mode 100644
index 0000000000..976a911707
--- /dev/null
+++ b/auth/custom_credential_supplier/aws/README.md
@@ -0,0 +1,107 @@
+# 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-cloud-go/auth](https://github.com/vverman/google-cloud-go/tree/main/auth) library, a custom AwsSecurityCredentialsProvider implementation may be specified when creating an AWS client. The supplier must return valid, unexpired AWS security credentials when called by the GCP credential. Currently, using ADC with your AWS workloads is only supported with EC2. An example of a good use case for using a custom credential suppliers is when your workloads are running in other AWS environments, such as ECS, EKS, Fargate, etc.
+
+
+This document provides instructions on how to run the custom credential supplier sample in different environments.
+
+## Running Locally
+
+To run the sample on your local system, you need to configure your AWS and GCP credentials as environment variables.
+
+```bash
+export AWS_ACCESS_KEY_ID="YOUR_AWS_ACCESS_KEY_ID"
+export AWS_SECRET_ACCESS_KEY="YOUR_AWS_SECRET_ACCESS_KEY"
+export GCP_WORKLOAD_AUDIENCE="YOUR_GCP_WORKLOAD_AUDIENCE"
+export GCS_BUCKET_NAME="YOUR_GCS_BUCKET_NAME"
+
+# Optional: If you want to use service account impersonation
+export GCP_SERVICE_ACCOUNT_IMPERSONATION_URL="YOUR_GCP_SERVICE_ACCOUNT_IMPERSONATION_URL"
+
+go run .
+```
+
+## 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.
+
+- Create an IAM OIDC provider for your cluster.
+- Create an IAM role and policy that grants the necessary AWS permissions.
+- Associate the IAM role with a Kubernetes service account.
+
+For detailed steps, see 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
+
+You need to configure your GCP project to trust the AWS IAM role you created. This is done by creating a Workload Identity Pool and Provider in GCP.
+
+- Create a Workload Identity Pool.
+- Create a Workload Identity Provider that trusts the AWS role ARN.
+- Grant the GCP service account the necessary permissions.
+
+### 4. Containerize and Package the Application
+
+Build a Docker image of the Go application and push it to a container registry (e.g., Amazon ECR) that your EKS cluster can access.
+
+```Dockerfile
+FROM golang:1.21
+
+WORKDIR /app
+
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . .
+
+CMD ["go", "run", "./aws"]
+```
+
+### 5. Deploy to EKS
+
+Create a Kubernetes deployment manifest (`pod.yaml`) to deploy your application to the EKS cluster.
+
+```yaml
+apiVersion: v1
+kind: Pod
+metadata:
+  name: custom-credential-pod
+spec:
+  serviceAccountName: your-k8s-service-account # The service account associated with the IAM role
+  containers:
+  - name: gcp-auth-sample
+    image: your-container-image:latest # Your image from ECR
+    env:
+    - name: AWS_REGION
+      value: "your-aws-region"
+    - name: GCP_WORKLOAD_AUDIENCE
+      value: "your-gcp-workload-audience"
+    - name: GOOGLE_CLOUD_PROJECT
+      value: "your-google-cloud-project"
+    # 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"
+```
+
+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
+```
diff --git a/auth/custom_credential_supplier/aws/aws_custom_credential.go b/auth/custom_credential_supplier/aws/aws_custom_credential.go
new file mode 100644
index 0000000000..541849c3a6
--- /dev/null
+++ b/auth/custom_credential_supplier/aws/aws_custom_credential.go
@@ -0,0 +1,175 @@
+// 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
+//
+//     https://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.
+
+package main
+
+// [START auth_custom_credential_supplier_aws]
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+
+	"cloud.google.com/go/auth/credentials/externalaccount"
+	"github.com/aws/aws-sdk-go-v2/config"
+)
+
+// customAwsSupplier implements externalaccount.AwsSecurityCredentialsProvider
+// using the official AWS SDK for Go v2.
+type customAwsSupplier struct{}
+
+// AwsRegion resolves the AWS region using the AWS SDK's default configuration chain.
+// It prioritizes the AWS_REGION environment variable to match standard behavior.
+func (s *customAwsSupplier) AwsRegion(ctx context.Context, opts *externalaccount.RequestOptions) (string, error) {
+	// Explicitly check environment variable first.
+	if region := os.Getenv("AWS_REGION"); region != "" {
+		return region, nil
+	}
+	if region := os.Getenv("AWS_DEFAULT_REGION"); region != "" {
+		return region, nil
+	}
+
+	cfg, err := config.LoadDefaultConfig(ctx)
+	if err != nil {
+		return "", fmt.Errorf("AWS SDK failed to load config for region: %w", err)
+	}
+
+	if cfg.Region == "" {
+		return "", fmt.Errorf("AWS region could not be resolved by SDK; ensure AWS_REGION is set")
+	}
+
+	return cfg.Region, nil
+}
+
+// AwsSecurityCredentials retrieves credentials using the AWS SDK's default provider chain.
+func (s *customAwsSupplier) AwsSecurityCredentials(ctx context.Context, opts *externalaccount.RequestOptions) (*externalaccount.AwsSecurityCredentials, error) {
+	// Load the default AWS configuration.
+	cfg, err := config.LoadDefaultConfig(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("AWS SDK failed to load config for credentials: %w", err)
+	}
+
+	// Retrieve the actual credentials values.
+	creds, err := cfg.Credentials.Retrieve(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("AWS SDK failed to retrieve credentials: %w", err)
+	}
+
+	return &externalaccount.AwsSecurityCredentials{
+		AccessKeyID:     creds.AccessKeyID,
+		SecretAccessKey: creds.SecretAccessKey,
+		SessionToken:    creds.SessionToken,
+	}, nil
+}
+
+// authenticateWithAwsCredentials demonstrates how to use a custom AWS credential supplier
+// to authenticate with Google Cloud and verify access to a specific bucket.
+//
+// impersonationURL is optional. If provided, the credential will exchange the federated
+// token for a Service Account token. If empty, the federated token is used directly.
+func authenticateWithAwsCredentials(w io.Writer, bucketName, audience, impersonationURL string) error {
+	// bucketName := "sample-bucket"
+	// audience := "//iam.googleapis.com/projects/sample-project/locations/global/workloadIdentityPools/sample-pool/providers/sample-provider"
+	// [Optional] impersonationURL := "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/myserviceaccount@iam.gserviceaccount.com:generateAccessToken"
+
+	ctx := context.Background()
+
+	// 1. Initialize Custom AWS Supplier
+	supplier := &customAwsSupplier{}
+
+	// 2. Configure the credentials options
+	opts := &externalaccount.Options{
+		Audience:                       audience,
+		SubjectTokenType:               "urn:ietf:params:aws:token-type:aws4_request",
+		ServiceAccountImpersonationURL: impersonationURL,
+		AwsSecurityCredentialsProvider: supplier,
+		Scopes:                         []string{"https://www.googleapis.com/auth/devstorage.read_write"},
+	}
+
+	// 3. Create the credentials object
+	creds, err := externalaccount.NewCredentials(opts)
+	if err != nil {
+		return fmt.Errorf("externalaccount.NewCredentials: %w", err)
+	}
+
+	// 4. Authenticate and Check Bucket Existence via Raw HTTP
+	bucketURL := fmt.Sprintf("https://storage.googleapis.com/storage/v1/b/%s", bucketName)
+	fmt.Fprintf(w, "Request URL: %s\n", bucketURL)
+	fmt.Fprintln(w, "Attempting to make authenticated request to Google Cloud Storage...")
+
+	// Retrieve a valid token
+	token, err := creds.Token(ctx)
+	if err != nil {
+		return fmt.Errorf("creds.Token: %w", err)
+	}
+
+	// Create HTTP request
+	req, err := http.NewRequestWithContext(ctx, "GET", bucketURL, nil)
+	if err != nil {
+		return fmt.Errorf("http.NewRequest: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token.Value)
+
+	// Execute request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("client.Do: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// Check for failure
+	if resp.StatusCode == http.StatusNotFound {
+		return fmt.Errorf("bucket '%s' does not exist (404)", bucketName)
+	}
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("request returned status %d: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	// Parse and display success output
+	var result map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return fmt.Errorf("json.Decode: %w", err)
+	}
+
+	fmt.Fprintln(w, "\n--- SUCCESS! ---")
+	fmt.Fprintln(w, "Successfully authenticated and retrieved bucket data:")
+	prettyJSON, _ := json.MarshalIndent(result, "", "  ")
+	fmt.Fprintln(w, string(prettyJSON))
+
+	return nil
+}
+
+// [END auth_custom_credential_supplier_aws]
+
+func main() {
+	gcpAudience := os.Getenv("GCP_WORKLOAD_AUDIENCE")
+	// Service account impersonation is optional
+	saImpersonationURL := os.Getenv("GCP_SERVICE_ACCOUNT_IMPERSONATION_URL")
+	gcsBucketName := os.Getenv("GCS_BUCKET_NAME")
+
+	if gcpAudience == "" || gcsBucketName == "" {
+		fmt.Fprintln(os.Stderr, "Missing required environment variables: GCP_WORKLOAD_AUDIENCE, GCS_BUCKET_NAME")
+		os.Exit(1)
+	}
+
+	// Pass os.Stdout as the writer to print to console
+	if err := authenticateWithAwsCredentials(os.Stdout, gcsBucketName, gcpAudience, saImpersonationURL); err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+}
diff --git a/auth/custom_credential_supplier/aws/aws_custom_credential_test.go b/auth/custom_credential_supplier/aws/aws_custom_credential_test.go
new file mode 100644
index 0000000000..cc14316600
--- /dev/null
+++ b/auth/custom_credential_supplier/aws/aws_custom_credential_test.go
@@ -0,0 +1,105 @@
+// 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
+//
+//     https://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.
+
+package main
+
+import (
+	"context"
+	"testing"
+
+	"cloud.google.com/go/auth/credentials/externalaccount"
+)
+
+// TestCustomAwsSupplier_AwsRegion verifies that the supplier correctly
+// resolves the Region from environment variables, respecting precedence.
+func TestCustomAwsSupplier_AwsRegion(t *testing.T) {
+	ctx := context.Background()
+	supplier := &customAwsSupplier{}
+	opts := &externalaccount.RequestOptions{}
+
+	tests := []struct {
+		name       string
+		env        map[string]string
+		wantRegion string
+	}{
+		{
+			name: "AWS_REGION is set (Highest Priority)",
+			env: map[string]string{
+				"AWS_REGION":         "us-west-1",
+				"AWS_DEFAULT_REGION": "us-east-1",
+			},
+			wantRegion: "us-west-1",
+		},
+		{
+			name: "Only AWS_DEFAULT_REGION is set (Fallback)",
+			env: map[string]string{
+				"AWS_REGION":         "",
+				"AWS_DEFAULT_REGION": "eu-central-1",
+			},
+			wantRegion: "eu-central-1",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			// Set environment variables for this test case
+			for k, v := range tc.env {
+				t.Setenv(k, v)
+			}
+
+			got, err := supplier.AwsRegion(ctx, opts)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			if got != tc.wantRegion {
+				t.Errorf("AwsRegion() = %v, want %v", got, tc.wantRegion)
+			}
+		})
+	}
+}
+
+// TestCustomAwsSupplier_AwsSecurityCredentials verifies that the supplier
+// correctly extracts credentials from the AWS environment variables.
+func TestCustomAwsSupplier_AwsSecurityCredentials(t *testing.T) {
+	ctx := context.Background()
+	supplier := &customAwsSupplier{}
+	opts := &externalaccount.RequestOptions{}
+
+	// Mock AWS Credentials via Environment Variables
+	// The AWS SDK v2 automatically picks these up.
+	expectedID := "AKIA_TEST_ACCESS_KEY"
+	expectedSecret := "TEST_SECRET_KEY"
+	expectedToken := "TEST_SESSION_TOKEN"
+
+	t.Setenv("AWS_ACCESS_KEY_ID", expectedID)
+	t.Setenv("AWS_SECRET_ACCESS_KEY", expectedSecret)
+	t.Setenv("AWS_SESSION_TOKEN", expectedToken)
+	t.Setenv("AWS_REGION", "us-east-1") // Required for SDK config to load successfully
+
+	creds, err := supplier.AwsSecurityCredentials(ctx, opts)
+	if err != nil {
+		t.Fatalf("AwsSecurityCredentials failed: %v", err)
+	}
+
+	if creds.AccessKeyID != expectedID {
+		t.Errorf("AccessKeyID = %v, want %v", creds.AccessKeyID, expectedID)
+	}
+	if creds.SecretAccessKey != expectedSecret {
+		t.Errorf("SecretAccessKey = %v, want %v", creds.SecretAccessKey, expectedSecret)
+	}
+	if creds.SessionToken != expectedToken {
+		t.Errorf("SessionToken = %v, want %v", creds.SessionToken, expectedToken)
+	}
+}
diff --git a/auth/custom_credential_supplier/okta/README.md b/auth/custom_credential_supplier/okta/README.md
new file mode 100644
index 0000000000..2cef5747ac
--- /dev/null
+++ b/auth/custom_credential_supplier/okta/README.md
@@ -0,0 +1,62 @@
+# Running the Custom Okta Credential Supplier Sample
+
+If you want to use OIDC or SAML2.0 that cannot be retrieved using methods supported natively by the [google-cloud-go/auth](https://github.com/vverman/google-cloud-go/tree/main/auth) library, a custom SubjectTokenProvider 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.
+
+## 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., `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, set the following environment variables:
+
+```bash
+export OKTA_DOMAIN="your-okta-domain"
+export OKTA_CLIENT_ID="your-okta-client-id"
+export OKTA_CLIENT_SECRET="your-okta-client-secret"
+export GCP_WORKLOAD_IDENTITY_POOL="your-gcp-workload-identity-pool"
+export GCP_WORKLOAD_IDENTITY_PROVIDER="your-gcp-workload-identity-provider"
+export GCS_BUCKET_NAME="your-gcs-bucket-name"
+export GOOGLE_CLOUD_PROJECT="your-google-cloud-project"
+
+go run .
+```
+
+The script will then authenticate with Okta, exchange the Okta token for a GCP token, and use the GCP token to list the objects in the specified GCS bucket.
diff --git a/auth/custom_credential_supplier/okta/okta_custom_credential.go b/auth/custom_credential_supplier/okta/okta_custom_credential.go
new file mode 100644
index 0000000000..2782b15b11
--- /dev/null
+++ b/auth/custom_credential_supplier/okta/okta_custom_credential.go
@@ -0,0 +1,215 @@
+// 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
+//
+//     https://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.
+
+package main
+
+// [START auth_custom_credential_supplier_okta]
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"cloud.google.com/go/auth/credentials/externalaccount"
+)
+
+// oktaClientCredentialsSupplier implements externalaccount.SubjectTokenProvider.
+// It fetches OIDC tokens from Okta using the Client Credentials grant.
+type oktaClientCredentialsSupplier struct {
+	TokenURL     string
+	ClientID     string
+	ClientSecret string
+
+	// Simple in-memory cache for the token.
+	mu          sync.Mutex
+	cachedToken string
+	expiry      time.Time
+}
+
+// SubjectToken returns a valid Okta access token, refreshing it if necessary.
+func (s *oktaClientCredentialsSupplier) SubjectToken(ctx context.Context, opts *externalaccount.RequestOptions) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Check if cached token is valid (with a 60-second buffer).
+	if s.cachedToken != "" && time.Now().Add(60*time.Second).Before(s.expiry) {
+		return s.cachedToken, nil
+	}
+
+	// Fetch a new token.
+	token, expiresIn, err := s.fetchToken(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	s.cachedToken = token
+	s.expiry = time.Now().Add(time.Duration(expiresIn) * time.Second)
+	return s.cachedToken, nil
+}
+
+// fetchToken performs the HTTP request to Okta.
+func (s *oktaClientCredentialsSupplier) fetchToken(ctx context.Context) (string, int64, error) {
+	v := url.Values{}
+	v.Set("grant_type", "client_credentials")
+	// The scope here is specific to the Okta application configuration.
+	v.Set("scope", "gcp.test.read")
+
+	req, err := http.NewRequestWithContext(ctx, "POST", s.TokenURL, strings.NewReader(v.Encode()))
+	if err != nil {
+		return "", 0, fmt.Errorf("http.NewRequest: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Set("Accept", "application/json")
+	req.SetBasicAuth(s.ClientID, s.ClientSecret)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", 0, fmt.Errorf("failed to fetch Okta token: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return "", 0, fmt.Errorf("okta token endpoint returned status: %d, body: %s", resp.StatusCode, string(body))
+	}
+
+	var result struct {
+		AccessToken string `json:"access_token"`
+		ExpiresIn   int64  `json:"expires_in"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return "", 0, fmt.Errorf("failed to decode Okta response: %w", err)
+	}
+
+	if result.AccessToken == "" {
+		return "", 0, fmt.Errorf("okta response missing access_token")
+	}
+
+	return result.AccessToken, result.ExpiresIn, nil
+}
+
+// authenticateWithOktaCredentials demonstrates how to use a custom Okta credential supplier
+// to authenticate with Google Cloud and verify access to a specific bucket.
+func authenticateWithOktaCredentials(w io.Writer, bucketName, audience, domain, clientID, clientSecret, impersonationURL string) error {
+	// bucketName := "sample-bucket"
+	// audience := "//iam.googleapis.com/projects/sample-project/locations/global/workloadIdentityPools/sample-pool/providers/sample-provider"
+	// domain := "https://sample.okta.com"
+	// clientID := "pqr123"
+	// clientSecret := "00124cas62huads68755"
+	// [Optional] impersonationURL := "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/myserviceaccount@iam.gserviceaccount.com:generateAccessToken"
+
+	ctx := context.Background()
+
+	// 1. Instantiate the custom supplier
+	// Note: Adjust the URL path if your Okta org uses a different auth server (e.g., not 'default')
+	oktaTokenURL := fmt.Sprintf("%s/oauth2/default/v1/token", strings.TrimRight(domain, "/"))
+
+	supplier := &oktaClientCredentialsSupplier{
+		TokenURL:     oktaTokenURL,
+		ClientID:     clientID,
+		ClientSecret: clientSecret,
+	}
+
+	// 2. Configure the credentials options
+	opts := &externalaccount.Options{
+		Audience:                       audience,
+		SubjectTokenType:               "urn:ietf:params:oauth:token-type:jwt",
+		SubjectTokenProvider:           supplier,
+		ServiceAccountImpersonationURL: impersonationURL,
+		Scopes:                         []string{"https://www.googleapis.com/auth/devstorage.read_write"},
+	}
+
+	// 3. Create the credentials object
+	creds, err := externalaccount.NewCredentials(opts)
+	if err != nil {
+		return fmt.Errorf("externalaccount.NewCredentials: %w", err)
+	}
+
+	// 4. Authenticate and Check Bucket Existence via Raw HTTP
+	bucketURL := fmt.Sprintf("https://storage.googleapis.com/storage/v1/b/%s", bucketName)
+	fmt.Fprintf(w, "Request URL: %s\n", bucketURL)
+	fmt.Fprintln(w, "Attempting to make authenticated request to Google Cloud Storage...")
+
+	// Retrieve a valid token
+	token, err := creds.Token(ctx)
+	if err != nil {
+		return fmt.Errorf("creds.Token: %w", err)
+	}
+
+	// Create HTTP request
+	req, err := http.NewRequestWithContext(ctx, "GET", bucketURL, nil)
+	if err != nil {
+		return fmt.Errorf("http.NewRequest: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+token.Value)
+
+	// Execute request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("client.Do: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// Check for failure
+	if resp.StatusCode == http.StatusNotFound {
+		return fmt.Errorf("bucket '%s' does not exist (404)", bucketName)
+	}
+	if resp.StatusCode != http.StatusOK {
+		bodyBytes, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("request returned status %d: %s", resp.StatusCode, string(bodyBytes))
+	}
+
+	// Parse and display success output
+	var result map[string]interface{}
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return fmt.Errorf("json.Decode: %w", err)
+	}
+
+	fmt.Fprintln(w, "\n--- SUCCESS! ---")
+	fmt.Fprintln(w, "Successfully authenticated and retrieved bucket data:")
+	prettyJSON, _ := json.MarshalIndent(result, "", "  ")
+	fmt.Fprintln(w, string(prettyJSON))
+
+	return nil
+}
+
+// [END auth_custom_credential_supplier_okta]
+
+// main is provided for local execution and testing purposes.
+func main() {
+	gcpAudience := os.Getenv("GCP_WORKLOAD_AUDIENCE")
+	oktaDomain := os.Getenv("OKTA_DOMAIN")
+	oktaClientID := os.Getenv("OKTA_CLIENT_ID")
+	oktaClientSecret := os.Getenv("OKTA_CLIENT_SECRET")
+	gcsBucketName := os.Getenv("GCS_BUCKET_NAME")
+	// Optional
+	saImpersonationURL := os.Getenv("GCP_SERVICE_ACCOUNT_IMPERSONATION_URL")
+
+	if gcpAudience == "" || oktaDomain == "" || oktaClientID == "" || oktaClientSecret == "" || gcsBucketName == "" {
+		fmt.Fprintln(os.Stderr, "Missing required environment variables: GCP_WORKLOAD_AUDIENCE, OKTA_DOMAIN, OKTA_CLIENT_ID, OKTA_CLIENT_SECRET, GCS_BUCKET_NAME")
+		os.Exit(1)
+	}
+
+	if err := authenticateWithOktaCredentials(os.Stdout, gcsBucketName, gcpAudience, oktaDomain, oktaClientID, oktaClientSecret, saImpersonationURL); err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+}
diff --git a/auth/custom_credential_supplier/okta/okta_custom_credential_test.go b/auth/custom_credential_supplier/okta/okta_custom_credential_test.go
new file mode 100644
index 0000000000..9aef6f6acc
--- /dev/null
+++ b/auth/custom_credential_supplier/okta/okta_custom_credential_test.go
@@ -0,0 +1,155 @@
+// 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
+//
+//     https://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.
+
+package main
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"cloud.google.com/go/auth/credentials/externalaccount"
+)
+
+// TestOktaClientCredentialsSupplier_SubjectToken verifies the fetch logic
+// and proper handling of caching.
+func TestOktaClientCredentialsSupplier_SubjectToken(t *testing.T) {
+	// 1. Setup Mock Okta Server
+	mockClientID := "client-id"
+	mockClientSecret := "client-secret"
+	expectedToken := "mock-okta-jwt-token"
+
+	// We count how many times the server is hit to verify caching
+	serverHitCount := 0
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		serverHitCount++
+
+		// Validate Method
+		if r.Method != "POST" {
+			t.Errorf("Expected POST request, got %s", r.Method)
+		}
+
+		// Validate Content Type
+		if r.Header.Get("Content-Type") != "application/x-www-form-urlencoded" {
+			t.Errorf("Expected Content-Type application/x-www-form-urlencoded, got %s", r.Header.Get("Content-Type"))
+		}
+
+		// Validate Basic Auth
+		authHeader := r.Header.Get("Authorization")
+		wantAuth := "Basic " + base64.StdEncoding.EncodeToString([]byte(mockClientID+":"+mockClientSecret))
+		if authHeader != wantAuth {
+			t.Errorf("Invalid Basic Auth header. Got %s, want %s", authHeader, wantAuth)
+		}
+
+		// Validate Body Params
+		if err := r.ParseForm(); err != nil {
+			t.Fatal(err)
+		}
+		if r.FormValue("grant_type") != "client_credentials" {
+			t.Errorf("Expected grant_type=client_credentials, got %s", r.FormValue("grant_type"))
+		}
+		if r.FormValue("scope") != "gcp.test.read" {
+			t.Errorf("Expected scope=gcp.test.read, got %s", r.FormValue("scope"))
+		}
+
+		// Return Success JSON
+		w.Header().Set("Content-Type", "application/json")
+		response := map[string]interface{}{
+			"access_token": expectedToken,
+			"expires_in":   3600, // 1 hour
+		}
+		json.NewEncoder(w).Encode(response)
+	}))
+	defer ts.Close()
+
+	// 2. Initialize Supplier with Mock URL
+	supplier := &oktaClientCredentialsSupplier{
+		TokenURL:     ts.URL,
+		ClientID:     mockClientID,
+		ClientSecret: mockClientSecret,
+	}
+	ctx := context.Background()
+
+	// 3. First Call: Should hit the server
+	token, err := supplier.SubjectToken(ctx, &externalaccount.RequestOptions{})
+	if err != nil {
+		t.Fatalf("First SubjectToken call failed: %v", err)
+	}
+	if token != expectedToken {
+		t.Errorf("Got token %s, want %s", token, expectedToken)
+	}
+	if serverHitCount != 1 {
+		t.Errorf("Expected 1 server hit, got %d", serverHitCount)
+	}
+
+	// 4. Second Call: Should use cache (no server hit)
+	token, err = supplier.SubjectToken(ctx, &externalaccount.RequestOptions{})
+	if err != nil {
+		t.Fatalf("Second SubjectToken call failed: %v", err)
+	}
+	if token != expectedToken {
+		t.Errorf("Got token %s, want %s", token, expectedToken)
+	}
+	if serverHitCount != 1 {
+		t.Errorf("Expected server hit count to remain 1 (cached), got %d", serverHitCount)
+	}
+}
+
+// TestOktaClientCredentialsSupplier_ExpiredCache verifies that the supplier
+// refetches if the cache is expired.
+func TestOktaClientCredentialsSupplier_ExpiredCache(t *testing.T) {
+	serverHitCount := 0
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		serverHitCount++
+		w.Header().Set("Content-Type", "application/json")
+		// Return a token that expires very soon (e.g., 1 second)
+		// The logic has a 60s buffer, so anything < 60s should be treated as expired immediately for next call
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"access_token": fmt.Sprintf("token-%d", serverHitCount),
+			"expires_in":   30,
+		})
+	}))
+	defer ts.Close()
+
+	supplier := &oktaClientCredentialsSupplier{
+		TokenURL:     ts.URL,
+		ClientID:     "id",
+		ClientSecret: "secret",
+	}
+	ctx := context.Background()
+
+	// First Call
+	_, err := supplier.SubjectToken(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if serverHitCount != 1 {
+		t.Errorf("Expected 1 hit, got %d", serverHitCount)
+	}
+
+	// Second Call immediately after
+	// Since expires_in (30s) is < buffer (60s), the supplier considers it expired immediately.
+	_, err = supplier.SubjectToken(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if serverHitCount != 2 {
+		t.Errorf("Expected 2 hits (cache should be invalid due to short expiry), got %d", serverHitCount)
+	}
+}