Feat/poc sign v4

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:

oks_cli/utils.py

@@ -10,14 +10,16 @@
 import pathlib
 import traceback
 import time
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 import OpenSSL
 import shutil
 import prettytable
 import dateutil.parser
 import human_readable
 import base64
 import sys
+import hmac
+import hashlib
 
 from click.shell_completion import CompletionItem
 
@@ -104,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', {})))
 
@@ -117,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:
@@ -131,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)
@@ -144,6 +153,84 @@ 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):
+    """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, '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):
+    """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, payload=''):
+    """
+    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(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)
+
+    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)
+
+    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_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(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')
+
+    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, signed_url_headers(url)),
+    })
+
 def build_headers():
     """Build HTTP headers for API requests based on environment authentication settings."""
     headers = {
@@ -160,12 +247,20 @@ def build_headers():
             if _h and _n:
                 headers[_h.strip()] = _n.strip() 
 
-    if is_jwt_enabled() and is_tokens_valid():
+    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 = 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_ACCESS_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")
@@ -257,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):
@@ -751,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:

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
+    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