From 6572262230d8736150ed1235119fac7f78aa936c Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Tue, 13 Oct 2026 14:53:45 +0200
Subject: [PATCH 01/11] generate signature and put it into token

---
 oks_cli/utils.py | 115 +++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 111 insertions(+), 4 deletions(-)

diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index 432b8db..24aa284 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -10,7 +10,7 @@
 import pathlib
 import traceback
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 import OpenSSL
 import shutil
 import prettytable
@@ -18,6 +18,8 @@
 import human_readable
 import base64
 import sys
+import hmac
+import hashlib
 
 from click.shell_completion import CompletionItem
 
@@ -144,6 +146,106 @@ def do_request(method, path, *args, **kwargs):
             errors = {"Error": f"Failed to reach the endpoint {url} ({err.__class__.__name__})"}
             raise JSONClickException(json.dumps(errors))
 
+def sign(key, msg):
+    """Generate HMAC-SHA256 signature of a message using a secret key."""
+    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
+
+def get_signature_key(key, date_stamp, region_name, service_name, scheme="OSC4"):
+    """Derive signature key (OSC4 or AWS4) using the secret access key and request details."""
+    k_date = sign((scheme + key).encode('utf-8'), date_stamp)
+    k_region = sign(k_date, region_name)
+    k_service = sign(k_region, service_name)
+    k_signing = sign(k_service, f"{scheme.lower()}_request")
+    return k_signing
+
+def hash_sha256(message):
+    """Return the SHA-256 hash (hex) of the input message."""
+    return hashlib.sha256(message.encode('utf-8')).hexdigest()
+
+def create_string_to_sign(amz_date, credential_scope, canonical_request_hash, scheme="OSC4"):
+    """Build the 'String to Sign' from request metadata and canonical request hash."""
+    return f"{scheme}-HMAC-SHA256\n{amz_date}\n{credential_scope}\n{canonical_request_hash}"
+
+def generate_signature(ak, sk, region, service, amz_date, uri_path):
+    """
+    Generate AWS Signature Version 4 for a request using credentials and metadata.
+    
+    Returns a dictionary with AccessKey, Signature, and StringToSign.
+    """
+    method = "POST"
+    canonical_uri = uri_path
+    canonical_querystring = ""
+    canonical_headers = f"content-type:application/json\nhost:{service}.{region}.outscale.com\nx-osc-date:{amz_date}\n"
+    signed_headers = "content-type;host;x-osc-date"
+    payload_hash = hash_sha256('')  # Empty body
+
+    canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
+    canonical_request_hash = hash_sha256(canonical_request)
+
+    date_stamp = amz_date[:8]
+    credential_scope = f"{date_stamp}/{region}/{service}/osc4_request"
+    string_to_sign = create_string_to_sign(amz_date, credential_scope, canonical_request_hash)
+
+    signing_key = get_signature_key(sk, date_stamp, region, service)
+    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+
+    return signature
+
+def generate_oos_head_bucket_signature(ak, sk, region, amz_date, bucket):
+    """
+    Generate AWS Signature Version 4 for an OOS (S3) HeadBucket request (path-style).
+
+    Returns a dictionary with AccessKey, Signature, StringToSign and PayloadHash.
+    """
+    service = "s3"
+    method = "HEAD"
+    canonical_uri = f"/{bucket}"
+    canonical_querystring = ""
+    payload_hash = hash_sha256('')  # Empty body
+    canonical_headers = f"host:oos.{region}.outscale.com\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n"
+    signed_headers = "host;x-amz-content-sha256;x-amz-date"
+
+    canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
+    canonical_request_hash = hash_sha256(canonical_request)
+
+    date_stamp = amz_date[:8]
+    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
+    string_to_sign = create_string_to_sign(amz_date, credential_scope, canonical_request_hash, scheme="AWS4")
+
+    signing_key = get_signature_key(sk, date_stamp, region, service, scheme="AWS4")
+    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+
+    return signature
+
+def build_signed_requests_token(ak, sk, region):
+    """
+    Sign ReadAccounts (OSC API) and HeadBucket (OOS) requests with ak/sk.
+
+    Returns both signatures packed into a base64-encoded JSON token.
+    """
+    read_accounts_path = "/api/v1/ReadAccounts"
+    head_bucket_name = "s3-bucket-oks-eim-user-check"
+
+    amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
+
+    read_accounts_signature = generate_signature(ak, sk, region, "api", amz_date, read_accounts_path)
+    head_bucket_signature = generate_oos_head_bucket_signature(ak, sk, region, amz_date, head_bucket_name)
+
+    signed_requests = {
+        "AccessKey": ak,
+        "AmzDate": amz_date,
+        "Region": region,
+        "ReadAccounts": {
+            "Signature": read_accounts_signature,
+        },
+        "HeadBucket": {
+            "Signature": head_bucket_signature,
+            "Bucket": head_bucket_name,
+        },
+    }
+
+    return base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
+
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""
     headers = {
@@ -163,9 +265,14 @@ def build_headers():
     if is_jwt_enabled() and is_tokens_valid():
         headers["AccessToken"] = get_token('access_token')
         headers["RefreshToken"] = get_token('refresh_token')
-    elif os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_ACCESS_KEY"):
-        headers["AccessKey"] =  os.getenv("OKS_ACCESS_KEY")
-        headers["SecretKey"] =  os.getenv("OKS_SECRET_KEY")
+    elif os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY"):
+        region = os.getenv("OKS_REGION")
+        if not region:
+            raise click.ClickException("region_name is required in the profile to sign requests with ak/sk")
+
+        token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region)
+        headers["Authorization"] = f"OSC-Signed-Requests {token}"
+
     elif os.getenv("OKS_USERNAME") and os.getenv("OKS_PASSWORD"):
         username = os.getenv("OKS_USERNAME")
         password = os.getenv("OKS_PASSWORD")

From f2af6da884bb78f45fdda96a749c1dcd5bb7f3ab Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Tue, 13 Oct 2026 15:02:17 +0200
Subject: [PATCH 02/11] use aws sign intead of osc

---
 oks_cli/utils.py | 83 ++++++++++++++++++++----------------------------
 1 file changed, 35 insertions(+), 48 deletions(-)

diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index 24aa284..e7eb1f2 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -150,72 +150,60 @@ def sign(key, msg):
     """Generate HMAC-SHA256 signature of a message using a secret key."""
     return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
 
-def get_signature_key(key, date_stamp, region_name, service_name, scheme="OSC4"):
-    """Derive signature key (OSC4 or AWS4) using the secret access key and request details."""
-    k_date = sign((scheme + key).encode('utf-8'), date_stamp)
+def get_signature_key(key, date_stamp, region_name, service_name):
+    """Derive AWS4 signature key using the secret access key and request details."""
+    k_date = sign(('AWS4' + key).encode('utf-8'), date_stamp)
     k_region = sign(k_date, region_name)
     k_service = sign(k_region, service_name)
-    k_signing = sign(k_service, f"{scheme.lower()}_request")
+    k_signing = sign(k_service, 'aws4_request')
     return k_signing
 
 def hash_sha256(message):
     """Return the SHA-256 hash (hex) of the input message."""
     return hashlib.sha256(message.encode('utf-8')).hexdigest()
 
-def create_string_to_sign(amz_date, credential_scope, canonical_request_hash, scheme="OSC4"):
-    """Build the 'String to Sign' from request metadata and canonical request hash."""
-    return f"{scheme}-HMAC-SHA256\n{amz_date}\n{credential_scope}\n{canonical_request_hash}"
+def create_string_to_sign(amz_date, credential_scope, canonical_request_hash):
+    """Build the AWS4 'String to Sign' from request metadata and canonical request hash."""
+    return f"AWS4-HMAC-SHA256\n{amz_date}\n{credential_scope}\n{canonical_request_hash}"
 
-def generate_signature(ak, sk, region, service, amz_date, uri_path):
+def generate_aws4_signature(sk, region, service, amz_date, method, canonical_uri, headers):
     """
-    Generate AWS Signature Version 4 for a request using credentials and metadata.
-    
-    Returns a dictionary with AccessKey, Signature, and StringToSign.
-    """
-    method = "POST"
-    canonical_uri = uri_path
-    canonical_querystring = ""
-    canonical_headers = f"content-type:application/json\nhost:{service}.{region}.outscale.com\nx-osc-date:{amz_date}\n"
-    signed_headers = "content-type;host;x-osc-date"
-    payload_hash = hash_sha256('')  # Empty body
-
-    canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
-    canonical_request_hash = hash_sha256(canonical_request)
-
-    date_stamp = amz_date[:8]
-    credential_scope = f"{date_stamp}/{region}/{service}/osc4_request"
-    string_to_sign = create_string_to_sign(amz_date, credential_scope, canonical_request_hash)
-
-    signing_key = get_signature_key(sk, date_stamp, region, service)
-    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+    Generate AWS Signature Version 4 for a request with an empty body.
 
-    return signature
-
-def generate_oos_head_bucket_signature(ak, sk, region, amz_date, bucket):
+    All entries of `headers` (lowercase name -> value) are signed.
+    Returns the hex signature.
     """
-    Generate AWS Signature Version 4 for an OOS (S3) HeadBucket request (path-style).
-
-    Returns a dictionary with AccessKey, Signature, StringToSign and PayloadHash.
-    """
-    service = "s3"
-    method = "HEAD"
-    canonical_uri = f"/{bucket}"
     canonical_querystring = ""
     payload_hash = hash_sha256('')  # Empty body
-    canonical_headers = f"host:oos.{region}.outscale.com\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n"
-    signed_headers = "host;x-amz-content-sha256;x-amz-date"
+    signed_header_names = sorted(headers)
+    canonical_headers = "".join(f"{name}:{headers[name]}\n" for name in signed_header_names)
+    signed_headers = ";".join(signed_header_names)
 
     canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
     canonical_request_hash = hash_sha256(canonical_request)
 
     date_stamp = amz_date[:8]
     credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
-    string_to_sign = create_string_to_sign(amz_date, credential_scope, canonical_request_hash, scheme="AWS4")
-
-    signing_key = get_signature_key(sk, date_stamp, region, service, scheme="AWS4")
-    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+    string_to_sign = create_string_to_sign(amz_date, credential_scope, canonical_request_hash)
 
-    return signature
+    signing_key = get_signature_key(sk, date_stamp, region, service)
+    return hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
+
+def generate_read_accounts_signature(sk, region, amz_date):
+    """Generate AWS4 signature for an OUTSCALE API ReadAccounts request."""
+    return generate_aws4_signature(sk, region, "api", amz_date, "POST", "/api/v1/ReadAccounts", {
+        "content-type": "application/json",
+        "host": f"api.{region}.outscale.com",
+        "x-amz-date": amz_date,
+    })
+
+def generate_oos_head_bucket_signature(sk, region, amz_date, bucket):
+    """Generate AWS4 signature for an OOS (S3) HeadBucket request (path-style)."""
+    return generate_aws4_signature(sk, region, "s3", amz_date, "HEAD", f"/{bucket}", {
+        "host": f"oos.{region}.outscale.com",
+        "x-amz-content-sha256": hash_sha256(''),
+        "x-amz-date": amz_date,
+    })
 
 def build_signed_requests_token(ak, sk, region):
     """
@@ -223,13 +211,12 @@ def build_signed_requests_token(ak, sk, region):
 
     Returns both signatures packed into a base64-encoded JSON token.
     """
