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

@@ -5,7 +5,7 @@
 import argparse
 import os
 import sys
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Type, TypeVar, cast
 
 from linodecli.exit_codes import ExitCodes
 
@@ -27,6 +27,8 @@
 
 ENV_TOKEN_NAME = "LINODE_CLI_TOKEN"
 
+T = TypeVar("T")
+
 
 class CLIConfig:
     """
@@ -216,13 +218,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 +235,52 @@ 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_get_config_value_or_set_default(
+        self, 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 = self.plugin_get_value(key)
+
+        if value is None:
+            # option not set - set to default and store it in the config file
+            value_as_str = (
+                ("yes" if default else "no")
+                if value_type is bool
+                else str(default)
+            )
+            self.plugin_set_value(key, value_as_str)
+            self.write_config()
+            return default
+
+        if value_type is bool:
+            return self.parse_boolean(value)
+
+        return cast(T, value_type(value))
+
+    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 +688,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

@@ -4,20 +4,21 @@
 """
 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
 
+from pytimeparse import parse as parse_time
 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
@@ -322,6 +323,16 @@ def get_obj_args_parser():
         type=str,
         help="The cluster to use for the operation",
     )
+    parser.add_argument(
+        "--skip-key-cleanup",
+        action="store_true",
+        help="Skip cleanup of old linode-cli generated Object Storage keys",
+    )
+    parser.add_argument(
+        "--force-key-cleanup",
+        action="store_true",
+        help="Force cleanup of old linode-cli generated Object Storage keys",
+    )
 
     return parser
 
@@ -400,8 +411,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:
+        if not parsed.skip_key_cleanup:
+            _cleanup_keys(context.client, parsed.force_key_cleanup)
         access_key, secret_key = get_credentials(context.client)
 
     cluster = parsed.cluster
