new: STORIF-108: Added functionality to clean-up stale (e.g. longer than 30 days) Object Storage keys created by linode-cli.

Author: skulpok-akamai

linodecli/configuration/config.py

@@ -216,13 +216,8 @@ def plugin_set_value(self, key: str, value: Any):
         :param value: The value to set for this key
         :type value: any
         """
-        if self.running_plugin is None:
-            raise RuntimeError(
-                "No running plugin to retrieve configuration for!"
-            )
-
         username = self.username or self.default_username()
-        self.config.set(username, f"plugin-{self.running_plugin}-{key}", value)
+        self.config.set(username, self._get_plugin_key(key), value)
 
     def plugin_get_value(self, key: str) -> Optional[Any]:
         """
@@ -238,15 +233,27 @@ def plugin_get_value(self, key: str) -> Optional[Any]:
         :returns: The value for this plugin for this key, or None if not set
         :rtype: any
         """
+        username = self.username or self.default_username() or "DEFAULT"
+        return self.config.get(
+            username, self._get_plugin_key(key), fallback=None
+        )
+
+    def plugin_remove_option(self, key: str):
+        """
+        Removes a plugin configuration option.
+
+        :param key: The key of the option to remove
+        """
+        username = self.username or self.default_username()
+        self.config.remove_option(username, self._get_plugin_key(key))
+
+    def _get_plugin_key(self, key: str) -> str:
         if self.running_plugin is None:
             raise RuntimeError(
                 "No running plugin to retrieve configuration for!"
             )
 
-        username = self.username or self.default_username() or "DEFAULT"
-        full_key = f"plugin-{self.running_plugin}-{key}"
-
-        return self.config.get(username, full_key, fallback=None)
+        return f"plugin-{self.running_plugin}-{key}"
 
     # TODO: this is more of an argparsing function than it is a config function
     # might be better to move this to argparsing during refactor and just have
@@ -654,3 +661,12 @@ def get_custom_aliases(self) -> Dict[str, str]:
             if (self.config.has_section("custom_aliases"))
             else {}
         )
+
+    def parse_boolean(self, value: str) -> Optional[bool]:
+        """
+        Parses a string config value into a boolean.
+
+        :param value: The string value to parse.
+        :return: The parsed boolean value.
+        """
+        return self.config.BOOLEAN_STATES.get(value.lower(), None)

linodecli/plugins/obj/__init__.py

@@ -2,22 +2,23 @@
 """
 CLI Plugin for handling OBJ
 """
+import concurrent.futures
 import getpass
 import os
+import re
 import socket
 import sys
 import time
 from argparse import ArgumentParser
 from contextlib import suppress
 from datetime import datetime
 from math import ceil
-from typing import List
+from typing import List, Optional, Type, TypeVar, cast
 
 from rich import print as rprint
 from rich.table import Table
 
 from linodecli.cli import CLI
-from linodecli.configuration import _do_get_request
 from linodecli.configuration.helpers import _default_text_input
 from linodecli.exit_codes import ExitCodes
 from linodecli.help_formatter import SortingHelpFormatter
@@ -65,6 +66,9 @@
     HAS_BOTO = False
 
 
+T = TypeVar("T")
+
+
 def generate_url(get_client, args, **kwargs):  # pylint: disable=unused-argument
     """
     Generates a URL to an object