-    read_accounts_path = "/api/v1/ReadAccounts"
     head_bucket_name = "s3-bucket-oks-eim-user-check"
 
     amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
 
-    read_accounts_signature = generate_signature(ak, sk, region, "api", amz_date, read_accounts_path)
-    head_bucket_signature = generate_oos_head_bucket_signature(ak, sk, region, amz_date, head_bucket_name)
+    read_accounts_signature = generate_read_accounts_signature(sk, region, amz_date)
+    head_bucket_signature = generate_oos_head_bucket_signature(sk, region, amz_date, head_bucket_name)
 
     signed_requests = {
         "AccessKey": ak,

From 7c7eb9bd6be897c50f433ea77e0a29722e0a62d1 Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Tue, 13 Oct 2026 15:30:16 +0200
Subject: [PATCH 03/11] and --aws-sigv4 flag to generate oks-sigv4-v1 token

---
 oks_cli/main.py  |  6 +++++-
 oks_cli/utils.py | 19 ++++++++++++-------
 2 files changed, 17 insertions(+), 8 deletions(-)

diff --git a/oks_cli/main.py b/oks_cli/main.py
index 12dff4f..5150823 100755
--- a/oks_cli/main.py
+++ b/oks_cli/main.py
@@ -19,8 +19,9 @@
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)
 @click.option('--verbose', '-v', count=True, help="Increase verbosity")
+@click.option('--aws-sigv4', is_flag=True, help="Authenticate with AWS SigV4 signed requests")
 @click.pass_context
-def cli(ctx, project_name, cluster_name, profile, verbose):
+def cli(ctx, project_name, cluster_name, profile, verbose, aws_sigv4):
     """
     CLI tool to manage projects and clusters.
 
@@ -29,6 +30,9 @@ def cli(ctx, project_name, cluster_name, profile, verbose):
     """
     ctx_update(ctx, project_name, cluster_name, profile)
 
+    if aws_sigv4:
+        os.environ["OKS_AWS_SIGV4"] = "1"
+
     LOGLEVEL = os.environ.get('LOGLEVEL', 'WARNING').upper()
 
     if verbose >= 1:
diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index e7eb1f2..8214587 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -106,6 +106,8 @@ def do_request(method, path, *args, **kwargs):
 
     kwargs.setdefault('headers', {}).update(headers)
 
+    logging.debug("request headers: %s", kwargs['headers'])
+
     logging.info("%s request %s?%s", method, url,
                  urlencode(kwargs.get('params', {})))
 
@@ -212,6 +214,7 @@ def build_signed_requests_token(ak, sk, region):
     Returns both signatures packed into a base64-encoded JSON token.
     """
     head_bucket_name = "s3-bucket-oks-eim-user-check"
+    prefix = "oks-sigv4-v1."
 
     amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
 
@@ -231,7 +234,7 @@ def build_signed_requests_token(ak, sk, region):
         },
     }
 
-    return base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
+    return prefix + base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
 
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""
@@ -249,17 +252,19 @@ def build_headers():
             if _h and _n:
                 headers[_h.strip()] = _n.strip() 
 
-    if is_jwt_enabled() and is_tokens_valid():
-        headers["AccessToken"] = get_token('access_token')
-        headers["RefreshToken"] = get_token('refresh_token')
-    elif os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY"):
+    if os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY") and os.getenv("OKS_AWS_SIGV4"):
         region = os.getenv("OKS_REGION")
         if not region:
             raise click.ClickException("region_name is required in the profile to sign requests with ak/sk")
 
         token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region)
-        headers["Authorization"] = f"OSC-Signed-Requests {token}"
-
+        headers["Authorization"] = f"Bearer {token}"
+    elif is_jwt_enabled() and is_tokens_valid():
+        headers["AccessToken"] = get_token('access_token')
+        headers["RefreshToken"] = get_token('refresh_token')
+    elif os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY"):
+        headers["AccessKey"] = os.getenv("OKS_ACCESS_KEY")
+        headers["SecretKey"] = os.getenv("OKS_SECRET_KEY")
     elif os.getenv("OKS_USERNAME") and os.getenv("OKS_PASSWORD"):
         username = os.getenv("OKS_USERNAME")
         password = os.getenv("OKS_PASSWORD")

From e10693d8741e296a8bf3031d8acee8cbe67521a5 Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Wed, 14 Oct 2026 14:48:35 +0200
Subject: [PATCH 04/11] provide token in the kube-friendly format

---
 oks_cli/cluster.py | 39 +++++++++++++++++++++++++++++++++++++--
 1 file changed, 37 insertions(+), 2 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 75b15ff..4995842 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -6,7 +6,7 @@
 
 import time
 import os
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 import dateutil.parser
 import human_readable
 import pathlib
@@ -23,7 +23,8 @@
                    ctx_update, set_cluster_id, get_cluster_id, get_project_id,  \
                    get_template, get_cluster_name, format_changed_row,          \
                    is_interesting_status, profile_completer, project_completer, \
-                   kubeconfig_parse_fields, print_table, format_row, apply_set_fields
+                   kubeconfig_parse_fields, print_table, format_row, apply_set_fields, \
+                   build_signed_requests_token
 
 from .profile import add_profile
 from .project import project_create, project_login
@@ -717,6 +718,40 @@ def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, outp
         else:
             click.echo(kubeconfig)
 
+# GET TOKEN
+@cluster.command('token', help="Print an AWS SigV4 token as a Kubernetes ExecCredential")
+@click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
+@click.pass_context
+def cluster_token_command(ctx, profile):
+    """CLI command to print a SigV4 token in the client.authentication.k8s.io/v1 ExecCredential format."""
+    _, _, profile = ctx_update(ctx, None, None, profile)
+    login_profile(profile)
+
+    access_key = os.getenv("OKS_ACCESS_KEY")
+    secret_key = os.getenv("OKS_SECRET_KEY")
+    region = os.getenv("OKS_REGION")
+
+    if not access_key or not secret_key:
+        raise click.ClickException("An ak/sk profile is required to generate a token")
+    if not region:
+        raise click.ClickException("region_name is required in the profile to generate a token")
+
+    # SigV4 signatures are accepted for 15 minutes, expire the token slightly earlier
+    expiration = datetime.now(timezone.utc) + timedelta(minutes=14)
+    token = build_signed_requests_token(access_key, secret_key, region)
+
+    exec_credential = {
+        "apiVersion": "client.authentication.k8s.io/v1",
+        "kind": "ExecCredential",
+        "spec": {},
+        "status": {
+            "expirationTimestamp": expiration.strftime('%Y-%m-%dT%H:%M:%SZ'),
+            "token": token,
+        },
+    }
+
+    click.echo(json.dumps(exec_credential))
+
 
 def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=False):
     """Run a kubectl command using the cached kubeconfig for the specified cluster, refreshing it if needed."""

From abcee27137ac30849dbee1da0c1b01924f4a4c83 Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Thu, 15 Oct 2026 13:46:14 +0200
Subject: [PATCH 05/11] implement oks-sigv4-v2 token based on the eim ReadUser

---
 oks_cli/cluster.py |  2 +-
 oks_cli/main.py    | 20 ++++++++++++----
 oks_cli/utils.py   | 57 ++++++++++++++++++++++++++++++++++++++--------
 3 files changed, 64 insertions(+), 15 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 4995842..45676ac 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -738,7 +738,7 @@ def cluster_token_command(ctx, profile):
 
     # SigV4 signatures are accepted for 15 minutes, expire the token slightly earlier
     expiration = datetime.now(timezone.utc) + timedelta(minutes=14)
-    token = build_signed_requests_token(access_key, secret_key, region)
+    token = build_signed_requests_token(access_key, secret_key, region, os.getenv("OKS_AWS_SIGV4", "v1"))
 
     exec_credential = {
         "apiVersion": "client.authentication.k8s.io/v1",
diff --git a/oks_cli/main.py b/oks_cli/main.py
index 5150823..5687d2a 100755
--- a/oks_cli/main.py
+++ b/oks_cli/main.py
@@ -3,6 +3,7 @@
 import os
 import json
 import logging
+import re
 from .project import project
 from .cluster import cluster
 from .profile import profile
@@ -11,15 +12,26 @@
 from .netpeering import netpeering
 from .user import user
 
-from .utils import ctx_update, install_completions, profile_completer, cluster_completer, project_completer
+from .utils import ctx_update, install_completions, profile_completer, cluster_completer, project_completer, \
+                   SIGV4_TOKEN_BUILDERS
+
+class CLIGroup(click.Group):
+    def parse_args(self, ctx, args):
+        # Click would consume the subcommand name as the value of a bare `--aws-sigv4`
+        for i, arg in enumerate(args):
+            if arg == "--aws-sigv4" and (i + 1 == len(args) or not re.fullmatch(r"v\d+", args[i + 1])):
+                args = args[:i + 1] + ["v1"] + args[i + 1:]
+                break
+        return super().parse_args(ctx, args)
 
 # Main CLI entry point
-@click.group(invoke_without_command=True)
+@click.group(cls=CLIGroup, invoke_without_command=True)
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)
 @click.option('--verbose', '-v', count=True, help="Increase verbosity")
-@click.option('--aws-sigv4', is_flag=True, help="Authenticate with AWS SigV4 signed requests")
+@click.option('--aws-sigv4', type=click.Choice(list(SIGV4_TOKEN_BUILDERS)),
+              help="Authenticate with an AWS SigV4 signed token: v1 (ReadAccounts + HeadBucket, used when no version is given) or v2 (EIM GetUser)")
 @click.pass_context
 def cli(ctx, project_name, cluster_name, profile, verbose, aws_sigv4):
     """
@@ -31,7 +43,7 @@ def cli(ctx, project_name, cluster_name, profile, verbose, aws_sigv4):
     ctx_update(ctx, project_name, cluster_name, profile)
 
     if aws_sigv4:
-        os.environ["OKS_AWS_SIGV4"] = "1"
+        os.environ["OKS_AWS_SIGV4"] = aws_sigv4
 
     LOGLEVEL = os.environ.get('LOGLEVEL', 'WARNING').upper()
 
diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index 8214587..6d2696b 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -168,15 +168,15 @@ def create_string_to_sign(amz_date, credential_scope, canonical_request_hash):
     """Build the AWS4 'String to Sign' from request metadata and canonical request hash."""
     return f"AWS4-HMAC-SHA256\n{amz_date}\n{credential_scope}\n{canonical_request_hash}"
 
-def generate_aws4_signature(sk, region, service, amz_date, method, canonical_uri, headers):
+def generate_aws4_signature(sk, region, service, amz_date, method, canonical_uri, headers, payload=''):
     """
-    Generate AWS Signature Version 4 for a request with an empty body.
+    Generate AWS Signature Version 4 for a request.
 
     All entries of `headers` (lowercase name -> value) are signed.
     Returns the hex signature.
     """
     canonical_querystring = ""
