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 75b15ff..cd2e7c1 100644 --- a/oks_cli/cluster.py +++ b/oks_cli/cluster.py @@ -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, get_sigv4_token_expiration from .profile import add_profile from .project import project_create, project_login @@ -622,6 +623,103 @@ def cluster_delete_command(ctx, project_name, cluster_name, output, dry_run, for set_cluster_id("") print_output(data, output) +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, "certificate-authority-data": ca_data}}], + "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": [{"name": "OKS_PROFILE", "value": profile}], + "interactiveMode": "Never", + }, + }, + }], + } + +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) @@ -634,61 +732,43 @@ 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('--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, 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 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) @@ -717,33 +797,70 @@ 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('--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, 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) + + 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") + + token = build_signed_requests_token(access_key, secret_key, region, url) + expiration = get_sigv4_token_expiration(token) -def _run_kubectl(project_id, cluster_id, user, group, args, input=None, capture=False): + 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, 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) @@ -765,24 +882,28 @@ 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) @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.") @@ -790,18 +911,25 @@ def cluster_kubectl_command(ctx, project_name, cluster_name, user, group, args, @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) @@ -812,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") @@ -845,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') @@ -866,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/oks_cli/utils.py b/oks_cli/utils.py index 432b8db..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 +from datetime import datetime, timedelta, 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 @@ -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: 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() 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())