diff --git a/LICENSE b/LICENSE index 58848b7..0855779 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2025 LeoTN +Copyright (c) 2025-2026 LeoTN Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/assets/CheckDependencies.ps1 b/assets/CheckDependencies.ps1 index bd5e833..cc9337e 100644 --- a/assets/CheckDependencies.ps1 +++ b/assets/CheckDependencies.ps1 @@ -117,5 +117,4 @@ else { Write-Host "`n[INFO] No transitive dependencies detected." -ForegroundColor Green } -Write-Host "`n[INFO] Done." -ForegroundColor Cyan -pause \ No newline at end of file +Write-Host "`n[INFO] Done." -ForegroundColor Cyan \ No newline at end of file diff --git a/assets/CheckDependencies_Launcher.bat b/assets/CheckDependencies_Launcher.bat index 5744023..208481b 100644 --- a/assets/CheckDependencies_Launcher.bat +++ b/assets/CheckDependencies_Launcher.bat @@ -5,4 +5,6 @@ echo Terminal ready... set psScriptPath="%CD%\CheckDependencies.ps1" -powershell.exe -executionPolicy bypass -file %psScriptPath% \ No newline at end of file +powershell.exe -executionPolicy bypass -file %psScriptPath% + +pause \ No newline at end of file diff --git a/assets/readme.gif b/assets/readme.gif index cb81d8f..6ac0438 100644 Binary files a/assets/readme.gif and b/assets/readme.gif differ diff --git a/assets/readme.tape b/assets/readme.tape index 07b67ac..4b5c8ef 100644 --- a/assets/readme.tape +++ b/assets/readme.tape @@ -1,6 +1,6 @@ Output readme.gif -Require sct.exe +Require sct Set Shell powershell Set Width 1500 diff --git a/pyproject.toml b/pyproject.toml index 5e6375e..78b03b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,37 +2,45 @@ requires = ["poetry-core>=1.0.0"] build-backend = "poetry.core.masonry.api" -[tool.poetry] +[project] name = "step-cli-tools" version = "0.0.0" description = "Prebuilt tools for the smallstep step-cli" +license = "MIT" +license-files = ["LICENSE"] readme = "README.md" -authors = ["LeoTN "] -license = "LICENSE" -repository = "https://github.com/LeoTN/step-cli-tools" -keywords = ["step", "cli", "step-cli", "step-ca", "certificate", "root-ca", "automation"] +authors = [ + { name = "LeoTN", email = "LeoTN.GitHub@gmx.net" } +] +keywords = ["step-ca", "step-cli", "ca", "cli", "certificate", "certificate-authority"] classifiers = [ + "Topic :: Utilities", + "Environment :: Console", "Programming Language :: Python :: 3", - "License :: OSI Approved :: MIT License", "Operating System :: OS Independent" ] + +[project.urls] +homepage = "https://github.com/LeoTN/step-cli-tools#readme" +repository = "https://github.com/LeoTN/step-cli-tools" +"Bug Tracker" = "https://github.com/LeoTN/step-cli-tools/issues" + +[project.scripts] +step-cli-tools = "step_cli_tools.main:main" +sct = "step_cli_tools.main:main" + +[tool.poetry] packages = [ { include = "step_cli_tools" } ] [tool.poetry.dependencies] python = ">=3.10,<4.0" -cryptography = ">=46.0.2" -rich = ">=14.2.0" -questionary = "^2.1.1" -ruamel-yaml = "^0.18.16" +cryptography = "^46.0.3" packaging = "^25.0" +questionary = "^2.1.1" +rich = "^14.2.0" +ruamel-yaml = "^0.19.1" -[tool.poetry.scripts] -step-cli-tools = "step_cli_tools.main:main" -sct = "step_cli_tools.main:main" - -[dependency-groups] -dev = [ - "deptry (>=0.23.1,<0.24.0)" -] \ No newline at end of file +[tool.poetry.group.dev.dependencies] +deptry = ">=0.24.0,<0.25.0" \ No newline at end of file diff --git a/step_cli_tools/common.py b/step_cli_tools/common.py index a87e05e..4cde2c8 100644 --- a/step_cli_tools/common.py +++ b/step_cli_tools/common.py @@ -1,20 +1,35 @@ # --- Standard library imports --- +import logging +from logging.handlers import RotatingFileHandler import os import platform # --- Third-party imports --- import questionary from rich.console import Console +from rich.logging import RichHandler +from rich.theme import Theme __all__ = [ "console", "qy", "DEFAULT_QY_STYLE", "SCRIPT_HOME_DIR", + "SCRIPT_LOGGING_DIR", "STEP_BIN", + "logger", ] -console = Console() + +custom_logging_theme = Theme( + { + "logging.level.info": "none", + "logging.level.warning": "#F9ED69", + "logging.level.error": "#B83B5E", + "logging.level.critical": "bold reverse #B83B5E", + } +) +console = Console(theme=custom_logging_theme) qy = questionary # Default style to use for questionary DEFAULT_QY_STYLE = qy.Style( @@ -25,18 +40,18 @@ ("answer", "fg:#F08A5D"), ] ) + +# --- Directories and files --- SCRIPT_HOME_DIR = os.path.expanduser("~/.step-cli-tools") +SCRIPT_LOGGING_DIR = os.path.normpath(os.path.join(SCRIPT_HOME_DIR, "logs")) -def get_step_binary_path() -> str: +def _get_step_binary_path() -> str: """ Get the absolute path to the step-cli binary based on the operating system. Returns: str: Absolute path to the step binary. - - Raises: - OSError: If the operating system is not supported. """ bin_dir = os.path.join(SCRIPT_HOME_DIR, "bin") @@ -51,4 +66,67 @@ def get_step_binary_path() -> str: return os.path.normpath(binary) -STEP_BIN = get_step_binary_path() +STEP_BIN = _get_step_binary_path() + +# --- Logging --- + + +def _setup_logger( + name: str, + log_file: str = "step-cli-tools.log", + level=logging.DEBUG, + console: Console = console, + max_bytes: int = 5_000_000, + backup_count: int = 5, +) -> logging.Logger: + """ + Sets up a reusable logger with Rich console output. + + Args: + name: Name of the logger. + log_file: Path to the log file. + level: Logging level. + console: Console instance used for output. + max_bytes: Maximum size of log file in bytes. + backup_count: Number of log files to keep. + + Returns: + A logger instance. + """ + + # Ensure log directory exists + if os.path.dirname(log_file): + os.makedirs(os.path.dirname(log_file), exist_ok=True) + + logger = logging.getLogger(name) + logger.setLevel(level) + # Avoid duplicate logs if root logger is configured + logger.propagate = False + + if not logger.handlers: + # Rotating file handler (plain text) + file_handler = RotatingFileHandler( + log_file, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" + ) + file_handler.setFormatter( + logging.Formatter( + "%(asctime)s %(levelname)-8s [%(funcName)-30s] %(message)s" + ) + ) + file_handler.setLevel(level) + logger.addHandler(file_handler) + + # Rich console handler (colorful) + console_handler = RichHandler( + console=console, rich_tracebacks=True, show_time=False, show_path=False + ) + console_handler.setLevel(level) + logger.addHandler(console_handler) + + return logger + + +logger = _setup_logger( + name="main", + log_file=os.path.join(SCRIPT_LOGGING_DIR, "step-cli-tools.log"), +) diff --git a/step_cli_tools/configuration.py b/step_cli_tools/configuration.py index 5b9a117..1f8676e 100644 --- a/step_cli_tools/configuration.py +++ b/step_cli_tools/configuration.py @@ -1,13 +1,15 @@ # --- Standard library imports --- +from datetime import datetime +from logging.handlers import RotatingFileHandler +from pathlib import Path import os import platform import shutil import subprocess import sys -from datetime import datetime -from pathlib import Path # --- Third-party imports --- +from rich.logging import RichHandler from ruamel.yaml import YAML from ruamel.yaml.comments import CommentedMap @@ -50,7 +52,7 @@ def load(self): try: loaded = yaml.load(self.file_location.read_text()) or {} except Exception as e: - console.print(f"[WARNING] Failed to load config: {e}", style="#F9ED69") + logger.warning(f"Failed to load config: {e}") loaded = {} else: loaded = {} @@ -63,10 +65,7 @@ def save(self): with self.file_location.open("w", encoding="utf-8") as f: yaml.dump(self._data, f) except (OSError, IOError) as e: - console.print( - f"[ERROR] Could not save settings to '{self.file_location}': {e}", - style="#B83B5E", - ) + logger.error(f"Could not save settings to '{self.file_location}': {e}") def generate_default(self, overwrite: bool = False): """ @@ -77,9 +76,8 @@ def generate_default(self, overwrite: bool = False): """ try: if self.file_location.exists() and not overwrite: - console.print( - f"[WARNING] Config file already exists: {self.file_location}. Use overwrite=True to replace it.", - style="#F9ED69", + logger.warning( + f"Config file already exists: {self.file_location}. Use overwrite=True to replace it." ) return @@ -90,7 +88,7 @@ def generate_default(self, overwrite: bool = False): f"{self.file_location.stem}_backup_{timestamp}{self.file_location.suffix}" ) shutil.copy2(self.file_location, backup_path) - console.print(f"[INFO] Created backup before overwrite: {backup_path}") + logger.info(f"Created backup before overwrite: {backup_path}") # This is a bit akward but the file is technically repaired without keeping any data. default_data = self._build_commented_data( @@ -101,19 +99,27 @@ def generate_default(self, overwrite: bool = False): with self.file_location.open("w", encoding="utf-8") as f: yaml.dump(default_data, f) - console.print( - f"[INFO] Default configuration file was generated successfully: {self.file_location}", - style="green", + logger.info( + f"Default configuration file was generated successfully: {self.file_location}" ) # Load the data into memory so it's ready for use self._data = default_data except Exception as e: - console.print( - f"[ERROR] Failed to generate default configuration: {e}", - style="#B83B5E", - ) + logger.error(f"Failed to generate default configuration: {e}") + + def apply(self): + """Apply current configuration data to relevant parts of the application.""" + + # Apply the logging configuration + for handler in logger.handlers: + # Console handler level + if isinstance(handler, RichHandler): + handler.setLevel((self.get("logging_config.log_level_console"))) + # File handler level + if isinstance(handler, RotatingFileHandler): + handler.setLevel((self.get("logging_config.log_level_file"))) def get(self, key: str): """Retrieve a setting value using dotted key path; fallback to default if missing. @@ -128,9 +134,8 @@ def get(self, key: str): data = self._data for part in parts: if not isinstance(data, dict) or part not in data: - console.print( - f"[WARNING] Failed to extract the value for '{key}' from the configuration file.", - style="#F9ED69", + logger.warning( + f"Failed to extract the value for '{key}' from the configuration file." ) return self._nested_get_default(parts) data = data[part] @@ -149,9 +154,8 @@ def set(self, key: str, value): # Check if key exists in schema schema_meta = self._nested_get_meta(parts) if not schema_meta: - console.print( - f"[WARNING] Key '{key}' does not exist in the config schema. Value '{value}' will still be set.", - style="#F9ED69", + logger.warning( + f"Key '{key}' does not exist in the config schema. Value '{value}' will still be set." ) # Navigate or create nested dictionaries @@ -166,9 +170,8 @@ def set(self, key: str, value): try: value = expected_type(value) except Exception: - console.print( - f"[WARNING] Failed to cast value '{value}' to {expected_type.__name__} for key '{key}'", - style="#F9ED69", + logger.warning( + f"Failed to cast value '{value}' to {expected_type.__name__} for key '{key}'." ) data[parts[-1]] = value @@ -197,7 +200,7 @@ def validate(self, key: str | None = None) -> bool: parts = key.split(".") meta = self._nested_get_meta(parts) if not meta: - console.print(f"[WARNING] No schema entry for '{key}'", style="#F9ED69") + logger.warning(f"No schema entry for '{key}'.") return False validator = meta.get("validator") @@ -209,33 +212,25 @@ def validate(self, key: str | None = None) -> bool: value = self.get(key) try: if not callable(validator): - console.print( - f"[WARNING] Validator for '{key}' is not callable: {validator!r}", - style="#F9ED69", + logger.warning( + f"Validator for '{key}' is not callable: {validator!r}" ) return False result = validator(value) except Exception as e: - console.print( - f"[ERROR] Validator for '{key}' raised an exception: {e}", - style="#B83B5E", - ) + logger.error(f"Validator for '{key}' raised an exception: {e}") return False if result is None: return True if isinstance(result, str): - console.print( - f"[WARNING] Validation failed for '{key}': {result}", - style="#F9ED69", - ) + logger.warning(f"Validation failed for '{key}': {result}") return False - console.print( - f"[ERROR] Validator for '{key}' returned unsupported value: {result!r}", - style="#B83B5E", + logger.error( + f"Validator for '{key}' returned unsupported value: {result!r}" ) return False @@ -281,9 +276,8 @@ def _validate_recursive(self, data: dict, schema: dict, prefix: str) -> bool: if "type" not in meta: sub_data = data.get(k, {}) if not isinstance(sub_data, dict): - console.print( - f"[WARNING] Expected dict at '{full_key}', got {type(sub_data).__name__}", - style="#F9ED69", + logger.warning( + f"Expected dict at '{full_key}', got {type(sub_data).__name__}." ) ok = False elif not self._validate_recursive(sub_data, meta, full_key): @@ -298,32 +292,24 @@ def _validate_recursive(self, data: dict, schema: dict, prefix: str) -> bool: value = data.get(k, meta.get("default")) if not callable(validator): - console.print( - f"[WARNING] Validator for '{full_key}' is not callable: {validator!r}", - style="#F9ED69", + logger.warning( + f"Validator for '{full_key}' is not callable: {validator!r}" ) ok = False continue result = validator(value) if isinstance(result, str): - console.print( - f"[WARNING] Validation failed for '{full_key}': {result}", - style="#F9ED69", - ) + logger.warning(f"Validation failed for '{full_key}': {result}") ok = False elif result is not None: - console.print( - f"[ERROR] Validator for '{full_key}' returned unsupported type: {result!r}", - style="#B83B5E", + logger.error( + f"Validator for '{full_key}' returned unsupported type: {result!r}" ) ok = False except Exception as e: - console.print( - f"[ERROR] Validator for '{full_key}' raised: {e}", - style="#B83B5E", - ) + logger.error(f"Validator for '{full_key}' raised: {e}") ok = False return ok @@ -342,10 +328,7 @@ def _nested_get_default(self, keys: list[str]): data = self.schema for k in keys: if not isinstance(data, dict) or k not in data: - console.print( - f"[WARNING] Missing default for key '{'.'.join(keys)}'", - style="#F9ED69", - ) + logger.warning(f"Missing default for key '{'.'.join(keys)}'.") return None data = data[k] if isinstance(data, dict) and "default" in data: @@ -406,9 +389,7 @@ def _build_commented_data( if node[key] is None: if repair_damaged_keys: if not suppress_repair_messages: - console.print( - f"[INFO] Repairing key '{key}' from config schema." - ) + logger.info(f"Repairing key '{key}' from config schema.") node[key] = meta.get("default") else: continue @@ -464,7 +445,7 @@ def check_and_repair_config_file(): # Generate default config if missing if not os.path.exists(config.file_location): config.generate_default() - console.print("[INFO] A default config file has been generated.") + logger.info("A default config file has been generated.") automatic_repair_failed = False @@ -473,16 +454,15 @@ def check_and_repair_config_file(): config.load() is_valid = config.validate() except Exception as e: - console.print( - f"[ERROR] Config validation raised an exception: {e}", style="#B83B5E" - ) + logger.error(f"Config validation raised an exception: {e}") is_valid = False if is_valid: + config.apply() break # valid -> exit if not automatic_repair_failed: - console.print("[INFO] Attempting automatic config file repair...") + logger.info("Attempting automatic config file repair...") config.repair() automatic_repair_failed = True continue # check the repaired file again @@ -552,7 +532,7 @@ def let_user_change_config_file(reset_instead_of_discard: bool = False): and reload if valid. If invalid, allow the user to discard or retry. Args: - reset_instead_of_discard: Replace the option "Discard changes" with "Reset config file" if True + reset_instead_of_discard: Replace the option "Discard changes" with "Reset config file" if True. """ # Backup current config @@ -571,17 +551,16 @@ def let_user_change_config_file(reset_instead_of_discard: bool = False): config.load() is_valid = config.validate() except Exception as e: - console.print( - f"[ERROR] Validation raised an exception: {e}", style="#B83B5E" - ) + logger.error(f"Validation raised an exception: {e}") is_valid = False if is_valid: - console.print("[INFO] Configuration saved successfully.", style="green") + config.apply() + logger.info("Configuration saved successfully.") break # exit loop if valid # If validation failed - console.print("[ERROR] Configuration is invalid.", style="#B83B5E") + logger.error("Configuration is invalid.") console.print() selected_action = qy.select( message="Choose an action:", @@ -600,7 +579,8 @@ def let_user_change_config_file(reset_instead_of_discard: bool = False): # Restore backup shutil.copy(backup_path, config.file_location) config.load() - console.print("[INFO] Changes discarded.") + config.apply() + logger.info("Changes discarded.") break # else: loop continues for "Edit again" @@ -610,9 +590,9 @@ def open_in_editor(file_path: str | Path): Open the given file in the user's preferred text editor and wait until it is closed. Respects the environment variable EDITOR if set, otherwise: - - On Windows: opens with 'notepad' - - On macOS: uses 'open -W -t' - - On Linux: tries common editors (nano, vim) or falls back to xdg-open (non-blocking) + - On Windows: opens with 'notepad'. + - On macOS: uses 'open -W -t'. + - On Linux: tries common editors (nano, vim) or falls back to xdg-open (non-blocking). """ path = Path(file_path).expanduser().resolve() @@ -651,28 +631,31 @@ def open_in_editor(file_path: str | Path): return # fallback: GUI open (non-blocking) subprocess.Popen(["xdg-open", str(path)]) - console.print( - "[INFO] File opened in default GUI editor. Please close it manually." - ) - input("[INFO] Press Enter here when you're done editing...") + logger.info("File opened in default GUI editor. Please close it manually.") + input("Press Enter here when you're done editing...") def validate_with_feedback(): + """Validate the config file and apply changes if valid.""" + config.load() result = config.validate() if result is True: - console.print("[INFO] Configuration is valid.", style="green") + config.apply() + logger.info("Configuration is valid.") else: - console.print("[ERROR] Configuration is invalid.", style="#B83B5E") + logger.error("Configuration is invalid.") return result def reset_with_feedback(): + """Reset the config file to its default settings.""" + result = config.generate_default(overwrite=True) if result is True: - console.print("[INFO] Configuration successfully reset.", style="green") + logger.info("Configuration successfully reset.") else: - console.print("[ERROR] Configuration reset failed.", style="#B83B5E") + logger.error("Configuration reset failed.") return result @@ -680,7 +663,6 @@ def reset_with_feedback(): config_file_location = os.path.join(SCRIPT_HOME_DIR, "config.yml") config_schema = { "update_config": { - "comment": "Settings for controlling the update check", "check_for_updates_at_launch": { "type": bool, "default": True, @@ -703,7 +685,6 @@ def reset_with_feedback(): }, }, "ca_server_config": { - "comment": "Settings that affect the CA server behavior", "default_ca_server": { "type": str, "default": "", @@ -723,6 +704,22 @@ def reset_with_feedback(): "comment": "If false, the root certificate won't be fetched automatically from the CA server. You will need to enter the fingerprint manually when installing a root CA certificate", }, }, + "logging_config": { + "log_level_console": { + "type": str, + "default": "INFO", + "allowed": ["DEBUG", "INFO", "WARNING", "ERROR"], + "validator": str_allowed_validator, + "comment": "The logging level to be used for console output", + }, + "log_level_file": { + "type": str, + "default": "DEBUG", + "allowed": ["DEBUG", "INFO", "WARNING", "ERROR"], + "validator": str_allowed_validator, + "comment": "The logging level to be used for log files", + }, + }, } # This object will be used to manipulate the config file diff --git a/step_cli_tools/data_classes.py b/step_cli_tools/data_classes.py index eb1191e..06f8719 100644 --- a/step_cli_tools/data_classes.py +++ b/step_cli_tools/data_classes.py @@ -1,6 +1,6 @@ +# --- Standard library imports --- from dataclasses import dataclass - __all__ = ["CARootInfo"] diff --git a/step_cli_tools/main.py b/step_cli_tools/main.py index 694859b..1063da6 100644 --- a/step_cli_tools/main.py +++ b/step_cli_tools/main.py @@ -3,6 +3,9 @@ import sys from importlib.metadata import PackageNotFoundError, version +# --- Third-party imports --- +from rich.logging import RichHandler + # Allows the script to be run directly and still find the package modules if __name__ == "__main__" and __package__ is None: parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -12,8 +15,8 @@ # --- Local application imports --- from .common import * from .configuration import * -from .support_functions import * from .operations import * +from .support_functions import * # --- Main function --- @@ -26,11 +29,24 @@ def main(): except PackageNotFoundError: pkg_version = "0.0.0" + # Mute console logging + for handler in logger.handlers: + if isinstance(handler, RichHandler): + handler.setLevel("CRITICAL") + # Mark the log starting point + bannerText = f"# {pkg_name} - Version {pkg_version} #" + textArray = ["", "#" * len(bannerText), bannerText, "#" * len(bannerText), ""] + for text in textArray: + logger.info(text) + # Unmute console logging + for handler in logger.handlers: + if isinstance(handler, RichHandler): + handler.setLevel(logger.level) + # Verify and load the config file check_and_repair_config_file() - config.load() - # Check for updates and when running a release version (not 0.0.0) + # Check for updates when running a release version (not 0.0.0) if ( config.get("update_config.check_for_updates_at_launch") and pkg_version != "0.0.0" @@ -39,7 +55,9 @@ def main(): "update_config.consider_beta_versions_as_available_updates" ) latest_version = check_for_update( - pkg_version, include_prerelease=include_prerelease + pkg_name=pkg_name, + current_pkg_version=pkg_version, + include_prerelease=include_prerelease, ) else: latest_version = None @@ -78,11 +96,11 @@ def main(): console.print(f"{logo}") console.print(version_text) - # Ensure Step CLI is installed + # Ensure step-cli is installed if not os.path.exists(STEP_BIN): console.print() answer = qy.confirm( - message="Step CLI not found. Do you want to install it now?", + message="step-cli not found. Do you want to install it now?", style=DEFAULT_QY_STYLE, ).ask() if answer: diff --git a/step_cli_tools/operations.py b/step_cli_tools/operations.py index b2ca8c4..0e23528 100644 --- a/step_cli_tools/operations.py +++ b/step_cli_tools/operations.py @@ -5,7 +5,6 @@ # --- Third-party imports --- from rich.panel import Panel - # --- Local application imports --- from .common import * from .configuration import * @@ -20,13 +19,9 @@ def operation1(): """ - Perform the CA bootstrap operation. - - Prompts the user for the CA server and fingerprint, then executes - the step-ca bootstrap command. + Install a root certificate in the system trust store. - Returns: - None + Prompt the user for the CA server and (optionally) root CA fingerprint, then execute the step-ca bootstrap command. """ warning_text = ( @@ -47,7 +42,7 @@ def operation1(): ).ask() if not ca_input or not ca_input.strip(): - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return # Parse host and port @@ -82,7 +77,7 @@ def operation1(): # Ask the user if they would like to use this fingerprint or enter it manually console.print() use_fingerprint = qy.confirm( - message=f"Continue with installation of this root CA? (Abort to enter the fingerprint manually)", + message="Continue with installation of this root CA? (Abort to enter the fingerprint manually)", style=DEFAULT_QY_STYLE, ).ask() @@ -98,7 +93,7 @@ def operation1(): ).ask() # Check for empty input if not fingerprint or not fingerprint.strip(): - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return # step-cli expects the fingerprint without colons fingerprint = fingerprint.replace(":", "") @@ -112,15 +107,14 @@ def operation1(): elif system == "Linux": cert_info = find_linux_cert_by_sha256(fingerprint) else: - console.print( - f"[WARNING] Could not check for existing certificates on unsupported platform: {system}", - style="#F9ED69", + logger.warning( + f"Could not check for existing certificates on unsupported platform: {system}" ) + # Confirm overwrite if cert_info: - console.print( - f"[INFO] Certificate with fingerprint '{fingerprint}' already exists in the system trust store.", - style="#F08A5D", + logger.info( + f"Certificate with fingerprint '{fingerprint}' already exists in the system trust store." ) console.print() overwrite_certificate = qy.confirm( @@ -129,7 +123,7 @@ def operation1(): style=DEFAULT_QY_STYLE, ).ask() if not overwrite_certificate: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return # Run step-ca bootstrap @@ -146,8 +140,8 @@ def operation1(): result = execute_step_command(bootstrap_args, STEP_BIN) if isinstance(result, str): - console.print( - "[NOTE] You may need to restart your system for the changes to take full effect." + logger.info( + "You may need to restart your system for the changes to take full effect." ) @@ -155,11 +149,8 @@ def operation2(): """ Uninstall a root CA certificate from the system trust store. - Prompts the user for the certificate fingerprint or a search term and removes it from + Prompt the user for the certificate fingerprint or a search term and remove it from the appropriate trust store based on the platform. - - Returns: - None """ warning_text = ( @@ -179,7 +170,7 @@ def operation2(): # Check for empty input if not fingerprint_or_search_term or not fingerprint_or_search_term.strip(): - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return fingerprint_or_search_term = fingerprint_or_search_term.replace(":", "").strip() @@ -199,18 +190,16 @@ def operation2(): if fingerprint: cert_info = find_windows_cert_by_sha256(fingerprint) if not cert_info: - console.print( - f"[ERROR] No certificate with fingerprint '{fingerprint}' was found in the Windows user ROOT trust store.", - style="#B83B5E", + logger.error( + f"No certificate with fingerprint '{fingerprint}' was found in the Windows user ROOT trust store." ) return elif search_term: certs_info = find_windows_certs_by_name(search_term) if not certs_info: - console.print( - f"[ERROR] No certificates matching '{search_term}' were found in the Windows user ROOT trust store.", - style="#B83B5E", + logger.error( + f"No certificates matching '{search_term}' were found in the Windows user ROOT trust store." ) return @@ -224,28 +213,26 @@ def operation2(): ) if not cert_info: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return thumbprint, cn = cert_info - delete_windows_cert_by_sha256(thumbprint, cn) + delete_windows_cert_by_thumbprint(thumbprint, cn) elif system == "Linux": if fingerprint: cert_info = find_linux_cert_by_sha256(fingerprint) if not cert_info: - console.print( - f"[ERROR] No certificate with fingerprint '{fingerprint}' was found in the Linux trust store.", - style="#B83B5E", + logger.error( + f"No certificate with fingerprint '{fingerprint}' was found in the Linux trust store." ) return elif search_term: certs_info = find_linux_certs_by_name(search_term) if not certs_info: - console.print( - f"[ERROR] No certificates matching '{search_term}' were found in the Linux trust store.", - style="#B83B5E", + logger.error( + f"No certificates matching '{search_term}' were found in the Linux trust store." ) return @@ -259,14 +246,11 @@ def operation2(): ) if not cert_info: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return cert_path, cn = cert_info delete_linux_cert_by_path(cert_path, cn) else: - console.print( - f"[ERROR] Unsupported platform for this operation: {system}", - style="#B83B5E", - ) + logger.error(f"Unsupported platform for this operation: {system}") diff --git a/step_cli_tools/support_functions.py b/step_cli_tools/support_functions.py index 05c5296..6b74941 100644 --- a/step_cli_tools/support_functions.py +++ b/step_cli_tools/support_functions.py @@ -1,26 +1,27 @@ # --- Standard library imports --- +import base64 import json import os import platform -import shutil import re +import shutil import ssl import subprocess import tarfile import tempfile import time -import urllib.error import warnings from pathlib import Path from urllib.request import urlopen +import urllib.error from zipfile import ZipFile # --- Third-party imports --- from cryptography import x509 -from cryptography.x509.oid import NameOID from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes from cryptography.utils import CryptographyDeprecationWarning +from cryptography.x509.oid import NameOID from packaging import version # --- Local application imports --- @@ -38,73 +39,110 @@ "find_windows_certs_by_name", "find_linux_cert_by_sha256", "find_linux_certs_by_name", - "delete_windows_cert_by_sha256", + "delete_windows_cert_by_thumbprint", "delete_linux_cert_by_path", "choose_cert_from_list", ] def check_for_update( - current_version: str, include_prerelease: bool = False + pkg_name: str, current_pkg_version: str, include_prerelease: bool = False ) -> str | None: - """Check PyPI for newer releases of the package. + """ + Check PyPI for newer releases of the package. Args: - current_version: Current version string of the package. + pkg_name: Name of the package. + current_pkg_version: Current version string of the package. include_prerelease: Whether to consider pre-release versions. Returns: The latest version string if a newer version exists, otherwise None. """ - pkg = "step-cli-tools" - cache = Path.home() / f".{pkg}" / ".cache" / "update_check.json" + cache = Path.home() / f".{pkg_name}" / ".cache" / "update_check.json" cache.parent.mkdir(parents=True, exist_ok=True) now = time.time() + current_parsed_version = version.parse(current_pkg_version) + logger.debug(locals()) + + # Try reading from cache if cache.exists(): try: - data = json.loads(cache.read_text()) + with cache.open("r", encoding="utf-8") as file: + data = json.load(file) + latest_version = data.get("latest_version") cache_lifetime = int( config.get("update_config.check_for_updates_cache_lifetime_seconds") ) + if ( latest_version and now - data.get("time", 0) < cache_lifetime - and version.parse(latest_version) > version.parse(current_version) + and version.parse(latest_version) > current_parsed_version ): + logger.debug("Returning newer version from cache") return latest_version - except json.JSONDecodeError: - pass - try: - with urlopen(f"https://pypi.org/pypi/{pkg}/json", timeout=5) as r: - data = json.load(r) - releases = [r for r, files in data["releases"].items() if files] + except (json.JSONDecodeError, OSError) as e: + logger.debug(f"Failed to read update cache: {e}") + # Fetch the latest releases from PyPI when the cache is empty, expired, or the cached version is older than the current version + try: + logger.debug("Fetching release metadata from PyPI") + with urlopen(f"https://pypi.org/pypi/{pkg_name}/json", timeout=5) as response: + data = json.load(response) + + # Filter releases (exclude ones with yanked files) + releases = [ + ver + for ver, files in data["releases"].items() + if files and all(not file.get("yanked", False) for file in files) + ] + + # Exclude pre-releases if not requested if not include_prerelease: releases = [r for r in releases if not version.parse(r).is_prerelease] if not releases: + logger.debug("No valid releases found") return None latest_version = max(releases, key=version.parse) - cache.write_text(json.dumps({"time": now, "latest_version": latest_version})) + latest_parsed_version = version.parse(latest_version) + + logger.debug(f"Latest available version on PyPI: {latest_version}") - if version.parse(latest_version) > version.parse(current_version): + # Write cache + try: + with cache.open("w", encoding="utf-8") as file: + json.dump({"time": now, "latest_version": latest_version}, file) + except OSError as e: + logger.debug(f"Failed to write update cache: {e}") + + if latest_parsed_version > current_parsed_version: + logger.debug(f"Update available: {latest_version}") return latest_version - except Exception: + except Exception as e: + logger.debug(f"Update check failed: {e}") return None def install_step_cli(step_bin: str): - """Download and install the step CLI binary for the current platform.""" + """ + Download and install the step-cli binary for the current platform. + + Args: + step_bin: Path to the step binary. + """ system = platform.system() arch = platform.machine() - console.print(f"[INFO] Detected platform: {system} {arch}") + logger.info(f"Detected platform: {system} {arch}") + logger.info(f"Target installation path: {step_bin}") if system == "Windows": url = "https://github.com/smallstep/cli/releases/latest/download/step_windows_amd64.zip" @@ -116,16 +154,19 @@ def install_step_cli(step_bin: str): url = "https://github.com/smallstep/cli/releases/latest/download/step_darwin_amd64.tar.gz" archive_type = "tar.gz" else: - console.print(f"[ERROR] Unsupported platform: {system}", style="#B83B5E") + logger.error(f"Unsupported platform: {system}") return tmp_dir = tempfile.mkdtemp() tmp_path = os.path.join(tmp_dir, os.path.basename(url)) - console.print(f"[INFO] Downloading step CLI from {url}...") + logger.info(f"Downloading step-cli from '{url}'...") + with urlopen(url) as response, open(tmp_path, "wb") as out_file: out_file.write(response.read()) - console.print(f"[INFO] Extracting {archive_type} archive...") + logger.debug(f"Archive downloaded to temporary path: {tmp_path}") + + logger.info(f"Extracting '{archive_type}' archive...") if archive_type == "zip": with ZipFile(tmp_path, "r") as zip_ref: zip_ref.extractall(tmp_dir) @@ -133,22 +174,21 @@ def install_step_cli(step_bin: str): with tarfile.open(tmp_path, "r:gz") as tar_ref: tar_ref.extractall(tmp_dir) - step_bin_name = "step.exe" if system == "Windows" else "step" + step_bin_name = os.path.basename(step_bin) # Search recursively for the binary matches = [] - for root, dirs, files in os.walk(tmp_dir): + for root, _, files in os.walk(tmp_dir): if step_bin_name in files: - matches.append(os.path.join(root, step_bin_name)) + found_path = os.path.join(root, step_bin_name) + matches.append(found_path) if not matches: - console.print( - f"[ERROR] Could not find {step_bin_name} in the extracted archive.", - style="#B83B5E", - ) + logger.error(f"Could not find '{step_bin_name}' in the extracted archive.") return extracted_path = matches[0] # Take the first found binary + logger.debug(f"Using extracted binary: {extracted_path}") # Prepare installation path binary_dir = os.path.dirname(step_bin) @@ -156,25 +196,27 @@ def install_step_cli(step_bin: str): # Delete old binary if exists if os.path.exists(step_bin): + logger.debug("Removing existing step binary") os.remove(step_bin) shutil.move(extracted_path, step_bin) os.chmod(step_bin, 0o755) - console.print(f"[INFO] step CLI installed: {step_bin}") + logger.info(f"step-cli installed: {step_bin}") try: result = subprocess.run([step_bin, "version"], capture_output=True, text=True) - console.print(f"[INFO] Installed step version:\n{result.stdout.strip()}") + logger.info(f"Installed step version:\n{result.stdout.strip()}") except Exception as e: - console.print(f"[ERROR] Failed to run step CLI: {e}", style="#B83B5E") + logger.error(f"Failed to run step-cli: {e}") -def execute_step_command(args, step_bin: str, interactive: bool = False): - """Execute a step CLI command and return output or log errors. +def execute_step_command(args, step_bin: str, interactive: bool = False) -> str | None: + """ + Execute a step-cli command and return output or log errors. Args: - args: List of command arguments to pass to step CLI. + args: List of command arguments to pass to step-cli. step_bin: Path to the step binary. interactive: If True, run the command interactively without capturing output. @@ -182,33 +224,34 @@ def execute_step_command(args, step_bin: str, interactive: bool = False): Command output as a string if successful, otherwise None. """ + logger.debug(locals()) + if not step_bin or not os.path.exists(step_bin): - console.print( - "[ERROR] step CLI not found. Please install it first.", style="#B83B5E" - ) + logger.error("step-cli not found. Please install it first.") return None try: if interactive: result = subprocess.run([step_bin] + args) + logger.debug(f"step-cli command exit code: {result.returncode}") + if result.returncode != 0: - console.print( - f"[ERROR] step command failed with exit code {result.returncode}", - style="#B83B5E", - ) + logger.error(f"step-cli command exit code: {result.returncode}") return None + return "" else: result = subprocess.run([step_bin] + args, capture_output=True, text=True) + logger.debug(f"step-cli command exit code: {result.returncode}") + if result.returncode != 0: - console.print( - f"[ERROR] step command failed: {result.stderr.strip()}", - style="#B83B5E", - ) + logger.error(f"step-cli command failed: {result.stderr.strip()}") return None + return result.stdout.strip() + except Exception as e: - console.print(f"[ERROR] Failed to execute step command: {e}", style="#B83B5E") + logger.error(f"Failed to execute step-cli command: {e}") return None @@ -220,12 +263,20 @@ def execute_ca_request( """ Perform an HTTPS request to the CA, handling untrusted certificates if needed. + Args: + url: URL to request. + trust_unknown_default: If True, trust unverified SSL certificates. + timeout: Timeout in seconds. + Returns: - Response body as string, or None on failure or user abort + Response body as string, or None on failure or user abort. """ + logger.debug(locals()) + def do_request(context): with urlopen(url, context=context, timeout=timeout) as response: + logger.debug(f"Received HTTP response status code: {response.status}") return response.read().decode("utf-8").strip() context = ( @@ -240,11 +291,10 @@ def do_request(context): except urllib.error.URLError as e: reason = getattr(e, "reason", None) + logger.debug(f"URLError: {e}") + if isinstance(reason, ssl.SSLCertVerificationError): - console.print( - "[WARNING] Server provided an unknown or self-signed certificate.", - style="#F9ED69", - ) + logger.warning("Server provided an unknown or self-signed certificate.") console.print() answer = qy.confirm( @@ -254,34 +304,44 @@ def do_request(context): ).ask() if not answer: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return None + logger.debug("Retrying request with unverified SSL context") + try: return do_request(ssl._create_unverified_context()) except Exception as retry_error: - console.print( - f"[ERROR] Retry failed: {retry_error}\n\nIs the port correct and the server available?", - style="#B83B5E", + logger.error( + f"Retry failed: {retry_error}\n\nIs the port correct and the server available?" ) return None - console.print( - f"[ERROR] Connection failed: {e}\n\nIs the port correct and the server available?", - style="#B83B5E", + logger.error( + f"Connection failed: {e}\n\nIs the port correct and the server available?" ) return None except Exception as e: - console.print( - f"[ERROR] Request failed: {e}\n\nIs the port correct and the server available?", - style="#B83B5E", + logger.error( + f"Request failed: {e}\n\nIs the port correct and the server available?" ) return None def check_ca_health(ca_base_url: str, trust_unknown_default: bool = False) -> bool: - """Check the health endpoint of a CA server via HTTPS.""" + """ + Check the health endpoint of a CA server via HTTPS. + + Args: + ca_base_url: Base URL of the CA server, including protocol and port. + trust_unknown_default: If True, trust unverified SSL certificates. + + Returns: + True if the CA is healthy, False otherwise. + """ + + logger.debug(locals()) health_url = ca_base_url.rstrip("/") + "/health" @@ -291,16 +351,16 @@ def check_ca_health(ca_base_url: str, trust_unknown_default: bool = False) -> bo ) if response is None: + logger.debug("CA health check failed due to missing response") return False + logger.debug(f"Health endpoint response: {response}") + if "ok" in response.lower(): - console.print(f"[INFO] CA at '{ca_base_url}' is healthy.", style="green") + logger.info(f"CA at '{ca_base_url}' is healthy.") return True - console.print( - f"[ERROR] CA health check failed for '{ca_base_url}'.", - style="#B83B5E", - ) + logger.error(f"CA health check failed for '{ca_base_url}'.") return False @@ -313,13 +373,15 @@ def get_ca_root_info( and SHA256 fingerprint. Args: - ca_base_url: Base URL of the CA (e.g. https://my-ca-host:9000) - trust_unknown_default: Skip SSL verification immediately if True + ca_base_url: Base URL of the CA (e.g. https://my-ca-host:9000). + trust_unknown_default: Skip SSL verification immediately if True. Returns: - CARootInfo on success, None on error or user cancel + CARootInfo on success, None on error or user cancel. """ + logger.debug(locals()) + roots_url = ca_base_url.rstrip("/") + "/roots.pem" pem_bundle = execute_ca_request( @@ -328,6 +390,7 @@ def get_ca_root_info( ) if pem_bundle is None: + logger.debug("Failed to retrieve roots.pem") return None try: @@ -338,9 +401,10 @@ def get_ca_root_info( re.S, ) if not match: - console.print("[ERROR] No certificate found in roots.pem", style="#B83B5E") + logger.error("No certificate found in roots.pem") return None + logger.debug("Loading PEM certificate") cert = x509.load_pem_x509_certificate( match.group(0).encode(), default_backend(), @@ -353,6 +417,8 @@ def get_ca_root_info( ) # Extract CA name (CN preferred, always string) + logger.debug(f"Computed SHA256 fingerprint: {fingerprint}") + try: cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME) ca_name = ( @@ -361,16 +427,10 @@ def get_ca_root_info( else str(cert.subject.rfc4514_string()) ) except Exception as e: - console.print( - f"[WARNING] Unable to retrieve CA name: {e}", - style="#F9ED69", - ) + logger.warning(f"Unable to retrieve CA name: {e}") ca_name = "Unknown CA" - console.print( - "[INFO] Root CA information retrieved successfully.", - style="green", - ) + logger.info("Root CA information retrieved successfully.") return CARootInfo( ca_name=ca_name, @@ -378,10 +438,7 @@ def get_ca_root_info( ) except Exception as e: - console.print( - f"[ERROR] Failed to process CA root certificate: {e}", - style="#B83B5E", - ) + logger.error(f"Failed to process CA root certificate: {e}") return None @@ -400,6 +457,8 @@ def find_windows_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | No Returns None if no matching certificate is found or if the query fails. """ + logger.debug(f"Starting Windows certificate search by SHA256: {sha256_fingerprint}") + ps_cmd = r""" $sha = [System.Security.Cryptography.SHA256]::Create() $store = New-Object System.Security.Cryptography.X509Certificates.X509Store "Root","CurrentUser" @@ -424,23 +483,28 @@ def find_windows_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | No errors="replace", ) + logger.debug(f"PowerShell output: {result.stdout}") + logger.debug(f"PowerShell stderr: {result.stderr}") + logger.debug(f"PowerShell exit code: {result.returncode}") + if result.returncode != 0: - console.print( - f"[ERROR] Failed to query certificates: {result.stderr.strip()}", - style="#B83B5E", - ) + logger.error(f"Failed to query certificates: {result.stderr.strip()}") return None + normalized_fp = sha256_fingerprint.lower().replace(":", "") + for line in result.stdout.strip().splitlines(): try: obj = json.loads(line) - if obj["Sha256"].strip().lower() == sha256_fingerprint.lower().replace( - ":", "" - ): + logger.debug(f"Processing certificate subject: {obj.get('Subject')}") + if obj["Sha256"].strip().lower() == normalized_fp: + logger.debug("Matching certificate found") return (obj["Thumbprint"].strip(), obj["Subject"].strip()) except (ValueError, KeyError, json.JSONDecodeError): + logger.debug("Skipping invalid or malformed certificate entry") continue + logger.debug("No matching Windows certificate found") return None @@ -451,12 +515,14 @@ def find_windows_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: each component like CN=..., OU=..., O=..., C=... Args: - name_pattern: Name or partial name to search (wildcard * allowed) + name_pattern: Name or partial name to search (wildcard * allowed). Returns: - List of tuples (thumbprint, subject) for all matching certificates + List of tuples (thumbprint, subject) for all matching certificates. """ + logger.debug(f"Starting Windows certificate search by name pattern: {name_pattern}") + ps_cmd = r""" $store = New-Object System.Security.Cryptography.X509Certificates.X509Store "Root","CurrentUser" $store.Open([System.Security.Cryptography.X509Certificates.OpenFlags]::ReadOnly) @@ -477,8 +543,12 @@ def find_windows_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: errors="replace", ) + logger.debug(f"PowerShell output: {result.stdout}") + logger.debug(f"PowerShell stderr: {result.stderr}") + logger.debug(f"PowerShell exit code: {result.returncode}") + if result.returncode != 0: - console.print(f"[ERROR] Failed to query certificates: {result.stderr.strip()}") + logger.error(f"Failed to query certificates: {result.stderr.strip()}") return [] # Convert wildcard * to regex @@ -493,18 +563,23 @@ def find_windows_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: thumbprint = obj["Thumbprint"].strip() subject = obj["Subject"].strip() + logger.debug(f"Evaluating certificate subject: {subject}") + components = [comp.strip() for comp in subject.split(",")] for comp in components: # Delete leading CN=, O=, OU=, etc. match = re.match(r"^(?:CN|O|OU|C|DC)=(.*)$", comp, re.IGNORECASE) value = match.group(1).strip() if match else comp if pattern_re.match(value): + logger.debug("Name pattern matched certificate") matches.append((thumbprint, subject)) - break # Search the next certificate if a match is found + break except (ValueError, KeyError, json.JSONDecodeError): + logger.debug("Skipping invalid or malformed certificate entry") continue + logger.debug(f"Total matching Windows certificates found: {len(matches)}") return matches @@ -514,7 +589,7 @@ def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None Args: sha256_fingerprint: SHA256 fingerprint of the certificate to search for. - Can include colons or be in uppercase/lowercase. + Can include colons and may be in uppercase/lowercase. Returns: A tuple (path, subject) of the matching certificate if found: @@ -523,11 +598,13 @@ def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None Returns None if no matching certificate is found or if the trust store directory is missing. """ + logger.debug(f"Starting Linux certificate search by SHA256: {sha256_fingerprint}") + cert_dir = "/etc/ssl/certs" fingerprint = sha256_fingerprint.lower().replace(":", "") if not os.path.isdir(cert_dir): - console.print(f"[ERROR] Cert directory not found: {cert_dir}", style="#B83B5E") + logger.error(f"Cert directory not found: {cert_dir}") return None # Ignore deprecation warnings about non-positive serial numbers @@ -538,6 +615,7 @@ def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None path = os.path.join(cert_dir, cert_file) if os.path.isfile(path): try: + logger.debug(f"Reading certificate file: {path}") with open(path, "rb") as f: cert_data = f.read() try: @@ -552,10 +630,13 @@ def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None ) fp = cert.fingerprint(hashes.SHA256()).hex() if fp.lower() == fingerprint: + logger.debug("Matching Linux certificate found") return (path, cert.subject.rfc4514_string()) - except Exception: + except Exception as e: + logger.debug(f"Failed to process certificate file '{path}': {e}") continue + logger.debug("No matching Linux certificate found") return None @@ -567,15 +648,17 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: Duplicates of the same certificate (e.g. from different files / symlinks) are ignored. Args: - name_pattern: Name or partial name to search (wildcard * allowed) + name_pattern: Name or partial name to search (wildcard * allowed). Returns: - List of tuples (path, subject) for all matching certificates + List of tuples (path, subject) for all matching certificates. """ + logger.debug(f"Starting Linux certificate search by name pattern: {name_pattern}") + cert_dir = "/etc/ssl/certs" if not os.path.isdir(cert_dir): - console.print(f"[ERROR] Cert directory not found: {cert_dir}") + logger.error(f"Cert directory not found: {cert_dir}") return [] # Convert wildcard * to regex @@ -598,9 +681,12 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: # Skip duplicate certificates pointing to the same real file if real_path in seen_real_paths: + logger.debug(f"Skipping duplicate certificate path: {real_path}") continue seen_real_paths.add(real_path) + logger.debug(f"Processing certificate file: {path}") + with open(path, "rb") as f: cert_data = f.read() try: @@ -622,24 +708,30 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: ) value = match.group(1).strip() if match else comp if pattern_re.match(value): + logger.debug("Name pattern matched certificate") matches.append((path, subject_str)) break - except Exception: + except Exception as e: + logger.debug(f"Failed to process certificate file '{path}': {e}") continue + logger.debug(f"Total matching Linux certificates found: {len(matches)}") return matches -def delete_windows_cert_by_sha256(thumbprint: str, cn: str): +def delete_windows_cert_by_thumbprint(thumbprint: str, cn: str, elevated: bool = False): """ - Delete a certificate from the Windows user ROOT store using certutil. + Delete a certificate from the Windows user ROOT store using PowerShell. Args: thumbprint: Thumbprint of the certificate to delete. cn: Common Name (CN) of the certificate for display purposes. + elevated: Whether to execute the PowerShell command with elevated privileges. """ + logger.debug(locals()) + console.print() answer = qy.confirm( message=f"Do you really want to remove the certificate: '{cn}'?", @@ -647,39 +739,141 @@ def delete_windows_cert_by_sha256(thumbprint: str, cn: str): style=DEFAULT_QY_STYLE, ).ask() if not answer: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return - # Validate thumbprint format + # Validate thumbprint format (SHA-1, 40 hex chars) if not re.fullmatch(r"[A-Fa-f0-9]{40}", thumbprint): - console.print( - f"[ERROR] Invalid thumbprint format: {thumbprint}", style="#B83B5E" - ) + logger.error(f"Invalid thumbprint format: {thumbprint}") return - delete_cmd = ["certutil", "-delstore", "-user", "ROOT", thumbprint] - result = subprocess.run(delete_cmd, capture_output=True, text=True) - if result.returncode == 0: - console.print(f"[INFO] Certificate '{cn}' removed from Windows ROOT store.") - console.print( - "[NOTE] You may need to restart your system for the changes to take full effect." - ) + ps_cmd = f""" + Import-Module Microsoft.PowerShell.Security -RequiredVersion 3.0.0.0 + $certPath = "Cert:\\CurrentUser\\Root\\{thumbprint}" + if (-not (Test-Path -Path $certPath)) {{ + exit 1 + }} + try {{ + Remove-Item -Path $certPath -ErrorAction Stop + exit 0 + }} + catch {{ + # Access denied + if ($_.Exception.NativeErrorCode -eq 5) {{ + exit 2 + }} + # User cancelled + if ($_.Exception.NativeErrorCode -eq 1223) {{ + exit 3 + }} + exit 4 + }} + """ + ps_cmd_encoded = base64.b64encode(ps_cmd.encode("utf-16le")).decode("ascii") + + if elevated: + ps_args = [ + "powershell", + "-NoProfile", + "-Command", + # Capture the exit code and pass it through + f""" + $proc = Start-Process powershell -WindowStyle Hidden -ArgumentList '-NoProfile','-EncodedCommand','{ps_cmd_encoded}' -Verb RunAs -Wait -PassThru; + exit $proc.ExitCode + """, + ] else: - console.print( - f"[ERROR] Failed to remove certificate: {result.stderr.strip()}", - style="#B83B5E", + ps_args = [ + "powershell", + "-NoProfile", + "-EncodedCommand", + ps_cmd_encoded, + ] + + logger.debug(f"PowerShell command: {' '.join(ps_args)}") + + result = subprocess.run( + ps_args, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + + logger.debug(f"PowerShell output: {result.stdout}") + logger.debug(f"PowerShell stderr: {result.stderr}") + logger.debug(f"PowerShell exit code: {result.returncode}") + + if result.returncode == 0: + logger.info(f"Certificate '{cn}' removed from Windows ROOT store.") + logger.info( + "You may need to restart your system for the changes to take full effect." ) + return + + if result.returncode == 1: + logger.warning(f"Certificate '{cn}' not found.") + return + + # Access denied, offer to retry with elevated privileges + if result.returncode == 2: + logger.warning(f"Access denied to remove certificate '{cn}'.") + console.print() + retry_with_admin_privileges = qy.confirm( + message="Retry with elevated privileges?", style=DEFAULT_QY_STYLE + ).ask() + if not retry_with_admin_privileges: + logger.info("Operation cancelled by user.") + return None + + delete_windows_cert_by_thumbprint(thumbprint, cn, elevated=True) + return + + if result.returncode == 3: + logger.info("Operation cancelled by user.") + return + + logger.error(f"Failed to remove certificate with thumbprint '{thumbprint}'") -def delete_linux_cert_by_path(cert_path: str, cn: str): +def delete_linux_cert_by_path(cert_path: str, cn: str, elevated: bool = False): """ Delete a certificate from the Linux system trust store. Args: cert_path: Full path to the certificate symlink in /etc/ssl/certs. cn: Common Name (CN) of the certificate for display purposes. + elevated: Whether to execute commands with elevated privileges. """ + cert_path_obj = Path(cert_path) + local_dir = Path("/usr/local/share/ca-certificates").resolve() + package_dir = Path("/usr/share/ca-certificates").resolve() + ca_conf_path = Path("/etc/ca-certificates.conf") + + logger.debug(locals()) + + def run_cmd(args: list[str], input: str | None = None): + cmd = ["sudo", *args] if elevated else args + logger.debug(f"Running command: {' '.join(cmd)}") + result = subprocess.run( + cmd, + input=input, + capture_output=True, + text=True, + check=True, + encoding="utf-8", + errors="replace", + ) + logger.debug(f"Command output: {result.stdout}") + logger.debug(f"Command stderr: {result.stderr}") + logger.debug(f"Command exit code: {result.returncode}") + return result + + def confirm_retry(message: str) -> bool: + console.print() + return qy.confirm(message=message, default=True, style=DEFAULT_QY_STYLE).ask() + console.print() answer = qy.confirm( message=f"Do you really want to remove the certificate: '{cn}'?", @@ -687,62 +881,121 @@ def delete_linux_cert_by_path(cert_path: str, cn: str): style=DEFAULT_QY_STYLE, ).ask() if not answer: - console.print("[INFO] Operation cancelled by user.") + logger.info("Operation cancelled by user.") return - try: - cert_dir = Path("/etc/ssl/certs").resolve() - source_dir = Path("/usr/local/share/ca-certificates").resolve() + if not cert_path_obj.is_symlink(): + logger.warning(f"'{cert_path}' is not a symlink, skipping.") + return - cert_path_obj = Path(cert_path) + target_path = cert_path_obj.resolve() + logger.debug(f"Resolved symlink target: {target_path}") - # Handle symlink target - if cert_path_obj.is_symlink(): - target_path = cert_path_obj.resolve() + try: + # Handle local certificates + if target_path.is_relative_to(local_dir): + try: + if not elevated: + target_path.touch(exist_ok=True) + except PermissionError: + logger.warning(f"No write access to '{target_path}' detected.") + if confirm_retry("Retry with elevated privileges?"): + return delete_linux_cert_by_path(cert_path, cn, elevated=True) + logger.info("Operation cancelled by user.") + return + run_cmd(["rm", str(target_path)]) + logger.info(f"Removed locally installed CA certificate '{cn}'.") + + # Handle package certificates + elif target_path.is_relative_to(package_dir): + relative_cert = target_path.relative_to(package_dir) + logger.debug(f"Certificate originates from package store: {relative_cert}") + + if not ca_conf_path.exists(): + logger.error(f"CA configuration file '{ca_conf_path}' does not exist.") + return - if target_path.is_relative_to(source_dir): - subprocess.run(["sudo", "rm", str(target_path)], check=True) - else: - console.print( - f"[WARNING] Symlink target '{target_path}' is outside {source_dir}, skipping deletion.", - style="#F9ED69", + try: + if not elevated: + ca_conf_path.touch(exist_ok=True) + except PermissionError: + logger.warning(f"No write access to '{ca_conf_path}' detected.") + if confirm_retry("Retry with elevated privileges?"): + return delete_linux_cert_by_path(cert_path, cn, elevated=True) + logger.info("Operation cancelled by user.") + return + + # Disable the certificate in the configuration file + lines = ca_conf_path.read_text(encoding="utf-8").splitlines() + updated_lines, found, disabled = [], False, False + for line in lines: + stripped = line.lstrip("!").strip() + if stripped == str(relative_cert): + found = True + if not line.startswith("!"): + updated_lines.append(f"!{relative_cert}") + disabled = True + else: + updated_lines.append(line) + logger.debug(f"CA '{cn}' already disabled") + else: + updated_lines.append(line) + + if not found: + logger.warning( + f"Certificate '{cn}' not found in '{ca_conf_path}'. It may already be disabled or managed externally." ) + return + + backup_path = ca_conf_path.with_suffix(".conf.bak") + run_cmd(["cp", str(ca_conf_path), str(backup_path)]) + logger.info(f"Backup saved as '{backup_path}'.") + run_cmd(["tee", str(ca_conf_path)], input="\n".join(updated_lines) + "\n") + # Show the log message once the file has been updated + if disabled: + logger.info(f"Disabled CA '{cn}' in '{ca_conf_path}'.") - # Delete the symlink itself if it lives inside /etc/ssl/certs - if cert_path_obj.parent.resolve().is_relative_to(cert_dir): - subprocess.run(["sudo", "rm", str(cert_path_obj)], check=True) else: - console.print( - f"[WARNING] Certificate path '{cert_path_obj}' is outside {cert_dir}, skipping deletion.", - style="#F9ED69", + logger.warning( + f"Symlink target '{target_path}' is outside known CA source directories, skipping source modification." ) - subprocess.run(["sudo", "update-ca-certificates", "--fresh"], check=True) - - console.print(f"[INFO] Certificate '{cn}' removed from Linux trust store.") - console.print( - "[NOTE] You may need to restart your system for the changes to take full effect." + run_cmd(["update-ca-certificates", "--fresh"]) + logger.info(f"Certificate '{cn}' removed from Linux trust store.") + logger.info( + "You may need to restart your system for the changes to take full effect." ) except subprocess.CalledProcessError as e: - console.print(f"[ERROR] Failed to remove certificate: {e}", style="#B83B5E") + logger.debug(f"Command stdout: {e.stdout}") + logger.debug(f"Command stderr: {e.stderr}") + if not elevated and confirm_retry("Retry with elevated privileges?"): + return delete_linux_cert_by_path(cert_path, cn, elevated=True) + logger.warning( + f"Could not remove certificate '{cn}'. Operation cancelled." + if not elevated + else f"Failed to remove certificate '{cn}'." + ) def choose_cert_from_list( certs: list[tuple[str, str]], message: str = "Select a certificate:" ) -> tuple[str, str] | None: """ - Presents a alphabetically sorted list of certificates to the user and returns the chosen tuple (fingerprint/path, subject). + Presents an alphabetically sorted list of certificates to the user and returns the chosen tuple (fingerprint/path, subject). Args: - certs: List of tuples (id, subject) to choose from - message: message text for the questionary select + certs: List of tuples (id, subject) to choose from. + message: message text for the questionary select. Returns: - The selected tuple or None if user cancels + The selected tuple or None if user cancels. """ + logger.debug(f"Presenting certificate selection list with {len(certs)} entries") + if not certs: + logger.debug("No certificates available for selection") return None # Sort certificates alphabetically by subject (case-insensitive) @@ -761,11 +1014,16 @@ def choose_cert_from_list( ).ask() if selected_subject is None: + logger.debug("User cancelled certificate selection") return None # Return the full tuple matching the selected subject for cert in sorted_certs: if cert[1] == selected_subject: + logger.debug( + f"User selected a certificate with subject: {selected_subject}" + ) return cert + logger.debug("Selected certificate not found in internal list") return None diff --git a/step_cli_tools/validators.py b/step_cli_tools/validators.py index b964453..d3136ad 100644 --- a/step_cli_tools/validators.py +++ b/step_cli_tools/validators.py @@ -1,6 +1,10 @@ +# --- Standard library imports --- import ipaddress import re -from questionary import Validator, ValidationError + +# --- Third-party imports --- +from questionary import ValidationError, Validator + __all__ = [ "HostnamePortValidator",