-    payload_hash = hash_sha256('')  # Empty body
+    payload_hash = hash_sha256(payload)
     signed_header_names = sorted(headers)
     canonical_headers = "".join(f"{name}:{headers[name]}\n" for name in signed_header_names)
     signed_headers = ";".join(signed_header_names)
@@ -207,21 +207,33 @@ def generate_oos_head_bucket_signature(sk, region, amz_date, bucket):
         "x-amz-date": amz_date,
     })
 
-def build_signed_requests_token(ak, sk, region):
+def generate_eim_get_user_signature(sk, region, amz_date):
+    """Generate AWS4 signature for an OUTSCALE EIM GetUser request."""
+    # The verifier must replay this exact body, its hash is part of the signature
+    body = "Version=2010-05-08&Action=GetUser"
+    return generate_aws4_signature(sk, region, "eim", amz_date, "POST", "/", {
+        "host": f"eim.{region}.outscale.com",
+        "x-amz-date": amz_date,
+    }, payload=body)
+
+def encode_signed_requests_token(prefix, signed_requests):
+    """Pack signed request data into a prefixed base64-encoded JSON token."""
+    return prefix + base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
+
+def build_signed_requests_token_v1(ak, sk, region):
     """
     Sign ReadAccounts (OSC API) and HeadBucket (OOS) requests with ak/sk.
 
-    Returns both signatures packed into a base64-encoded JSON token.
+    Returns both signatures packed into an oks-sigv4-v1 token.
     """
     head_bucket_name = "s3-bucket-oks-eim-user-check"
-    prefix = "oks-sigv4-v1."
 
     amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
 
     read_accounts_signature = generate_read_accounts_signature(sk, region, amz_date)
     head_bucket_signature = generate_oos_head_bucket_signature(sk, region, amz_date, head_bucket_name)
 
-    signed_requests = {
+    return encode_signed_requests_token("oks-sigv4-v1.", {
         "AccessKey": ak,
         "AmzDate": amz_date,
         "Region": region,
@@ -232,9 +244,33 @@ def build_signed_requests_token(ak, sk, region):
             "Signature": head_bucket_signature,
             "Bucket": head_bucket_name,
         },
-    }
+    })
 
-    return prefix + base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
+def build_signed_requests_token_v2(ak, sk, region):
+    """
+    Sign an EIM GetUser request with ak/sk.
+
+    Returns the signature packed into an oks-sigv4-v2 token.
+    """
+    amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
+
+    return encode_signed_requests_token("oks-sigv4-v2.", {
+        "AccessKey": ak,
+        "AmzDate": amz_date,
+        "Region": region,
+        "Signature": generate_eim_get_user_signature(sk, region, amz_date),
+    })
+
+SIGV4_TOKEN_BUILDERS = {
+    "v1": build_signed_requests_token_v1,
+    "v2": build_signed_requests_token_v2,
+}
+
+def build_signed_requests_token(ak, sk, region, version="v1"):
+    """Build a SigV4 signed requests token of the given version."""
+    if version not in SIGV4_TOKEN_BUILDERS:
+        raise click.ClickException(f"Unsupported SigV4 token version: {version}")
+    return SIGV4_TOKEN_BUILDERS[version](ak, sk, region)
 
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""
@@ -257,7 +293,8 @@ def build_headers():
         if not region:
             raise click.ClickException("region_name is required in the profile to sign requests with ak/sk")
 
-        token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region)
+        token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region,
+                                            os.getenv("OKS_AWS_SIGV4"))
         headers["Authorization"] = f"Bearer {token}"
     elif is_jwt_enabled() and is_tokens_valid():
         headers["AccessToken"] = get_token('access_token')

From 429117f526a6f14a325785eb329ffb294962075b Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Thu, 15 Oct 2026 15:24:49 +0200
Subject: [PATCH 06/11] include x-oks-target signed header

---
 oks_cli/cluster.py |  5 +++--
 oks_cli/utils.py   | 33 ++++++++++++++++++++++-----------
 2 files changed, 25 insertions(+), 13 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 45676ac..99de4a9 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -720,9 +720,10 @@ def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, outp
 
 # GET TOKEN
 @cluster.command('token', help="Print an AWS SigV4 token as a Kubernetes ExecCredential")
+@click.option('--url', help="URL to sign into the token as the X-OKS-Target header")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.pass_context
-def cluster_token_command(ctx, profile):
+def cluster_token_command(ctx, url, profile):
     """CLI command to print a SigV4 token in the client.authentication.k8s.io/v1 ExecCredential format."""
     _, _, profile = ctx_update(ctx, None, None, profile)
     login_profile(profile)
@@ -738,7 +739,7 @@ def cluster_token_command(ctx, profile):
 
     # SigV4 signatures are accepted for 15 minutes, expire the token slightly earlier
     expiration = datetime.now(timezone.utc) + timedelta(minutes=14)
-    token = build_signed_requests_token(access_key, secret_key, region, os.getenv("OKS_AWS_SIGV4", "v1"))
+    token = build_signed_requests_token(access_key, secret_key, region, os.getenv("OKS_AWS_SIGV4", "v1"), url)
 
     exec_credential = {
         "apiVersion": "client.authentication.k8s.io/v1",
diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index 6d2696b..676d400 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -191,47 +191,57 @@ def generate_aws4_signature(sk, region, service, amz_date, method, canonical_uri
     signing_key = get_signature_key(sk, date_stamp, region, service)
     return hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
 
-def generate_read_accounts_signature(sk, region, amz_date):
+def generate_read_accounts_signature(sk, region, amz_date, extra_headers=None):
     """Generate AWS4 signature for an OUTSCALE API ReadAccounts request."""
     return generate_aws4_signature(sk, region, "api", amz_date, "POST", "/api/v1/ReadAccounts", {
         "content-type": "application/json",
         "host": f"api.{region}.outscale.com",
         "x-amz-date": amz_date,
+        **(extra_headers or {}),
     })
 
-def generate_oos_head_bucket_signature(sk, region, amz_date, bucket):
+def generate_oos_head_bucket_signature(sk, region, amz_date, bucket, extra_headers=None):
     """Generate AWS4 signature for an OOS (S3) HeadBucket request (path-style)."""
     return generate_aws4_signature(sk, region, "s3", amz_date, "HEAD", f"/{bucket}", {
         "host": f"oos.{region}.outscale.com",
         "x-amz-content-sha256": hash_sha256(''),
         "x-amz-date": amz_date,
+        **(extra_headers or {}),
     })
 
-def generate_eim_get_user_signature(sk, region, amz_date):
+def generate_eim_get_user_signature(sk, region, amz_date, extra_headers=None):
     """Generate AWS4 signature for an OUTSCALE EIM GetUser request."""
     # The verifier must replay this exact body, its hash is part of the signature
     body = "Version=2010-05-08&Action=GetUser"
     return generate_aws4_signature(sk, region, "eim", amz_date, "POST", "/", {
         "host": f"eim.{region}.outscale.com",
         "x-amz-date": amz_date,
+        **(extra_headers or {}),
     }, payload=body)
 
+def signed_url_headers(url):
+    OKS_URL_HEADER = "x-oks-target"
+    """Return the extra signed headers binding a token to `url`, if any."""
+    return {OKS_URL_HEADER: url.strip()} if url else {}
+
 def encode_signed_requests_token(prefix, signed_requests):
     """Pack signed request data into a prefixed base64-encoded JSON token."""
     return prefix + base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
 
-def build_signed_requests_token_v1(ak, sk, region):
+def build_signed_requests_token_v1(ak, sk, region, url=None):
     """
     Sign ReadAccounts (OSC API) and HeadBucket (OOS) requests with ak/sk.
 
+    When `url` is set, it is signed into both requests as the X-OKS-Target header.
     Returns both signatures packed into an oks-sigv4-v1 token.
     """
     head_bucket_name = "s3-bucket-oks-eim-user-check"
+    extra_headers = signed_url_headers(url)
 
     amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
 
-    read_accounts_signature = generate_read_accounts_signature(sk, region, amz_date)
-    head_bucket_signature = generate_oos_head_bucket_signature(sk, region, amz_date, head_bucket_name)
+    read_accounts_signature = generate_read_accounts_signature(sk, region, amz_date, extra_headers)
+    head_bucket_signature = generate_oos_head_bucket_signature(sk, region, amz_date, head_bucket_name, extra_headers)
 
     return encode_signed_requests_token("oks-sigv4-v1.", {
         "AccessKey": ak,
@@ -246,10 +256,11 @@ def build_signed_requests_token_v1(ak, sk, region):
         },
     })
 
-def build_signed_requests_token_v2(ak, sk, region):
+def build_signed_requests_token_v2(ak, sk, region, url=None):
     """
     Sign an EIM GetUser request with ak/sk.
 
+    When `url` is set, it is signed into the request as the X-OKS-Target header.
     Returns the signature packed into an oks-sigv4-v2 token.
     """
     amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
@@ -258,7 +269,7 @@ def build_signed_requests_token_v2(ak, sk, region):
         "AccessKey": ak,
         "AmzDate": amz_date,
         "Region": region,
-        "Signature": generate_eim_get_user_signature(sk, region, amz_date),
+        "Signature": generate_eim_get_user_signature(sk, region, amz_date, signed_url_headers(url)),
     })
 
 SIGV4_TOKEN_BUILDERS = {
@@ -266,11 +277,11 @@ def build_signed_requests_token_v2(ak, sk, region):
     "v2": build_signed_requests_token_v2,
 }
 
-def build_signed_requests_token(ak, sk, region, version="v1"):
-    """Build a SigV4 signed requests token of the given version."""
+def build_signed_requests_token(ak, sk, region, version="v1", url=None):
+    """Build a SigV4 signed requests token of the given version, optionally bound to `url`."""
     if version not in SIGV4_TOKEN_BUILDERS:
         raise click.ClickException(f"Unsupported SigV4 token version: {version}")
-    return SIGV4_TOKEN_BUILDERS[version](ak, sk, region)
+    return SIGV4_TOKEN_BUILDERS[version](ak, sk, region, url)
 
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""

From bbacf2e024756a38428d8b43e427447ceb8b7e69 Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Thu, 15 Oct 2026 15:49:30 +0200
Subject: [PATCH 07/11] generate kubeconfig locally if url is provided