@@ -580,7 +593,7 @@ def _get_s3_creds(client: CLI, force: bool = False):
 
 def _configure_plugin(client: CLI):
     """
-    Configures a default cluster value.
+    Configures Object Storage plugin.
     """
 
     cluster = _default_text_input(  # pylint: disable=protected-access
@@ -590,4 +603,211 @@ def _configure_plugin(client: CLI):
 
     if cluster:
         client.config.plugin_set_value("cluster", cluster)
+
+    perform_key_cleanup = _default_text_input(  # pylint: disable=protected-access
+        "Perform automatic cleanup of old linode-cli generated Object Storage keys? (Y/n)",
+        default="y",
+        validator=lambda x: (
+            "Please enter 'y' or 'n'"
+            if x.lower() not in ("y", "n", "yes", "no")
+            else None
+        ),
+    )
+    perform_key_cleanup = (
+        "yes" if perform_key_cleanup.lower() in ("y", "yes") else "no"
+    )
+    client.config.plugin_set_value("perform-key-cleanup", perform_key_cleanup)
+    if perform_key_cleanup == "yes":
+        key_lifespan = _default_text_input(  # pylint: disable=protected-access
+            "Linode-cli Object Storage key lifespan (e.g. `30d` for 30 days)",
+            default="30d",
+            validator=lambda x: (
+                "Please enter a valid time duration"
+                if parse_time(x) is None
+                else None
+            ),
+        )
+        client.config.plugin_set_value("key-lifespan", key_lifespan)
+
+        key_rotation_period = _default_text_input(  # pylint: disable=protected-access
+            "Linode-cli Object Storage key rotation period (e.g. `10d` for 10 days)",
+            default="10d",
+            validator=lambda x: (
+                "Please enter a valid time duration"
+                if parse_time(x) is None
+                else None
+            ),
+        )
+        client.config.plugin_set_value(
+            "key-rotation-period-days", key_rotation_period
+        )
+
+        cleanup_batch_size = _default_text_input(  # pylint: disable=protected-access
+            "Number of old linode-cli Object Storage keys to clean up at once",
+            default="10",
+            validator=lambda x: (
+                "Please enter a valid integer" if not x.isdigit() else None
+            ),
+        )
+        client.config.plugin_set_value(
+            "key-cleanup-batch-size", str(int(cleanup_batch_size))
+        )
+
     client.config.write_config()
+
+
+def _cleanup_keys(client: CLI, force_key_cleanup: bool) -> None:
+    """
+    Cleans up stale linode-cli generated object storage keys.
+    """
+    try:
+        if not _should_perform_key_cleanup(client):
+            return
+
+        current_timestamp = int(time.time())
+        last_cleanup = client.config.plugin_get_value(
+            "last-key-cleanup-timestamp"
+        )
+        if (
+            not force_key_cleanup
+            and last_cleanup
+            and int(last_cleanup) > current_timestamp - 24 * 60 * 60
+        ):
+            # if we did a cleanup in the last 24 hours, skip
+            return
+
+        print(
+            "Cleaning up old linode-cli Object Storage keys."
+            " To disable this, use the --skip-key-cleanup option.",
+            file=sys.stderr,
+        )
+        status, keys = client.call_operation("object-storage", "keys-list")
+        if status != 200:
+            print(
+                "Failed to list object storage keys for cleanup",
+                file=sys.stderr,
+            )
+            return
+
+        key_lifespan = _get_key_lifespan(client)
+        key_rotation_period = _get_key_rotation_period(client)
+        cleanup_batch_size = _get_cleanup_batch_size(client)
+
+        linode_cli_keys = _get_linode_cli_keys(
+            keys["data"], key_lifespan, key_rotation_period, current_timestamp
+        )
+        _rotate_current_key_if_needed(client, linode_cli_keys)
+        _delete_stale_keys(client, linode_cli_keys, cleanup_batch_size)
+
+        client.config.plugin_set_value(
+            "last-key-cleanup-timestamp", str(current_timestamp)
+        )
+        client.config.write_config()
+
+    except Exception as e:
+        print(
+            "Unable to clean up stale linode-cli Object Storage keys",
+            e,
+            file=sys.stderr,
+        )
+
+
+def _should_perform_key_cleanup(client: CLI) -> bool:
+    return client.config.plugin_get_config_value_or_set_default(
+        "perform-key-cleanup", True, bool
+    )
+
+
+def _get_key_lifespan(client) -> str:
+    return client.config.plugin_get_config_value_or_set_default(
+        "key-lifespan", "30d"
+    )
+
+
+def _get_key_rotation_period(client) -> str:
+    return client.config.plugin_get_config_value_or_set_default(
+        "key-rotation-period-days", "10d"
+    )
+
+
+def _get_cleanup_batch_size(client) -> int:
+    return client.config.plugin_get_config_value_or_set_default(
+        "key-cleanup-batch-size", 10, int
+    )
+
+
+def _get_linode_cli_keys(
+    keys_data: list,
+    key_lifespan: str,
+    key_rotation_period: str,
+    current_timestamp: int,
+) -> list:
+
+    stale_threshold = current_timestamp - parse_time(key_lifespan)
+    rotation_threshold = current_timestamp - parse_time(key_rotation_period)
+
+    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))
+        is_stale = created_timestamp <= stale_threshold
+        needs_rotation = is_stale or created_timestamp <= rotation_threshold
+        return {
+            "id": key["id"],
+            "label": key["label"],
+            "access_key": key["access_key"],
+            "created_timestamp": created_timestamp,
+            "is_stale": is_stale,
+            "needs_rotation": needs_rotation,
+        }
+
+    return sorted(
+        [info for key in keys_data if (info := extract_key_info(key))],
+        key=lambda k: k["created_timestamp"],
+    )
+
+
+def _rotate_current_key_if_needed(client: CLI, linode_cli_keys: list) -> None:
+    current_access_key = client.config.plugin_get_value("access-key")
+    key_to_rotate = next(
+        (
+            key_info
+            for key_info in linode_cli_keys
+            if key_info["access_key"] == current_access_key
+            and key_info["needs_rotation"]
+        ),
+        None,
+    )
+    if key_to_rotate:
+        _delete_key(client, key_to_rotate["id"], key_to_rotate["label"])
+        linode_cli_keys.remove(key_to_rotate)
+        client.config.plugin_remove_option("access-key")
+        client.config.plugin_remove_option("secret-key")
+        client.config.write_config()
+
+
+def _delete_stale_keys(
+    client: CLI, linode_cli_keys: list, batch_size: int
+) -> None:
+    stale_keys = [k for k in linode_cli_keys if k["is_stale"]]
+    for key_info in stale_keys[:batch_size]:
+        _delete_key(client, key_info["id"], key_info["label"])
+
+
+def _delete_key(client: CLI, key_id: str, label: str) -> None:
+    try:
+        print(f"Deleting linode-cli Object Storage key: {label}")
+        status, _ = client.call_operation(
+            "object-storage", "keys-delete", [str(key_id)]
+        )
+        if status != 200:
+            print(
+                f"Failed to delete key: {label}; status {status}",
+                file=sys.stderr,
+            )
+    except Exception as e:
+        print(
+            f"Exception occurred while deleting key: {label}; {e}",
+            file=sys.stderr,
+        )

pyproject.toml

@@ -14,10 +14,11 @@ dependencies = [
     "openapi3",
     "requests",
     "PyYAML",
-    "packaging",
-    "rich",
+    "packaging<25",
+    "rich<14",
     "urllib3<3",
-    "linode-metadata>=0.3.0"
+    "linode-metadata>=0.3.0",
+    "pytimeparse"
 ]
 dynamic = ["version"]