@@ -400,9 +404,10 @@ def call(
     access_key = None
     secret_key = None
 
-    # make a client, but only if we weren't printing help
+    # make a client and clean-up keys, but only if we weren't printing help
     if not is_help:
         access_key, secret_key = get_credentials(context.client)
+        _cleanup_keys(context.client)
 
     cluster = parsed.cluster
     if context.client.defaults:
@@ -591,3 +596,125 @@ def _configure_plugin(client: CLI):
     if cluster:
         client.config.plugin_set_value("cluster", cluster)
     client.config.write_config()
+
+
+def _cleanup_keys(client: CLI):
+    """
+    Cleans up stale linode-cli generated object storage keys.
+    """
+    try:
+        perform_key_cleanup = _get_config_value_or_set_default(
+            client, "perform-key-cleanup", True, bool
+        )
+        if not perform_key_cleanup:
+            return
+
+        status, keys = client.call_operation("object-storage", "keys-list")
+        if status != 200:
+            print(
+                "Failed to list object storage keys for cleanup\n",
+                file=sys.stderr,
+            )
+            return
+
+        key_lifespan_in_days = _get_config_value_or_set_default(
+            client, "key-lifespan-in-days", 30, int
+        )
+        key_rotation_period = _get_config_value_or_set_default(
+            client, "key-rotation-period", 10, int
+        )
+        cleanup_batch_size = _get_config_value_or_set_default(
+            client, "key-cleanup-batch-size", 10, int
+        )
+
+        def extract_key_info(key: dict) -> Optional[dict]:
+            match = re.match(r"^linode-cli-.+@.+-(\d{10})$", key["label"])
+            if not match:
+                return None
+            created_timestamp = int(match.group(1))
+            current_timestamp = int(time.time())
+            stale_threshold = (
+                current_timestamp - key_lifespan_in_days * 24 * 60 * 60
+            )
+            rotation_threshold = (
+                current_timestamp - key_rotation_period * 24 * 60 * 60
+            )
+            return {
+                "id": key["id"],
+                "label": key["label"],
+                "access_key": key["access_key"],
+                "created_timestamp": created_timestamp,
+                "is_stale": created_timestamp <= stale_threshold,
+                "needs_rotation": created_timestamp <= rotation_threshold,
+            }
+
+        linode_cli_keys = sorted(
+            [
+                key_info
+                for key in keys["data"]
+                if (key_info := extract_key_info(key)) and key_info["is_stale"]
+            ],
+            key=lambda x: x["created_timestamp"],
+        )
+
+        current_access_key = client.config.plugin_get_value("access-key")
+        current_key_needs_rotation = any(
+            key_info["access_key"] == current_access_key
+            and key_info["needs_rotation"]
+            for key_info in linode_cli_keys
+        )
+        if current_key_needs_rotation:
+            client.config.plugin_remove_option("access-key")
+            client.config.plugin_remove_option("secret-key")
+            client.config.write_config()
+
+        def delete_key(key_info):
+            try:
+                status, _ = client.call_operation(
+                    "object-storage", "keys-delete", [str(key_info["id"])]
+                )
+                if status != 200:
+                    print(
+                        f"Failed to delete key: {key_info['label']}; status {status}\n",
+                        file=sys.stderr,
+                    )
+            except Exception as e:
+                print(
+                    f"Exception occurred while deleting key: {key_info['label']}; {e}\n",
+                    file=sys.stderr,
+                )
+
+        # concurrently delete keys up to the batch size
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            executor.map(delete_key, linode_cli_keys[:cleanup_batch_size])
+
+    except Exception as e:
+        print(
+            "Unable to clean up stale linode-cli Object Storage keys\n",
+            e,
+            file=sys.stderr,
+        )
+
+
+def _get_config_value_or_set_default(
+    client: CLI, key: str, default: T, value_type: Type[T] = str
+) -> T:
+    """
+    Retrieves a plugin option value of the given type from the config. If the
+    value is not set, sets it to the provided default value and returns that.
+    """
+    value = client.config.plugin_get_value(key)
+
+    if value is None:
+        # option not set - set to default and store it in the config file
+        value_as_str = (
+            ("on" if default else "off") if value_type is bool else str(default)
+        )
+        client.config.plugin_set_value(key, value_as_str)
+        client.config.write_config()
+        return default
+
+    if value_type is bool:
+        return client.config.parse_boolean(value)
+
+    return cast(T, value_type(value))