---
 oks_cli/cluster.py | 42 ++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 40 insertions(+), 2 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 99de4a9..59533f7 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -623,6 +623,34 @@ def cluster_delete_command(ctx, project_name, cluster_name, output, dry_run, for
             set_cluster_id("")
         print_output(data, output)
 
+def _build_token_kubeconfig(url, profile, cluster_name):
+    """Build a kubeconfig for `url` whose credentials come from `oks-cli cluster token`."""
+    token_env = [{"name": "OKS_PROFILE", "value": profile}]
+    if os.getenv("OKS_AWS_SIGV4"):
+        token_env.append({"name": "OKS_AWS_SIGV4", "value": os.getenv("OKS_AWS_SIGV4")})
+
+    cluster_name = cluster_name or "default"
+
+    return {
+        "apiVersion": "v1",
+        "kind": "Config",
+        "clusters": [{"name": cluster_name, "cluster": {"server": url}}],
+        "contexts": [{"name": cluster_name, "context": {"cluster": cluster_name, "user": "default"}}],
+        "current-context": cluster_name,
+        "users": [{
+            "name": "default",
+            "user": {
+                "exec": {
+                    "apiVersion": "client.authentication.k8s.io/v1",
+                    "command": "oks-cli",
+                    "args": ["cluster", "token", "--url", url],
+                    "env": token_env,
+                    "interactiveMode": "Never",
+                },
+            },
+        }],
+    }
+
 # GET KUBECONFIG
 @cluster.command('kubeconfig', help="Fetch the kubeconfig for a cluster")
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
@@ -635,13 +663,24 @@ def cluster_delete_command(ctx, project_name, cluster_name, output, dry_run, for
 @click.option('--user', type=click.STRING, help="User")
 @click.option('--group', type=click.STRING, help="Group")
 @click.option('--ttl', type=click.STRING, help="TTL in human readable format (5h, 1d, 1w)")
+@click.option('--url', help="Kubernetes API server URL, generate a kubeconfig authenticated by 'oks-cli cluster token' instead of fetching one")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.pass_context
-def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, output, wide, refresh, nacl, user, group, ttl, profile):
+def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, output, wide, refresh, nacl, user, group, ttl, url, profile):
     """CLI command to fetch and optionally print the kubeconfig for a specified cluster."""
     project_name, cluster_name, profile = ctx_update(ctx, project_name, cluster_name, profile)
     login_profile(profile)
 
+    if url:
+        if print_path or output == 'table':
+            raise click.UsageError("--url supports only yaml and json output")
+        kubeconfig = _build_token_kubeconfig(url, os.environ["OKS_PROFILE"], cluster_name)
+        if output == 'json':
+            click.echo(json.dumps(kubeconfig))
+        else:
+            click.echo(yaml.safe_dump(kubeconfig, sort_keys=False), nl=False)
+        return
+
     project_id = find_project_id_by_name(project_name)
     cluster_id = find_cluster_id_by_name(project_id, cluster_name)
 
@@ -801,7 +840,6 @@ def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=
 
         raise SystemExit(e.returncode)
 
-
 @cluster.command('kubectl', help='Fetch the kubeconfig for a cluster and run kubectl against it', context_settings={"ignore_unknown_options": True})
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)

From 1ed0fe8951c2af9d53260014ca59261f9917d059 Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Thu, 22 Oct 2026 10:13:30 +0200
Subject: [PATCH 08/11] cache token, and use v2 token only

---
 oks_cli/cluster.py  |  15 ++--
 oks_cli/main.py     |  22 +-----
 oks_cli/utils.py    | 168 +++++++++++++++++++++++++++-----------------
 tests/test_sigv4.py | 162 ++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 273 insertions(+), 94 deletions(-)
 create mode 100644 tests/test_sigv4.py

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 59533f7..0181055 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -6,7 +6,7 @@
 
 import time
 import os
-from datetime import datetime, timedelta, timezone
+from datetime import datetime
 import dateutil.parser
 import human_readable
 import pathlib
@@ -24,7 +24,7 @@
                    get_template, get_cluster_name, format_changed_row,          \
                    is_interesting_status, profile_completer, project_completer, \
                    kubeconfig_parse_fields, print_table, format_row, apply_set_fields, \
-                   build_signed_requests_token
+                   build_signed_requests_token, get_sigv4_token_expiration
 
 from .profile import add_profile
 from .project import project_create, project_login
@@ -625,10 +625,6 @@ def cluster_delete_command(ctx, project_name, cluster_name, output, dry_run, for
 
 def _build_token_kubeconfig(url, profile, cluster_name):
     """Build a kubeconfig for `url` whose credentials come from `oks-cli cluster token`."""
-    token_env = [{"name": "OKS_PROFILE", "value": profile}]
-    if os.getenv("OKS_AWS_SIGV4"):
-        token_env.append({"name": "OKS_AWS_SIGV4", "value": os.getenv("OKS_AWS_SIGV4")})
-
     cluster_name = cluster_name or "default"
 
     return {
@@ -644,7 +640,7 @@ def _build_token_kubeconfig(url, profile, cluster_name):
                     "apiVersion": "client.authentication.k8s.io/v1",
                     "command": "oks-cli",
                     "args": ["cluster", "token", "--url", url],
-                    "env": token_env,
+                    "env": [{"name": "OKS_PROFILE", "value": profile}],
                     "interactiveMode": "Never",
                 },
             },
@@ -776,9 +772,8 @@ def cluster_token_command(ctx, url, profile):
     if not region:
         raise click.ClickException("region_name is required in the profile to generate a token")
 
-    # SigV4 signatures are accepted for 15 minutes, expire the token slightly earlier
-    expiration = datetime.now(timezone.utc) + timedelta(minutes=14)
-    token = build_signed_requests_token(access_key, secret_key, region, os.getenv("OKS_AWS_SIGV4", "v1"), url)
+    token = build_signed_requests_token(access_key, secret_key, region, url)
+    expiration = get_sigv4_token_expiration(token)
 
     exec_credential = {
         "apiVersion": "client.authentication.k8s.io/v1",
diff --git a/oks_cli/main.py b/oks_cli/main.py
index 5687d2a..12dff4f 100755
--- a/oks_cli/main.py
+++ b/oks_cli/main.py
@@ -3,7 +3,6 @@
 import os
 import json
 import logging
-import re
 from .project import project
 from .cluster import cluster
 from .profile import profile
@@ -12,28 +11,16 @@
 from .netpeering import netpeering
 from .user import user
 
-from .utils import ctx_update, install_completions, profile_completer, cluster_completer, project_completer, \
-                   SIGV4_TOKEN_BUILDERS
-
-class CLIGroup(click.Group):
-    def parse_args(self, ctx, args):
-        # Click would consume the subcommand name as the value of a bare `--aws-sigv4`
-        for i, arg in enumerate(args):
-            if arg == "--aws-sigv4" and (i + 1 == len(args) or not re.fullmatch(r"v\d+", args[i + 1])):
-                args = args[:i + 1] + ["v1"] + args[i + 1:]
-                break
-        return super().parse_args(ctx, args)
+from .utils import ctx_update, install_completions, profile_completer, cluster_completer, project_completer
 
 # Main CLI entry point
-@click.group(cls=CLIGroup, invoke_without_command=True)
+@click.group(invoke_without_command=True)
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)
 @click.option('--verbose', '-v', count=True, help="Increase verbosity")
-@click.option('--aws-sigv4', type=click.Choice(list(SIGV4_TOKEN_BUILDERS)),
-              help="Authenticate with an AWS SigV4 signed token: v1 (ReadAccounts + HeadBucket, used when no version is given) or v2 (EIM GetUser)")
 @click.pass_context
-def cli(ctx, project_name, cluster_name, profile, verbose, aws_sigv4):
+def cli(ctx, project_name, cluster_name, profile, verbose):
     """
     CLI tool to manage projects and clusters.
 
@@ -42,9 +29,6 @@ def cli(ctx, project_name, cluster_name, profile, verbose, aws_sigv4):
     """
     ctx_update(ctx, project_name, cluster_name, profile)
 
-    if aws_sigv4:
-        os.environ["OKS_AWS_SIGV4"] = aws_sigv4
-
     LOGLEVEL = os.environ.get('LOGLEVEL', 'WARNING').upper()
 
     if verbose >= 1:
diff --git a/oks_cli/utils.py b/oks_cli/utils.py
index 676d400..92ef4b3 100644
--- a/oks_cli/utils.py
+++ b/oks_cli/utils.py
@@ -10,7 +10,7 @@
 import pathlib
 import traceback
 import time
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 import OpenSSL
 import shutil
 import prettytable
@@ -121,6 +121,7 @@ def do_request(method, path, *args, **kwargs):
                         data.reason, data.text[:50])
             data.raise_for_status()
             save_tokens(data.headers)
+            save_request_sigv4_token(kwargs['headers'])
             obj = find_response_object(data)
             return obj
         except requests.exceptions.HTTPError as err:
@@ -135,6 +136,10 @@ def do_request(method, path, *args, **kwargs):
             if jwt_response is not None:
                 return jwt_response
 
+            sigv4_response = handle_sigv4_error(err, method, path, args, kwargs)
+            if sigv4_response is not None:
+                return sigv4_response
+
             if attempt < retries and method.upper() == "GET":
                 logging.info("GET failed (attempt %s/%s), retrying in %s sec...", attempt, retries, backoff)
                 time.sleep(backoff)
@@ -191,24 +196,6 @@ def generate_aws4_signature(sk, region, service, amz_date, method, canonical_uri
     signing_key = get_signature_key(sk, date_stamp, region, service)
     return hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()
 
-def generate_read_accounts_signature(sk, region, amz_date, extra_headers=None):
-    """Generate AWS4 signature for an OUTSCALE API ReadAccounts request."""
-    return generate_aws4_signature(sk, region, "api", amz_date, "POST", "/api/v1/ReadAccounts", {
-        "content-type": "application/json",
-        "host": f"api.{region}.outscale.com",
-        "x-amz-date": amz_date,
-        **(extra_headers or {}),
-    })
-
-def generate_oos_head_bucket_signature(sk, region, amz_date, bucket, extra_headers=None):
-    """Generate AWS4 signature for an OOS (S3) HeadBucket request (path-style)."""
-    return generate_aws4_signature(sk, region, "s3", amz_date, "HEAD", f"/{bucket}", {
-        "host": f"oos.{region}.outscale.com",
-        "x-amz-content-sha256": hash_sha256(''),
-        "x-amz-date": amz_date,
-        **(extra_headers or {}),
-    })
-
 def generate_eim_get_user_signature(sk, region, amz_date, extra_headers=None):
     """Generate AWS4 signature for an OUTSCALE EIM GetUser request."""
     # The verifier must replay this exact body, its hash is part of the signature
@@ -228,35 +215,7 @@ def encode_signed_requests_token(prefix, signed_requests):
     """Pack signed request data into a prefixed base64-encoded JSON token."""
     return prefix + base64.b64encode(json.dumps(signed_requests).encode('utf-8')).decode('utf-8')
 
-def build_signed_requests_token_v1(ak, sk, region, url=None):
-    """
-    Sign ReadAccounts (OSC API) and HeadBucket (OOS) requests with ak/sk.
-
-    When `url` is set, it is signed into both requests as the X-OKS-Target header.
-    Returns both signatures packed into an oks-sigv4-v1 token.
-    """
-    head_bucket_name = "s3-bucket-oks-eim-user-check"
-    extra_headers = signed_url_headers(url)
-
-    amz_date = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
-
-    read_accounts_signature = generate_read_accounts_signature(sk, region, amz_date, extra_headers)
-    head_bucket_signature = generate_oos_head_bucket_signature(sk, region, amz_date, head_bucket_name, extra_headers)
-
-    return encode_signed_requests_token("oks-sigv4-v1.", {
-        "AccessKey": ak,
-        "AmzDate": amz_date,
-        "Region": region,
-        "ReadAccounts": {
-            "Signature": read_accounts_signature,
-        },
-        "HeadBucket": {
-            "Signature": head_bucket_signature,
-            "Bucket": head_bucket_name,
-        },
-    })
-
-def build_signed_requests_token_v2(ak, sk, region, url=None):
+def build_signed_requests_token(ak, sk, region, url=None):
     """
     Sign an EIM GetUser request with ak/sk.
 
@@ -272,17 +231,6 @@ def build_signed_requests_token_v2(ak, sk, region, url=None):
         "Signature": generate_eim_get_user_signature(sk, region, amz_date, signed_url_headers(url)),
     })
 
-SIGV4_TOKEN_BUILDERS = {
-    "v1": build_signed_requests_token_v1,
-    "v2": build_signed_requests_token_v2,
-}
-
-def build_signed_requests_token(ak, sk, region, version="v1", url=None):
-    """Build a SigV4 signed requests token of the given version, optionally bound to `url`."""
-    if version not in SIGV4_TOKEN_BUILDERS:
-        raise click.ClickException(f"Unsupported SigV4 token version: {version}")
-    return SIGV4_TOKEN_BUILDERS[version](ak, sk, region, url)
-
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""
     headers = {
@@ -299,20 +247,20 @@ def build_headers():
             if _h and _n:
                 headers[_h.strip()] = _n.strip() 
 
-    if os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY") and os.getenv("OKS_AWS_SIGV4"):
+    if os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY"):
         region = os.getenv("OKS_REGION")
         if not region:
             raise click.ClickException("region_name is required in the profile to sign requests with ak/sk")
 
-        token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region,
-                                            os.getenv("OKS_AWS_SIGV4"))
+        token = get_sigv4_token(os.getenv("OKS_ACCESS_KEY"), region)
+        if token:
+            logging.info("using cached sigv4 token")
+        else:
+            token = build_signed_requests_token(os.getenv("OKS_ACCESS_KEY"), os.getenv("OKS_SECRET_KEY"), region)
         headers["Authorization"] = f"Bearer {token}"
     elif is_jwt_enabled() and is_tokens_valid():
         headers["AccessToken"] = get_token('access_token')
         headers["RefreshToken"] = get_token('refresh_token')
-    elif os.getenv("OKS_ACCESS_KEY") and os.getenv("OKS_SECRET_KEY"):
-        headers["AccessKey"] = os.getenv("OKS_ACCESS_KEY")
-        headers["SecretKey"] = os.getenv("OKS_SECRET_KEY")
     elif os.getenv("OKS_USERNAME") and os.getenv("OKS_PASSWORD"):
         username = os.getenv("OKS_USERNAME")
         password = os.getenv("OKS_PASSWORD")
@@ -404,6 +352,20 @@ def handle_jwt_error(err, method, path, args, kwargs):
     except Exception:
         return None
 
+def handle_sigv4_error(err, method, path, args, kwargs):
+    """Handle a rejected cached SigV4 token by removing it and retrying with a fresh one."""
+    try:
+        response_body = json.loads(err.response.text)
+        # fresh tokens are cached only after a successful response, so this retries once
+        if response_body.get("message") == "Unauthorized: Invalid SigV4 token." \
+                and remove_sigv4_token():
+            kwargs.get("headers", {}).pop("Authorization", None)
+            return do_request(method, path, *args, **kwargs)
+    except JSONClickException:
+        raise
+    except Exception:
+        return None
+
 
 # TODO: Introduce `find_project_by_name` into `find_project_by_id` method for merging
 def find_project_by_name(project_name):
@@ -898,6 +860,82 @@ def remove_jwt_token(token_type):
     if os.path.exists(TOKEN_FILE):
         os.remove(TOKEN_FILE)
 
+# OUTSCALE accepts a SigV4 signature for 15 minutes, stop using the token slightly earlier
+SIGV4_TOKEN_LIFETIME = timedelta(minutes=14)
+
+def get_sigv4_token_file():
+    """Return the cache file of the SigV4 token for the current profile."""
+    CONFIG_FOLDER, _ = get_config_path()
+
+    if not os.getenv("OKS_PROFILE"):
+        return None
+
+    return f"{CONFIG_FOLDER}/{os.getenv('OKS_PROFILE')}.sigv4_token"
+
+def decode_signed_requests_token(token):
+    """Decode the JSON payload of an oks-sigv4 token."""
+    try:
+        return json.loads(base64.b64decode(token.partition('.')[2]))
+    except ValueError:
+        return None
+
+def get_sigv4_token_expiration(token):
+    """Return when the SigV4 token stops being usable, based on its AmzDate."""
+    payload = decode_signed_requests_token(token)
+    if not isinstance(payload, dict) or not isinstance(payload.get("AmzDate"), str):
+        return None
+    try:
+        signed_at = datetime.strptime(payload["AmzDate"], '%Y%m%dT%H%M%SZ').replace(tzinfo=timezone.utc)
+    except ValueError:
+        return None
+    return signed_at + SIGV4_TOKEN_LIFETIME
+
+def get_sigv4_token(ak, region):
+    """Return the cached SigV4 token if it was signed with `ak` for `region` and is still valid."""
+    TOKEN_FILE = get_sigv4_token_file()
+
+    if not TOKEN_FILE or not os.path.exists(TOKEN_FILE):
+        return None
+
+    with open(TOKEN_FILE, 'r') as file:
+        token = file.read()
+
+    payload = decode_signed_requests_token(token)
+    if not isinstance(payload, dict) or payload.get("AccessKey") != ak or payload.get("Region") != region:
+        return None
+
+    expiration = get_sigv4_token_expiration(token)
+    if not expiration or expiration < datetime.now(timezone.utc) + timedelta(seconds=5): # 5sec margin
+        return None
+
+    return token
+
+def save_sigv4_token(token):
+    """Save the SigV4 token for the current profile."""
+    TOKEN_FILE = get_sigv4_token_file()
+
+    if not TOKEN_FILE:
+        return
+
+    with os.fdopen(os.open(TOKEN_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), 'w') as file:
+        file.write(token)
+
+def remove_sigv4_token():
+    """Delete the cached SigV4 token, return True if there was one."""
+    TOKEN_FILE = get_sigv4_token_file()
+
+    if TOKEN_FILE and os.path.exists(TOKEN_FILE):
+        os.remove(TOKEN_FILE)
+        return True
+    return False
+
+def save_request_sigv4_token(headers):
+    """Cache the SigV4 token sent in request headers, once the API accepted it."""
+    authorization = headers.get("Authorization", "")
+    prefix = "Bearer oks-sigv4-"
+    if authorization.startswith(prefix):
+        save_sigv4_token(authorization[len("Bearer "):])
+
 def detect_and_parse_input(input_data):
     """Parse input as JSON or YAML; raise error if invalid."""
     try:
diff --git a/tests/test_sigv4.py b/tests/test_sigv4.py
new file mode 100644
index 0000000..d16b6fc
--- /dev/null
+++ b/tests/test_sigv4.py
@@ -0,0 +1,162 @@
+from click.testing import CliRunner
+from oks_cli.main import cli
+from oks_cli.utils import build_signed_requests_token, encode_signed_requests_token, \
+                          decode_signed_requests_token, get_sigv4_token_file, save_sigv4_token
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timedelta, timezone
+
+import json
+import os
+import pytest
+import requests
+
+
+CLUSTER_LIST = ["--profile", "default", "cluster", "list", "-p", "test", "-c", "test", "-o", "json"]
+
+
+@pytest.fixture(autouse=True)
+def clean_sigv4_env(monkeypatch):
+    # the CLI exports OKS_PROFILE to os.environ, make sure it does not leak between tests
+    monkeypatch.delenv("OKS_PROFILE", raising=False)
+    monkeypatch.setenv("OKS_RETRIES", "1")
+
+
+def run_token():
+    result = CliRunner().invoke(cli, ["--profile", "default", "cluster", "token"])
+    assert result.exit_code == 0, result.output
+    return json.loads(result.output)
+
+
+def clusters_responses():
+    return [
+        MagicMock(status_code=200, headers={}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers={}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345", "name": "test"}]}),
+    ]
+
+
+def rejected_response():
+    response = MagicMock(status_code=401, headers={}, text='{"message": "Unauthorized: Invalid SigV4 token."}')
+    response.raise_for_status.side_effect = requests.exceptions.HTTPError(response=response)
+    return response
+
+
+def record_authorizations(mock_request, responses):
+    """Answer with `responses` and record the Authorization header at the time of each call."""
+    sent = []
+    responses = iter(responses)
+
+    def request(method, url, *args, **kwargs):
+        # do_request retries reuse and mutate the same headers dict
+        sent.append(kwargs["headers"]["Authorization"])
+        return next(responses)
+
+    mock_request.side_effect = request
+    return sent
+
+
+# Test "cluster token": the token is not cached and expires at AmzDate + 14 minutes
+def test_cluster_token_is_not_cached(add_default_profile, monkeypatch):
+    monkeypatch.setenv("OKS_PROFILE", "default")
+    stale = build_signed_requests_token("AK", "SK", "eu-west-2")
+    save_sigv4_token(stale)
+
+    with patch("oks_cli.cluster.build_signed_requests_token", wraps=build_signed_requests_token) as build:
+        result = run_token()
+    build.assert_called_once()
+
+    token = result["status"]["token"]
+    assert token.startswith("oks-sigv4-v2.")
+    amz_date = datetime.strptime(decode_signed_requests_token(token)["AmzDate"], '%Y%m%dT%H%M%SZ')
+    assert result["status"]["expirationTimestamp"] == (amz_date + timedelta(minutes=14)).strftime('%Y-%m-%dT%H:%M:%SZ')
+
+    with open(get_sigv4_token_file()) as file:
+        assert file.read() == stale
+
+
+# Test API requests: an ak/sk profile sends a SigV4 token, never the raw ak/sk
+@patch("oks_cli.utils.requests.request")
+def test_api_request_uses_sigv4_for_aksk(mock_request, add_default_profile):
+    mock_request.side_effect = clusters_responses()
+
+    result = CliRunner().invoke(cli, CLUSTER_LIST)
+    assert result.exit_code == 0, result.output
+
+    headers = mock_request.call_args.kwargs["headers"]
+    assert headers["Authorization"].startswith("Bearer oks-sigv4-v2.")
+    assert "AccessKey" not in headers
+    assert "SecretKey" not in headers
+
+
+# Test API requests: an expired or foreign cached token is not used
+@pytest.mark.parametrize("access_key, amz_date", [
+    ("AK", "20200101T000000Z"),
+    ("OTHER", datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')),
+])
+@patch("oks_cli.utils.requests.request")
+def test_api_request_ignores_invalid_cached_token(mock_request, add_default_profile, monkeypatch, access_key, amz_date):
+    monkeypatch.setenv("OKS_PROFILE", "default")
+    stale = encode_signed_requests_token("oks-sigv4-v2.", {
+        "AccessKey": access_key, "AmzDate": amz_date, "Region": "eu-west-2", "Signature": "ab",
+    })
+    save_sigv4_token(stale)
+    authorizations = record_authorizations(mock_request, clusters_responses())
+
+    result = CliRunner().invoke(cli, CLUSTER_LIST)
+    assert result.exit_code == 0, result.output
+
+    assert f"Bearer {stale}" not in authorizations
+    token = authorizations[0][len("Bearer "):]
+    assert decode_signed_requests_token(token)["AccessKey"] == "AK"
+    with open(get_sigv4_token_file()) as file:
+        assert file.read() == token
+
+
+# Test API requests: the token is cached only after the API accepted it
+@patch("oks_cli.utils.requests.request")
+def test_api_request_caches_accepted_token(mock_request, add_default_profile):
+    authorizations = record_authorizations(mock_request, clusters_responses() + clusters_responses())
+
+    for _ in range(2):
+        result = CliRunner().invoke(cli, CLUSTER_LIST)
+        assert result.exit_code == 0, result.output
+
+    assert len(authorizations) == 4
+    assert authorizations[0].startswith("Bearer oks-sigv4-v2.")
+    assert len(set(authorizations)) == 1
+
+    os.environ["OKS_PROFILE"] = "default"
+    assert oct(os.stat(get_sigv4_token_file()).st_mode & 0o777) == "0o600"
+
+
+# Test API requests: a rejected cached token is dropped and the request retried once with a fresh one
+@patch("oks_cli.utils.requests.request")
+def test_api_request_retries_rejected_cached_token(mock_request, add_default_profile, monkeypatch):
+    monkeypatch.setenv("OKS_PROFILE", "default")
+    cached = build_signed_requests_token("AK", "SK", "eu-west-2")
+    save_sigv4_token(cached)
+    fresh = encode_signed_requests_token("oks-sigv4-v2.", {
+        "AccessKey": "AK", "AmzDate": datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ'),
+        "Region": "eu-west-2", "Signature": "cd",
+    })
+    authorizations = record_authorizations(mock_request, [rejected_response()] + clusters_responses())
+
+    with patch("oks_cli.utils.build_signed_requests_token", return_value=fresh):
+        result = CliRunner().invoke(cli, CLUSTER_LIST)
+    assert result.exit_code == 0, result.output
+
+    assert authorizations == [f"Bearer {cached}", f"Bearer {fresh}", f"Bearer {fresh}"]
+    with open(get_sigv4_token_file()) as file:
+        assert file.read() == fresh
+
+
+# Test API requests: a rejected fresh token is not cached and not retried
+@patch("oks_cli.utils.requests.request")
+def test_api_request_rejected_fresh_token(mock_request, add_default_profile):
+    authorizations = record_authorizations(mock_request, [rejected_response()])
+
+    result = CliRunner().invoke(cli, CLUSTER_LIST)
+    assert result.exit_code != 0
+    assert len(authorizations) == 1
+
+    os.environ["OKS_PROFILE"] = "default"
+    assert not os.path.exists(get_sigv4_token_file())

From 54650a407459f1206a85d6f465f2fbde6637c23d Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Thu, 29 Oct 2026 14:55:20 +0100
Subject: [PATCH 09/11] build kubeconfig with oks-cli cluster token auth

---
 oks_cli/cluster.py | 29 +++++++++++++++++++++++++----
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 0181055..1ccff31 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -623,14 +623,32 @@ def cluster_delete_command(ctx, project_name, cluster_name, output, dry_run, for
             set_cluster_id("")
         print_output(data, output)
 
-def _build_token_kubeconfig(url, profile, cluster_name):
+def _fetch_cluster_endpoint(cluster_id):
+    """Return the server URL and certificate-authority-data of the cluster, read from a kubeconfig issued for 1 second."""
+    kubeconfig_raw = do_request("GET", f'clusters/{cluster_id}/kubeconfig', params={"ttl": "1s"})['data']['kubeconfig']
+    if not kubeconfig_raw:
+        raise click.ClickException("Empty kubeconfig received, could not get the cluster endpoint")
+
+    kubeconfig = yaml.safe_load(kubeconfig_raw)
+    clusters = kubeconfig.get("clusters") or []
+    context_cluster = next((ctx["context"].get("cluster") for ctx in kubeconfig.get("contexts") or []
+                            if ctx.get("name") == kubeconfig.get("current-context")), None)
+    entry = next((c for c in clusters if c.get("name") == context_cluster), clusters[0] if clusters else {})
+
+    server = entry.get("cluster", {}).get("server")
+    ca_data = entry.get("cluster", {}).get("certificate-authority-data")
+    if not server or not ca_data:
+        raise click.ClickException("server or certificate-authority-data not found in the cluster kubeconfig")
+    return server, ca_data
+
+def _build_token_kubeconfig(url, profile, cluster_name, ca_data):
     """Build a kubeconfig for `url` whose credentials come from `oks-cli cluster token`."""
     cluster_name = cluster_name or "default"
 
     return {
         "apiVersion": "v1",
         "kind": "Config",
-        "clusters": [{"name": cluster_name, "cluster": {"server": url}}],
+        "clusters": [{"name": cluster_name, "cluster": {"server": url, "certificate-authority-data": ca_data}}],
         "contexts": [{"name": cluster_name, "context": {"cluster": cluster_name, "user": "default"}}],
         "current-context": cluster_name,
         "users": [{
@@ -659,7 +677,7 @@ def _build_token_kubeconfig(url, profile, cluster_name):
 @click.option('--user', type=click.STRING, help="User")
 @click.option('--group', type=click.STRING, help="Group")
 @click.option('--ttl', type=click.STRING, help="TTL in human readable format (5h, 1d, 1w)")
-@click.option('--url', help="Kubernetes API server URL, generate a kubeconfig authenticated by 'oks-cli cluster token' instead of fetching one")
+@click.option('--url', is_flag=True, help="Generate a kubeconfig authenticated by 'oks-cli cluster token' for the cluster API server URL")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.pass_context
 def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, output, wide, refresh, nacl, user, group, ttl, url, profile):
@@ -670,7 +688,10 @@ def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, outp
     if url:
         if print_path or output == 'table':
             raise click.UsageError("--url supports only yaml and json output")
-        kubeconfig = _build_token_kubeconfig(url, os.environ["OKS_PROFILE"], cluster_name)
+        project_id = find_project_id_by_name(project_name)
+        cluster_id = find_cluster_id_by_name(project_id, cluster_name)
+        server, ca_data = _fetch_cluster_endpoint(cluster_id)
+        kubeconfig = _build_token_kubeconfig(server, os.environ["OKS_PROFILE"], cluster_name, ca_data)
         if output == 'json':
             click.echo(json.dumps(kubeconfig))
         else:

From e6e857738370853a7a7ec75cdd7bf0c2c093e95c Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Fri, 30 Oct 2026 15:57:35 +0100
Subject: [PATCH 10/11] introduce --auth flag for kubeconfig and kubectl
 commands

---
 oks_cli/cluster.py | 181 ++++++++++++++++++++++++++-------------------
 1 file changed, 106 insertions(+), 75 deletions(-)

diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index 1ccff31..d35ae8c 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -665,6 +665,61 @@ def _build_token_kubeconfig(url, profile, cluster_name, ca_data):
         }],
     }
 
+def _is_token_kubeconfig_valid(kubeconfig_str, profile):
+    """Check if the cached token kubeconfig runs `oks-cli cluster token` with `profile`."""
+    try:
+        env = yaml.safe_load(kubeconfig_str)["users"][0]["user"]["exec"]["env"]
+    except (yaml.YAMLError, KeyError, IndexError, TypeError):
+        return False
+    return {"name": "OKS_PROFILE", "value": profile} in env
+
+def _get_token_kubeconfig(project_id, cluster_id, cluster_name, refresh=False):
+    """Return the path and content of the cached token kubeconfig, generating it when missing, stale or on `refresh`."""
+    profile = os.environ["OKS_PROFILE"]
+    kubeconfig_path = get_cache(project_id, cluster_id, 'kubeconfig-token', None, None)
+
+    if kubeconfig_path and not refresh:
+        with open(kubeconfig_path) as f:
+            kubeconfig = f.read()
+        if _is_token_kubeconfig_valid(kubeconfig, profile):
+            return kubeconfig_path, kubeconfig
+
+    logging.info("generating token kubeconfig")
+    server, ca_data = _fetch_cluster_endpoint(cluster_id)
+    kubeconfig = yaml.safe_dump(_build_token_kubeconfig(server, profile, cluster_name, ca_data), sort_keys=False)
+    return save_cache(project_id, cluster_id, 'kubeconfig-token', kubeconfig, None, None), kubeconfig
+
+def _fetch_cert_kubeconfig(cluster_id, user, group, ttl, nacl):
+    """Fetch a kubeconfig authenticated by a client certificate from the API."""
+    logging.info("extracting kubeconfig by api")
+
+    params = {}
+    if user:
+        params["user"] = user
+    if group:
+        params["group"] = group
+    if ttl:
+        params["ttl"] = ttl
+
+    if nacl:
+        ephemeral = PrivateKey.generate()
+        unsealbox = SealedBox(ephemeral)
+
+        headers = {
+          'x-encrypt-nacl': ephemeral.public_key.encode(Base64Encoder).decode('ascii')
+        }
+        kubeconfig_raw = do_request("POST", f'clusters/{cluster_id}/kubeconfig', params = params, headers = headers)['data']['kubeconfig']
+    else:
+        kubeconfig_raw = do_request("GET", f'clusters/{cluster_id}/kubeconfig', params = params)['data']['kubeconfig']
+
+    if not kubeconfig_raw:
+        logging.error("empty response")
+        raise SystemExit()
+    elif nacl:
+        logging.info("decrypting received kubeconfig")
+        return unsealbox.decrypt(kubeconfig_raw.encode('ascii'), encoder = Base64Encoder).decode('ascii')
+    return kubeconfig_raw
+
 # GET KUBECONFIG
 @cluster.command('kubeconfig', help="Fetch the kubeconfig for a cluster")
 @click.option('--project-name', '-p', required=False, help="Project Name", shell_complete=project_completer)
@@ -677,75 +732,43 @@ def _build_token_kubeconfig(url, profile, cluster_name, ca_data):
 @click.option('--user', type=click.STRING, help="User")
 @click.option('--group', type=click.STRING, help="Group")
 @click.option('--ttl', type=click.STRING, help="TTL in human readable format (5h, 1d, 1w)")
-@click.option('--url', is_flag=True, help="Generate a kubeconfig authenticated by 'oks-cli cluster token' for the cluster API server URL")
+@click.option('--auth', type=click.Choice(["cert", "token"]), default="cert",
+              help="Authentication of the kubeconfig: cert (client certificate, default) or token ('oks-cli cluster token' exec plugin)")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.pass_context
-def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, output, wide, refresh, nacl, user, group, ttl, url, profile):
+def cluster_kubeconfig_command(ctx, project_name, cluster_name, print_path, output, wide, refresh, nacl, user, group, ttl, auth, profile):
     """CLI command to fetch and optionally print the kubeconfig for a specified cluster."""
     project_name, cluster_name, profile = ctx_update(ctx, project_name, cluster_name, profile)
     login_profile(profile)
 
-    if url:
-        if print_path or output == 'table':
-            raise click.UsageError("--url supports only yaml and json output")
-        project_id = find_project_id_by_name(project_name)
-        cluster_id = find_cluster_id_by_name(project_id, cluster_name)
-        server, ca_data = _fetch_cluster_endpoint(cluster_id)
-        kubeconfig = _build_token_kubeconfig(server, os.environ["OKS_PROFILE"], cluster_name, ca_data)
-        if output == 'json':
-            click.echo(json.dumps(kubeconfig))
-        else:
-            click.echo(yaml.safe_dump(kubeconfig, sort_keys=False), nl=False)
-        return
+    if auth == "token":
+        if output == 'table':
+            raise click.UsageError("--auth token supports only yaml and json output")
+        if user or group or ttl or nacl:
+            raise click.UsageError("--user, --group, --ttl and --nacl are only supported with --auth cert")
 
     project_id = find_project_id_by_name(project_name)
     cluster_id = find_cluster_id_by_name(project_id, cluster_name)
 
-    # @TODO: check expiration in get_cache() code, etc
-    kubeconfig_path = get_cache(project_id, cluster_id, 'kubeconfig', user, group)
-
-    kubeconfig = None
-    is_cert_valid = False
-
-    if kubeconfig_path:
-        with open(kubeconfig_path) as f:
-            kubeconfig = f.read()
-
-    if kubeconfig:
-        is_cert_valid = verify_certificate(kubeconfig)
-
-    if not kubeconfig_path or refresh or not is_cert_valid:
-        logging.info("extracting kubeconfig by api")
+    if auth == "token":
+        kubeconfig_path, kubeconfig = _get_token_kubeconfig(project_id, cluster_id, cluster_name, refresh)
+    else:
+        # @TODO: check expiration in get_cache() code, etc
+        kubeconfig_path = get_cache(project_id, cluster_id, 'kubeconfig', user, group)
 
-        params = {}
-        if user:
-            params["user"] = user
-        if group:
-            params["group"] = group
-        if ttl:
-            params["ttl"] = ttl
+        kubeconfig = None
+        is_cert_valid = False
 
-        if nacl:
-            ephemeral = PrivateKey.generate()
-            unsealbox = SealedBox(ephemeral)
+        if kubeconfig_path:
+            with open(kubeconfig_path) as f:
+                kubeconfig = f.read()
 
-            headers = {
-              'x-encrypt-nacl': ephemeral.public_key.encode(Base64Encoder).decode('ascii')
-            }
-            kubeconfig_raw = do_request("POST", f'clusters/{cluster_id}/kubeconfig', params = params, headers = headers)['data']['kubeconfig']
-        else:
-            kubeconfig_raw = do_request("GET", f'clusters/{cluster_id}/kubeconfig', params = params)['data']['kubeconfig']
-
-        if not kubeconfig_raw:
-            logging.error("empty response")
-            raise SystemExit()
-        elif nacl:
-            logging.info("decrypting received kubeconfig")
-            kubeconfig = unsealbox.decrypt(kubeconfig_raw.encode('ascii'), encoder = Base64Encoder).decode('ascii')
-        else:
-            kubeconfig = kubeconfig_raw
+        if kubeconfig:
+            is_cert_valid = verify_certificate(kubeconfig)
 
-        kubeconfig_path = save_cache(project_id, cluster_id, 'kubeconfig', kubeconfig, user, group)
+        if not kubeconfig_path or refresh or not is_cert_valid:
+            kubeconfig = _fetch_cert_kubeconfig(cluster_id, user, group, ttl, nacl)
+            kubeconfig_path = save_cache(project_id, cluster_id, 'kubeconfig', kubeconfig, user, group)
 
     if print_path:
         click.echo(kubeconfig_path)
@@ -809,32 +832,35 @@ def cluster_token_command(ctx, url, profile):
     click.echo(json.dumps(exec_credential))
 
 
-def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=False):
+def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=False, auth="cert", cluster_name=None):
     """Run a kubectl command using the cached kubeconfig for the specified cluster, refreshing it if needed."""
-    # @TODO: check expiration in get_cache() code, etc
-    kubeconfig_path = get_cache(project_id, cluster_id, 'kubeconfig', user, group)
+    if auth == "token":
+        kubeconfig_path, _ = _get_token_kubeconfig(project_id, cluster_id, cluster_name)
+    else:
+        # @TODO: check expiration in get_cache() code, etc
+        kubeconfig_path = get_cache(project_id, cluster_id, 'kubeconfig', user, group)
 
-    kubeconfig = None
-    is_cert_valid = False
+        kubeconfig = None
+        is_cert_valid = False
 
-    if kubeconfig_path :
-        with open(kubeconfig_path) as f:
-            kubeconfig = f.read()
+        if kubeconfig_path :
+            with open(kubeconfig_path) as f:
+                kubeconfig = f.read()
 
-    if kubeconfig:
-        is_cert_valid = verify_certificate(kubeconfig)
+        if kubeconfig:
+            is_cert_valid = verify_certificate(kubeconfig)
 
-    if not kubeconfig_path or not is_cert_valid:
-        logging.info("extracting kubeconfig by api")
+        if not kubeconfig_path or not is_cert_valid:
+            logging.info("extracting kubeconfig by api")
 
-        kubeconfig_raw = do_request(
-            "GET", f'clusters/{cluster_id}/kubeconfig')['data']['kubeconfig']
+            kubeconfig_raw = do_request(
+                "GET", f'clusters/{cluster_id}/kubeconfig')['data']['kubeconfig']
 
-        if not kubeconfig_raw:
-            click.echo("Cannot get kubeconfig")
-            raise SystemExit(1)
+            if not kubeconfig_raw:
+                click.echo("Cannot get kubeconfig")
+                raise SystemExit(1)
 
-        kubeconfig_path = save_cache(project_id, cluster_id, 'kubeconfig', kubeconfig_raw, user, group)
+            kubeconfig_path = save_cache(project_id, cluster_id, 'kubeconfig', kubeconfig_raw, user, group)
 
     env = dict(os.environ)
     env['KUBECONFIG'] = str(kubeconfig_path)
@@ -861,18 +887,23 @@ def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)
 @click.option('--user', type=click.STRING, help="User")
 @click.option('--group', type=click.STRING, help="Group")
+@click.option('--auth', type=click.Choice(["cert", "token"]), default="cert",
+              help="Authentication of the kubeconfig: cert (client certificate, default) or token ('oks-cli cluster token' exec plugin)")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.argument("args", nargs=-1, type=click.UNPROCESSED)
 @click.pass_context
-def cluster_kubectl_command(ctx, project_name, cluster_name, user, group, args, profile):
+def cluster_kubectl_command(ctx, project_name, cluster_name, user, group, auth, args, profile):
     """CLI command to run kubectl against a specified cluster using its kubeconfig."""
     project_name, cluster_name, profile = ctx_update(ctx, project_name, cluster_name, profile)
     login_profile(profile)
 
+    if auth == "token" and (user or group):
+        raise click.UsageError("--user and --group are only supported with --auth cert")
+
     project_id = find_project_id_by_name(project_name)
     cluster_id = find_cluster_id_by_name(project_id, cluster_name)
 
-    _run_kubectl(project_id, cluster_id, user, group, args)
+    _run_kubectl(project_id, cluster_id, user, group, args, auth=auth, cluster_name=cluster_name)
 
 
 @click.group(help="Nodepool related commands.")

From d7d137f7b95e4941cfaf31808e2f0fcdc22f513e Mon Sep 17 00:00:00 2001
From: Yurii Kasper <yurii.kasper@outscale.com>
Date: Mon, 2 Nov 2026 10:42:06 +0100
Subject: [PATCH 11/11] token kubeconfig for nodepool and cache kubeconfigs
 commands

---
 oks_cli/cache.py       | 11 ++++--
 oks_cli/cluster.py     | 17 ++++++--
 tests/test_cache.py    | 17 +++++++-
 tests/test_cluster.py  | 90 ++++++++++++++++++++++++++++++++++++++++++
 tests/test_nodepool.py | 32 +++++++++++++++
 5 files changed, 158 insertions(+), 9 deletions(-)

diff --git a/oks_cli/cache.py b/oks_cli/cache.py
index 1de773d..8448a2f 100644
--- a/oks_cli/cache.py
+++ b/oks_cli/cache.py
@@ -1,5 +1,5 @@
 import click
-from .utils import clear_cache, find_project_id_by_name, find_cluster_id_by_name, get_all_cache, get_expiration_date, \
+from .utils import clear_cache, find_project_id_by_name, find_cluster_id_by_name, get_all_cache, get_cache, get_expiration_date, \
                    ctx_update, login_profile, profile_completer, cluster_completer, project_completer, print_table
 
 from prettytable import TableStyle
@@ -39,7 +39,7 @@ def list_kubeconfigs(ctx, project_name, cluster_name, plain, msword, profile):
     result = get_all_cache(project_id, cluster_id, "kubeconfig")
 
     data = []
-    fields = [["user", "user"],["group", "group"], ["expiration date", "expires_at"]]
+    fields = [["user", "user"],["group", "group"], ["auth", "auth"], ["expiration date", "expires_at"]]
 
     for element in result:
         kubeconfig = None
@@ -52,8 +52,11 @@ def list_kubeconfigs(ctx, project_name, cluster_name, plain, msword, profile):
 
         if kubeconfig:
             exp = get_expiration_date(kubeconfig)
-            row = user, group, exp
-            data.append({"user": user, "group": group, "expires_at": exp})
+            data.append({"user": user, "group": group, "auth": "cert", "expires_at": exp})
+
+    # the token kubeconfig runs `oks-cli cluster token`, it has no certificate user/group and does not expire
+    if get_cache(project_id, cluster_id, "kubeconfig-token", None, None):
+        data.append({"user": "-", "group": "-", "auth": "token", "expires_at": "never"})
 
     style = None
     if plain:
diff --git a/oks_cli/cluster.py b/oks_cli/cluster.py
index d35ae8c..cd2e7c1 100644
--- a/oks_cli/cluster.py
+++ b/oks_cli/cluster.py
@@ -911,18 +911,25 @@ def cluster_kubectl_command(ctx, project_name, cluster_name, user, group, auth,
 @click.option('--cluster-name', '--name', '-c', required=False, help="Cluster Name", shell_complete=cluster_completer)
 @click.option('--user', type=click.STRING, help="User")
 @click.option('--group', type=click.STRING, help="Group")
+@click.option('--auth', type=click.Choice(["cert", "token"]), default="cert",
+              help="Authentication of the kubeconfig: cert (client certificate, default) or token ('oks-cli cluster token' exec plugin)")
 @click.option('--profile', help="Configuration profile to use", shell_complete=profile_completer)
 @click.pass_context
-def nodepool(ctx, project_name, cluster_name, user, group, profile):
+def nodepool(ctx, project_name, cluster_name, user, group, auth, profile):
     """CLI group for nodepool-related commands, managing project, cluster, user, and profile context."""
     project_name, cluster_name, profile = ctx_update(ctx, project_name, cluster_name, profile)
     login_profile(profile)
 
+    if auth == "token" and (user or group):
+        raise click.UsageError("--user and --group are only supported with --auth cert")
+
     ctx.obj['project_id'] = find_project_id_by_name(project_name)
     ctx.obj['cluster_id'] = find_cluster_id_by_name(
         ctx.obj['project_id'], cluster_name)
+    ctx.obj['cluster_name'] = cluster_name
     ctx.obj['user'] = user
     ctx.obj['group'] = group
+    ctx.obj['auth'] = auth
 
 
 cluster.add_command(nodepool)
@@ -933,7 +940,7 @@ def nodepool(ctx, project_name, cluster_name, user, group, profile):
 def nodepool_list(ctx):
     """List nodepools in the specified cluster using kubectl."""
     _run_kubectl(ctx.obj['project_id'], ctx.obj['cluster_id'], ctx.obj['user'], ctx.obj['group'], [
-                 'get', 'nodepool', '-o', 'wide'])
+                 'get', 'nodepool', '-o', 'wide'], auth=ctx.obj['auth'], cluster_name=ctx.obj['cluster_name'])
 
 
 @nodepool.command('create', help="Create a new nodepool")
@@ -966,7 +973,7 @@ def setup_worker_pool(ctx, nodepool_name, count, vmtype, zone, output, dry_run,
         print_output(nodepool, output)
     else:
         _run_kubectl(ctx.obj['project_id'], ctx.obj['cluster_id'], ctx.obj['user'], ctx.obj['group'], [
-                 'create', '-f', '-'], input=json.dumps(nodepool))
+                 'create', '-f', '-'], input=json.dumps(nodepool), auth=ctx.obj['auth'], cluster_name=ctx.obj['cluster_name'])
 
 
 @nodepool.command('delete')
@@ -987,5 +994,7 @@ def delete_worker_pool(ctx, nodepool_name, force):
         ctx.obj['cluster_id'],
         ctx.obj['user'],
         ctx.obj['group'],
-        ['delete', 'nodepool', nodepool_name]
+        ['delete', 'nodepool', nodepool_name],
+        auth=ctx.obj['auth'],
+        cluster_name=ctx.obj['cluster_name'],
     )
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 6af023f..f938171 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -1,5 +1,6 @@
 from click.testing import CliRunner
 from oks_cli.main import cli
+from oks_cli.utils import save_cache
 from unittest.mock import patch, MagicMock
 
 def test_cache_clear_command():
@@ -17,4 +18,18 @@ def test_cache_kubeconfigs_command(mock_request, add_default_profile):
     runner = CliRunner()
     result = runner.invoke(cli, ["--profile", "default", "cache", "kubeconfigs", "-p", "test", "-c", "test"])
     assert result.exit_code == 0
-    assert '| user | group | expiration date |' in result.output
\ No newline at end of file
+    assert '| user | group | auth | expiration date |' in result.output
+
+# Test "cache kubeconfigs": the cached token kubeconfig is listed as a non expiring token entry
+@patch("oks_cli.utils.requests.request")
+def test_cache_kubeconfigs_lists_token_kubeconfig(mock_request, add_default_profile):
+    mock_request.side_effect = [
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345"}]})
+    ]
+    save_cache("12345", "12345", "kubeconfig-token", "kubeconfig", None, None)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["--profile", "default", "cache", "kubeconfigs", "-p", "test", "-c", "test"])
+    assert result.exit_code == 0, result.output
+    assert '| -    | -     | token | never           |' in result.output
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 249c00d..f35260b 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -350,6 +350,69 @@ def test_cluster_kubeconfig_command(mock_request, add_default_profile):
     assert result.exit_code == 0
     assert 'kubeconfig' in result.output
 
+# Test the "cluster kubeconfig --auth token" command: server and CA come from a 1 second kubeconfig, credentials from "cluster token"
+@patch("oks_cli.utils.requests.request")
+def test_cluster_kubeconfig_auth_token_command(mock_request, add_default_profile):
+    issued = yaml.safe_dump({
+        "clusters": [{"name": "test", "cluster": {"server": "https://kube.example:6443", "certificate-authority-data": "Q0E="}}],
+        "contexts": [{"name": "test", "context": {"cluster": "test", "user": "admin"}}],
+        "current-context": "test",
+    })
+    mock_request.side_effect = [
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Cluster":  {"data": {"kubeconfig": issued}}})
+    ]
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["cluster", "kubeconfig", "-p", "test", "-c", "test", "--auth", "token", "-o", "json"])
+    assert result.exit_code == 0, result.output
+
+    assert mock_request.call_args.kwargs["params"] == {"ttl": "1s"}
+    kubeconfig = json.loads(result.output)
+    assert kubeconfig["clusters"][0]["cluster"] == {"server": "https://kube.example:6443", "certificate-authority-data": "Q0E="}
+    assert kubeconfig["users"][0]["user"]["exec"]["args"] == ["cluster", "token", "--url", "https://kube.example:6443"]
+
+# Test the "cluster kubeconfig --auth token" command: the generated kubeconfig is cached until --refresh
+@patch("oks_cli.utils.requests.request")
+def test_cluster_kubeconfig_auth_token_is_cached(mock_request, add_default_profile):
+    issued = yaml.safe_dump({
+        "clusters": [{"name": "test", "cluster": {"server": "https://kube.example:6443", "certificate-authority-data": "Q0E="}}],
+    })
+    lookups = [
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345"}]}),
+    ]
+    issue = MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Cluster":  {"data": {"kubeconfig": issued}}})
+    mock_request.side_effect = lookups + [issue] + lookups + lookups + [issue]
+
+    runner = CliRunner()
+    command = ["cluster", "kubeconfig", "-p", "test", "-c", "test", "--auth", "token"]
+
+    first = runner.invoke(cli, command + ["--print-path"])
+    assert first.exit_code == 0, first.output
+    assert first.output.strip().endswith("kubeconfig-token")
+    assert mock_request.call_count == 3
+
+    second = runner.invoke(cli, command)
+    assert second.exit_code == 0, second.output
+    assert mock_request.call_count == 5
+    with open(first.output.strip()) as f:
+        assert second.output.strip() == f.read().strip()
+
+    refreshed = runner.invoke(cli, command + ["--refresh"])
+    assert refreshed.exit_code == 0, refreshed.output
+    assert mock_request.call_count == 8
+
+# Test the "cluster kubeconfig --auth token" command: client certificate options are rejected
+@patch("oks_cli.utils.requests.request")
+def test_cluster_kubeconfig_auth_token_rejects_cert_options(mock_request, add_default_profile):
+    runner = CliRunner()
+    result = runner.invoke(cli, ["cluster", "kubeconfig", "-p", "test", "-c", "test", "--auth", "token", "--ttl", "1d"])
+    assert result.exit_code != 0
+    assert "only supported with --auth cert" in result.output
+    mock_request.assert_not_called()
+
 # Test the "cluster kubeconfig --output table" command: verifies retrieving the kubeconfig and output as table
 @patch("oks_cli.utils.requests.request")
 def test_cluster_kubeconfig_info_command(mock_request, add_default_profile):
@@ -435,6 +498,33 @@ def test_cluster_kubectl_command(mock_request, mock_run, add_default_profile):
     assert ".oks_cli/cache/12345-12345/default/default/kubeconfig" in kwargs["env"]["KUBECONFIG"]
     assert args[0] == ["kubectl", "get", "pods"]
 
+# Test the "cluster kubectl --auth token" command: kubectl runs with the cached token kubeconfig
+@patch("oks_cli.utils.subprocess.run")
+@patch("oks_cli.utils.requests.request")
+def test_cluster_kubectl_auth_token_command(mock_request, mock_run, add_default_profile):
+    issued = yaml.safe_dump({
+        "clusters": [{"name": "test", "cluster": {"server": "https://kube.example:6443", "certificate-authority-data": "Q0E="}}],
+    })
+    lookups = [
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345"}]}),
+    ]
+    issue = MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Cluster":  {"data": {"kubeconfig": issued}}})
+    mock_request.side_effect = lookups + [issue] + lookups
+    mock_run.return_value = MagicMock(returncode = 0, stdout = "Success", stderr = "")
+
+    runner = CliRunner()
+    for _ in range(2):
+        result = runner.invoke(cli, ["cluster", "-p", "test", "-c", "test", "kubectl", "--auth", "token", "get", "pods"])
+        assert result.exit_code == 0, result.output
+
+    assert mock_request.call_count == 5
+    args, kwargs = mock_run.call_args
+    assert kwargs["env"]["KUBECONFIG"].endswith(".oks_cli/cache/12345-12345/default/default/kubeconfig-token")
+    assert args[0] == ["kubectl", "get", "pods"]
+    with open(kwargs["env"]["KUBECONFIG"]) as f:
+        assert yaml.safe_load(f)["users"][0]["user"]["exec"]["args"] == ["cluster", "token", "--url", "https://kube.example:6443"]
+
 # Test the "cluster create by one-click" command: verifies creating cluster interactively
 @patch("oks_cli.utils.os.fork")
 @patch("oks_cli.utils.time.sleep")
diff --git a/tests/test_nodepool.py b/tests/test_nodepool.py
index 7c11c83..b247cd8 100644
--- a/tests/test_nodepool.py
+++ b/tests/test_nodepool.py
@@ -2,6 +2,8 @@
 from oks_cli.main import cli
 from unittest.mock import patch, MagicMock
 
+import yaml
+
 
 @patch("oks_cli.utils.subprocess.run")
 @patch("oks_cli.utils.requests.request")
@@ -89,3 +91,33 @@ def test_nodepool_delete_command(mock_request, mock_run, add_default_profile):
     assert result.exit_code == 0
     assert ".oks_cli/cache/12345-12345/default/default/kubeconfig" in kwargs["env"]["KUBECONFIG"]
     assert args[0] == ["kubectl", "delete", "nodepool", "test"]
+
+@patch("oks_cli.utils.subprocess.run")
+@patch("oks_cli.utils.requests.request")
+def test_nodepool_list_auth_token_command(mock_request, mock_run, add_default_profile):
+    issued = yaml.safe_dump({
+        "clusters": [{"name": "test", "cluster": {"server": "https://kube.example:6443", "certificate-authority-data": "Q0E="}}],
+    })
+    mock_request.side_effect = [
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Projects": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Clusters": [{"id": "12345"}]}),
+        MagicMock(status_code=200, headers = {}, json=lambda: {"ResponseContext": {}, "Cluster":  {"data": {"kubeconfig": issued}}})
+    ]
+    mock_run.return_value = MagicMock(returncode = 0, stdout = "Success", stderr = "")
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["cluster", "nodepool", "-p", "test", "-c", "test", "--auth", "token", "list"])
+    assert result.exit_code == 0, result.output
+
+    assert mock_request.call_args.kwargs["params"] == {"ttl": "1s"}
+    args, kwargs = mock_run.call_args
+    assert kwargs["env"]["KUBECONFIG"].endswith(".oks_cli/cache/12345-12345/default/default/kubeconfig-token")
+    assert args[0] == ["kubectl", "get", "nodepool", "-o", "wide"]
+
+@patch("oks_cli.utils.requests.request")
+def test_nodepool_auth_token_rejects_user_group(mock_request, add_default_profile):
+    runner = CliRunner()
+    result = runner.invoke(cli, ["cluster", "nodepool", "-p", "test", "-c", "test", "--auth", "token", "--user", "admin", "list"])
+    assert result.exit_code != 0
+    assert "only supported with --auth cert" in result.output
+    mock_request.assert_not_called()