diff --git a/.gitignore b/.gitignore index b520c87..626940d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,3 @@ /dist -step_cli_tools/__pycache__ +__pycache__/ poetry.lock diff --git a/README.md b/README.md index a95a560..324501a 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,8 @@ sct | Feature | Description | |---------|-------------| -| ๐Ÿ“œ **Manage Root CA Certificates** | Install & uninstall your **root CA certificate** easily | +| ๐Ÿ“œ **Manage** root CA certificates | Install & uninstall your root CA certificate easily | +| ๐Ÿ“ **Request** certificates | Request TLS certificates from your step-ca server | โ„น๏ธ More features are planned. diff --git a/assets/readme.gif b/assets/readme.gif index 6ac0438..1a04517 100644 Binary files a/assets/readme.gif and b/assets/readme.gif differ diff --git a/assets/readme.tape b/assets/readme.tape index 4b5c8ef..bc7f64f 100644 --- a/assets/readme.tape +++ b/assets/readme.tape @@ -1,5 +1,6 @@ Output readme.gif +Require powershell Require sct Set Shell powershell @@ -13,7 +14,9 @@ Enter Sleep 5s -# Select the first operation + + +# Install root CA certificate operation Enter Sleep 3s @@ -30,7 +33,68 @@ Type@200ms y Sleep 5s -# Select the second operation + + +# Request certificate operation +Down@200ms 2 +Sleep 3s +Enter + +Sleep 3s + +# Enter the server name +Type@200ms my-step-ca-server +Sleep 1s +Enter + +Sleep 3s + +# Change subject name +Enter +Sleep 3s +Backspace@200ms 9 +Sleep 1s +Type@200ms demo.com +Sleep 1s +Enter + +Sleep 3s + +# Change SAN entries +Down@200ms 2 +Sleep 1s +Enter +Sleep 1s +Down +Sleep 1s +Enter +Sleep 1s +Type@200ms demonstration.com +Sleep 1s +Enter +Sleep 3s +Down@200ms 2 +Sleep 1s +Enter + +Sleep 5s + +# Proceed with request +Down@200ms 5 +Sleep 1s +Enter +Sleep 1s +Enter +Sleep 1s +# A published password!? :D +Type oCYsOa1NMUWhrU7EZHGHLHQ4i09hAO80JfONwzTV +Enter + +Sleep 5s + + + +# Uninstall root CA certificate operation Down Sleep 3s Enter diff --git a/pyproject.toml b/pyproject.toml index 78b03b9..2f5e2bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,10 +35,12 @@ packages = [ ] [tool.poetry.dependencies] -python = ">=3.10,<4.0" +# Can be unlocked once https://github.com/tmbo/questionary/issues/473 is resolved cryptography = "^46.0.3" packaging = "^25.0" -questionary = "^2.1.1" +python = ">=3.10,<4.0" +questionary = "==2.1.1" +requests = "^2.32.5" rich = "^14.2.0" ruamel-yaml = "^0.19.1" diff --git a/step_cli_tools/common.py b/step_cli_tools/common.py index 4cde2c8..89bc0a7 100644 --- a/step_cli_tools/common.py +++ b/step_cli_tools/common.py @@ -1,8 +1,9 @@ # --- Standard library imports --- import logging -from logging.handlers import RotatingFileHandler -import os import platform +from logging.handlers import RotatingFileHandler +from pathlib import Path +from urllib.parse import urlparse # --- Third-party imports --- import questionary @@ -10,17 +11,6 @@ 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", -] - - custom_logging_theme = Theme( { "logging.level.info": "none", @@ -42,28 +32,37 @@ ) # --- 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")) +SCRIPT_HOME_DIR = Path.home() / ".step-cli-tools" +SCRIPT_CACHE_DIR = SCRIPT_HOME_DIR / ".cache" +SCRIPT_CERT_DIR = SCRIPT_HOME_DIR / "certs" +SCRIPT_LOGGING_DIR = SCRIPT_HOME_DIR / "logs" + +ALL_DIRS = [ + SCRIPT_HOME_DIR, + SCRIPT_CACHE_DIR, + SCRIPT_CERT_DIR, + SCRIPT_LOGGING_DIR, +] -def _get_step_binary_path() -> str: +def _get_step_binary_path() -> Path: """ Get the absolute path to the step-cli binary based on the operating system. Returns: - str: Absolute path to the step binary. + The absolute path to the step-cli binary. """ - bin_dir = os.path.join(SCRIPT_HOME_DIR, "bin") + bin_dir = SCRIPT_HOME_DIR / "bin" system = platform.system() if system == "Windows": - binary = os.path.join(bin_dir, "step.exe") + binary = bin_dir / "step.exe" elif system in ("Linux", "Darwin"): - binary = os.path.join(bin_dir, "step") + binary = bin_dir / "step" else: raise OSError(f"Unsupported platform: {system}") - return os.path.normpath(binary) + return binary STEP_BIN = _get_step_binary_path() @@ -73,8 +72,8 @@ def _get_step_binary_path() -> str: def _setup_logger( name: str, - log_file: str = "step-cli-tools.log", - level=logging.DEBUG, + log_file: Path = SCRIPT_LOGGING_DIR / "step-cli-tools.log", + level: int = logging.DEBUG, console: Console = console, max_bytes: int = 5_000_000, backup_count: int = 5, @@ -95,8 +94,8 @@ def _setup_logger( """ # Ensure log directory exists - if os.path.dirname(log_file): - os.makedirs(os.path.dirname(log_file), exist_ok=True) + if log_file.parent: + log_file.parent.mkdir(parents=True, exist_ok=True) logger = logging.getLogger(name) logger.setLevel(level) @@ -128,5 +127,25 @@ def _setup_logger( logger = _setup_logger( name="main", - log_file=os.path.join(SCRIPT_LOGGING_DIR, "step-cli-tools.log"), ) + + +def get_masked_url_for_logging(url: str) -> str: + """ + Return a masked version of the URL suitable for logging. + + Args: + url: The URL to mask + + Returns: + A string containing only the scheme, hostname, and optional port + """ + + parsed_url = urlparse(url) + scheme = parsed_url.scheme or "unknown" + hostname = parsed_url.hostname or "unknown" + port = parsed_url.port + + if port: + return f"{scheme}://{hostname}:{port}" + return f"{scheme}://{hostname}" diff --git a/step_cli_tools/configuration.py b/step_cli_tools/configuration.py index 1f8676e..e8767e3 100644 --- a/step_cli_tools/configuration.py +++ b/step_cli_tools/configuration.py @@ -1,28 +1,29 @@ # --- 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 functools import partial +from logging.handlers import RotatingFileHandler +from pathlib import Path # --- Third-party imports --- from rich.logging import RichHandler from ruamel.yaml import YAML from ruamel.yaml.comments import CommentedMap +from ruamel.yaml.scalarstring import SingleQuotedScalarString # --- Local application imports --- -from .common import * -from .validators import * - -__all__ = [ - "config", - "check_and_repair_config_file", - "show_config_operations", -] - +from .common import DEFAULT_QY_STYLE, SCRIPT_HOME_DIR, console, logger, qy +from .utils.validators import ( + bool_validator, + certificate_subject_name_validator, + hostname_or_ip_address_and_optional_port_validator, + int_range_validator, + str_allowed_validator, +) yaml = YAML() yaml.indent(mapping=2, sequence=4, offset=2) @@ -30,7 +31,7 @@ class Configuration: - def __init__(self, file_location: str, schema: dict, autosave: bool = True): + def __init__(self, file_location: Path, schema: dict, autosave: bool = True): """ Initialize Configuration object. Note, that the load() method MUST be called manually once. @@ -39,7 +40,8 @@ def __init__(self, file_location: str, schema: dict, autosave: bool = True): schema: Dictionary defining the config schema with types, defaults, validators, and comments. autosave: Automatically save after each set() call if True. """ - self.file_location = Path(file_location) + + self.file_location = file_location self.file_location.parent.mkdir(parents=True, exist_ok=True) self.schema = schema self.autosave = autosave @@ -48,6 +50,7 @@ def __init__(self, file_location: str, schema: dict, autosave: bool = True): # --- File and public API handling --- def load(self): """Load YAML config and merge defaults into a CommentedMap with comments.""" + if self.file_location.exists(): try: loaded = yaml.load(self.file_location.read_text()) or {} @@ -61,6 +64,7 @@ def load(self): def save(self): """Save current configuration data to YAML file.""" + try: with self.file_location.open("w", encoding="utf-8") as f: yaml.dump(self._data, f) @@ -74,6 +78,7 @@ def generate_default(self, overwrite: bool = False): Args: overwrite: If True, existing file will be replaced. Otherwise, existing file will be kept. """ + try: if self.file_location.exists() and not overwrite: logger.warning( @@ -83,16 +88,16 @@ def generate_default(self, overwrite: bool = False): # Backup existing file before overwriting if self.file_location.exists() and overwrite: - timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S-%f") backup_path = self.file_location.with_name( f"{self.file_location.stem}_backup_{timestamp}{self.file_location.suffix}" ) shutil.copy2(self.file_location, backup_path) - logger.info(f"Created backup before overwrite: {backup_path}") + logger.info(f"Created backup before overwrite at '{backup_path}'.") - # This is a bit akward but the file is technically repaired without keeping any data. + # This is a bit akward but the file is technically repaired without keeping any data default_data = self._build_commented_data( - self.schema, repair_damaged_keys=True, suppress_repair_messages=True + self.schema, repair=True, log_repair=False ) # Save YAML file @@ -100,7 +105,7 @@ def generate_default(self, overwrite: bool = False): yaml.dump(default_data, f) logger.info( - f"Default configuration file was generated successfully: {self.file_location}" + f"Generated default configuration file at '{self.file_location}'." ) # Load the data into memory so it's ready for use @@ -130,6 +135,7 @@ def get(self, key: str): Returns: The current value or the schema default if missing. """ + parts = key.split(".") data = self._data for part in parts: @@ -148,6 +154,7 @@ def set(self, key: str, value): key: Dotted path to the setting. value: Value to set. """ + parts = key.split(".") data = self._data @@ -181,9 +188,8 @@ def set(self, key: str, value): def repair(self): """Restore missing keys and values from the schema and optionally autosave.""" - self._data = self._build_commented_data( - self.schema, self._data, repair_damaged_keys=True - ) + + self._data = self._build_commented_data(self.schema, self._data, repair=True) if self.autosave: self.save() @@ -196,6 +202,7 @@ def validate(self, key: str | None = None) -> bool: Returns: True if all checked values are valid, False otherwise. """ + if key: parts = key.split(".") meta = self._nested_get_meta(parts) @@ -248,6 +255,7 @@ def _wrap_validator(self, meta, validator): Returns: Callable validator with parameters applied if applicable. """ + if callable(validator): if validator is int_range_validator and "min" in meta and "max" in meta: return int_range_validator(meta["min"], meta["max"]) @@ -266,6 +274,7 @@ def _validate_recursive(self, data: dict, schema: dict, prefix: str) -> bool: Returns: True if all values valid, False otherwise. """ + ok = True for k, meta in schema.items(): if not isinstance(meta, dict): @@ -316,57 +325,100 @@ def _validate_recursive(self, data: dict, schema: dict, prefix: str) -> bool: def _nested_get_meta(self, keys: list[str]) -> dict | None: """Retrieve schema metadata for nested key path.""" + data = self.schema for k in keys: if not isinstance(data, dict) or k not in data: - return None + return data = data[k] return data if isinstance(data, dict) else None def _nested_get_default(self, keys: list[str]): """Retrieve default value from schema for nested key path.""" + data = self.schema for k in keys: if not isinstance(data, dict) or k not in data: logger.warning(f"Missing default for key '{'.'.join(keys)}'.") - return None + return data = data[k] if isinstance(data, dict) and "default" in data: return data["default"] - return None + return def _nested_get_type(self, keys: list[str]): """Retrieve expected type from schema for nested key path.""" + data = self.schema for k in keys: if not isinstance(data, dict) or k not in data: - return None + return data = data[k] return data.get("type") if isinstance(data, dict) else None - def _build_commented_data( + def _merge_schema_data(self, schema: dict, data: dict | None) -> dict: + """Merge schema structure with existing data.""" + + data = data or {} + result = {} + + for key, meta in schema.items(): + if not isinstance(meta, dict): + continue + + # Section + if "type" not in meta: + result[key] = self._merge_schema_data(meta, data.get(key, {})) + else: + result[key] = data.get(key) + + return result + + def _repair_data( self, schema: dict, - data: dict | None = None, + data: dict, + *, + log: bool = True, + ) -> dict: + """Restore missing or invalid keys using schema defaults.""" + + repaired = {} + + for key, meta in schema.items(): + if not isinstance(meta, dict): + continue + + # Section + if "type" not in meta: + repaired[key] = self._repair_data( + meta, + data.get(key, {}), + log=log, + ) + continue + + value = data.get(key) + + if value is None: + if log: + logger.info(f"Repairing key '{key}' from config schema.") + repaired[key] = meta.get("default") + else: + repaired[key] = value + + return repaired + + def _annotate_yaml( + self, + schema: dict, + data: dict, + *, indent: int = 0, top_level: bool = True, - repair_damaged_keys=False, - suppress_repair_messages=False, ) -> CommentedMap: - """Construct a CommentedMap with schema defaults and YAML comments. + """Convert dict into CommentedMap with YAML comments.""" - Args: - schema: Schema defining keys, types, defaults, validators, and comments. - data: Optional existing config data to merge. - indent: Indentation level for YAML comments. - top_level: True if building top-level mapping. - repair_damaged_keys: Restore missing keys from schema if True. - suppress_repair_messages: Suppress log messages from repairing keys if True. - - Returns: - CommentedMap populated with data and comments. - """ - data = data or {} node = CommentedMap() for i, (key, meta) in enumerate(schema.items()): @@ -374,25 +426,20 @@ def _build_commented_data( continue if "type" not in meta: - child_node = self._build_commented_data( + node[key] = self._annotate_yaml( meta, data.get(key, {}), - indent + 2, + indent=indent + 2, top_level=False, - repair_damaged_keys=repair_damaged_keys, - suppress_repair_messages=suppress_repair_messages, ) - node[key] = child_node else: - node[key] = data.get(key) - # The data could not be extracted from the config file - if node[key] is None: - if repair_damaged_keys: - if not suppress_repair_messages: - logger.info(f"Repairing key '{key}' from config schema.") - node[key] = meta.get("default") - else: - continue + value_to_set = data.get(key) + + # Wrap strings with single quotes for YAML + if isinstance(value_to_set, str): + value_to_set = SingleQuotedScalarString(value_to_set) + + node[key] = value_to_set type_obj = meta.get("type") type_name = type_obj.__name__ if type_obj else "unknown" @@ -404,13 +451,11 @@ def _build_commented_data( if allowed: type_info = f"[{type_name}: allowed: {', '.join(map(str, allowed))} | default: {default_val}]" elif min_val is not None or max_val is not None: - range_part = "" - if min_val is not None and max_val is not None: - range_part = f"{min_val} - {max_val}" - elif min_val is not None: - range_part = f">= {min_val}" - elif max_val is not None: - range_part = f"<= {max_val}" + range_part = ( + f"{min_val} - {max_val}" + if min_val is not None and max_val is not None + else f">= {min_val}" if min_val is not None else f"<= {max_val}" + ) type_info = f"[{type_name}: {range_part} | default: {default_val}]" else: type_info = f"[{type_name} | default: {default_val}]" @@ -419,33 +464,41 @@ def _build_commented_data( final_comment = ( f"{type_info} - {extra_comment}" if extra_comment else type_info ) + node.yaml_set_comment_before_after_key( key, before=final_comment, indent=indent ) - # Leave an empty line between top level keys + # Top-level spacing if top_level and i > 0: - node.yaml_set_comment_before_after_key( - key, - before="\n" - + ( - node.ca.items.get(key)[2].value - if node.ca.items.get(key) and node.ca.items.get(key)[2] - else "" - ), - indent=indent, - ) + node.yaml_set_comment_before_after_key(key, before="\n", indent=indent) return node + def _build_commented_data( + self, + schema: dict, + data: dict | None = None, + *, + repair: bool = False, + log_repair: bool = True, + ) -> CommentedMap: + """Convert data dict into CommentedMap with YAML comments.""" + + merged = self._merge_schema_data(schema, data) + + if repair: + merged = self._repair_data(schema, merged, log=log_repair) + + return self._annotate_yaml(schema, merged) + def check_and_repair_config_file(): """Ensure the config file exists and is valid. Allow repair/edit/reset if invalid.""" # Generate default config if missing - if not os.path.exists(config.file_location): + if not config.file_location.exists(): config.generate_default() - logger.info("A default config file has been generated.") automatic_repair_failed = False @@ -470,8 +523,10 @@ def check_and_repair_config_file(): # In case the automatic repair fails console.print() selected_action = qy.select( - message="Choose an action:", + message="Choose an action", choices=["Edit config file", "Reset config file"], + use_search_filter=True, + use_jk_keys=False, style=DEFAULT_QY_STYLE, ).ask() @@ -511,7 +566,7 @@ def show_config_operations(): # Prompt user to select an operation console.print() selected_operation = qy.select( - message="Config file operation:", + message="Config file operation", choices=config_operations, use_search_filter=True, use_jk_keys=False, @@ -563,11 +618,13 @@ def let_user_change_config_file(reset_instead_of_discard: bool = False): logger.error("Configuration is invalid.") console.print() selected_action = qy.select( - message="Choose an action:", + message="Choose an action", choices=[ "Edit again", "Reset config file" if reset_instead_of_discard else "Discard changes", ], + use_search_filter=True, + use_jk_keys=False, style=DEFAULT_QY_STYLE, ).ask() @@ -585,7 +642,7 @@ def let_user_change_config_file(reset_instead_of_discard: bool = False): # else: loop continues for "Edit again" -def open_in_editor(file_path: str | Path): +def open_in_editor(file_path: Path): """ Open the given file in the user's preferred text editor and wait until it is closed. @@ -595,7 +652,7 @@ def open_in_editor(file_path: str | Path): - On Linux: tries common editors (nano, vim) or falls back to xdg-open (non-blocking). """ - path = Path(file_path).expanduser().resolve() + path = file_path.expanduser().resolve() if not path.exists(): raise FileNotFoundError(f"File not found: {path}") @@ -604,33 +661,33 @@ def open_in_editor(file_path: str | Path): # --- Windows --- if platform.system() == "Windows": if editor: - subprocess.run([editor, str(path)], check=False) + subprocess.run([editor, path], check=False) else: # notepad blocks until file is closed - subprocess.run(["notepad", str(path)], check=False) + subprocess.run(["notepad", path], check=False) return # --- macOS --- if platform.system() == "Darwin": if editor: - subprocess.run([editor, str(path)], check=False) + subprocess.run([editor, path], check=False) else: # `open -W` waits until the app is closed - subprocess.run(["open", "-W", "-t", str(path)], check=False) + subprocess.run(["open", "-W", "-t", path], check=False) return # --- Linux / Unix --- if platform.system() == "Linux": if editor: - subprocess.run([editor, str(path)], check=False) + subprocess.run([editor, path], check=False) return # try common console editors for candidate in ["nano", "vim", "vi"]: if shutil.which(candidate): - subprocess.run([candidate, str(path)], check=False) + subprocess.run([candidate, path], check=False) return # fallback: GUI open (non-blocking) - subprocess.Popen(["xdg-open", str(path)]) + subprocess.Popen(["xdg-open", path]) logger.info("File opened in default GUI editor. Please close it manually.") input("Press Enter here when you're done editing...") @@ -660,7 +717,7 @@ def reset_with_feedback(): # --- Config file defintions --- -config_file_location = os.path.join(SCRIPT_HOME_DIR, "config.yml") +config_file_location = SCRIPT_HOME_DIR / "config.yml" config_schema = { "update_config": { "check_for_updates_at_launch": { @@ -688,22 +745,27 @@ def reset_with_feedback(): "default_ca_server": { "type": str, "default": "", - "validator": server_validator, - "comment": "The CA server which will be used by default (optionally with :port)", - }, - "trust_unknow_certificates_by_default": { - "type": bool, - "default": False, - "validator": bool_validator, - "comment": "If true, any CA server providing an unknown self-signed certificate will be trusted by default", + "validator": partial( + hostname_or_ip_address_and_optional_port_validator, accept_blank=True + ), + "comment": "The step-ca server which will be used by default (optionally with :port)", }, "fetch_root_ca_certificate_automatically": { "type": bool, "default": True, "validator": bool_validator, - "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", + "comment": "If false, the root certificate won't be fetched automatically from the step-ca server. You will need to enter the fingerprint manually when installing a root CA certificate", }, }, + "certificate_request_config": { + "default_subject_name": { + "type": str, + "default": "change.me", + "validator": certificate_subject_name_validator, + "comment": "The default subject name for certificate requests", + } + # The remaining certificate request configurations are WIP and will be added in a future release + }, "logging_config": { "log_level_console": { "type": str, diff --git a/step_cli_tools/data_classes.py b/step_cli_tools/data_classes.py deleted file mode 100644 index 06f8719..0000000 --- a/step_cli_tools/data_classes.py +++ /dev/null @@ -1,10 +0,0 @@ -# --- Standard library imports --- -from dataclasses import dataclass - -__all__ = ["CARootInfo"] - - -@dataclass(frozen=True) -class CARootInfo: - ca_name: str - fingerprint_sha256: str diff --git a/step_cli_tools/main.py b/step_cli_tools/main.py index 1063da6..e5694b8 100644 --- a/step_cli_tools/main.py +++ b/step_cli_tools/main.py @@ -1,24 +1,15 @@ # --- Standard library imports --- -import os 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__))) - sys.path.insert(0, parent_dir) - __package__ = "step_cli_tools" - # --- Local application imports --- -from .common import * -from .configuration import * -from .operations import * -from .support_functions import * - -# --- Main function --- +from .common import ALL_DIRS, DEFAULT_QY_STYLE, STEP_BIN, console, logger, qy +from .configuration import check_and_repair_config_file, config, show_config_operations +from .operations import operation1, operation2, operation3 +from .utils.general import check_for_update, install_step_cli def main(): @@ -38,6 +29,11 @@ def main(): textArray = ["", "#" * len(bannerText), bannerText, "#" * len(bannerText), ""] for text in textArray: logger.info(text) + # Ensure necessary directories exist + for directory in ALL_DIRS: + if not directory.exists(): + logger.debug(f"Creating directory: {directory}") + directory.mkdir(parents=True) # Unmute console logging for handler in logger.handlers: if isinstance(handler, RichHandler): @@ -97,10 +93,10 @@ def main(): console.print(version_text) # Ensure step-cli is installed - if not os.path.exists(STEP_BIN): + if not STEP_BIN.exists(): console.print() answer = qy.confirm( - message="step-cli not found. Do you want to install it now?", + message="step-cli binary not found. Do you want to download it now?", style=DEFAULT_QY_STYLE, ).ask() if answer: @@ -111,18 +107,23 @@ def main(): # Define operations and their corresponding functions operations = [ qy.Choice( - title="Install root CA on the system", - description="Add a root certificate of your step-ca server into the system trust store.", + title="Install Root CA", + description="Add a root certificate of your step-ca server to the system trust store.", value=operation1, ), qy.Choice( - title="Uninstall root CA from the system (Windows & Linux)", + title="Uninstall Root CA (Windows & Linux)", description="Delete a root certificate (of your step-ca server) from the system trust store.", value=operation2, ), + qy.Choice( + title="Request Certificate", + description="Request a new certificate from your step-ca server.", + value=operation3, + ), qy.Choice( title="Configuration", - description="View and edit the configuration file.", + description="View and edit the config file.", value=show_config_operations, ), qy.Choice( @@ -135,7 +136,7 @@ def main(): console.print() # Prompt the user to select an operation selected_operation = qy.select( - message="Operation:", + message="Operation", choices=operations, use_search_filter=True, use_jk_keys=False, diff --git a/step_cli_tools/models/data.py b/step_cli_tools/models/data.py new file mode 100644 index 0000000..7b5d10f --- /dev/null +++ b/step_cli_tools/models/data.py @@ -0,0 +1,293 @@ +# --- Standard library imports --- +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path + +# --- Local application imports --- +from ..common import SCRIPT_CERT_DIR, logger +from ..utils.paths import sanitize_filename + +# --- Root CA Info --- + + +@dataclass(frozen=True) +class RootCAInfo: + ca_name: str + fingerprint_sha256: str + + +# --- Certificate Request Info --- + + +@dataclass +class CRI_OutputFormat_Information: + """Information about supported certificate output formats which is used for questionary prompts.""" + + menu_item_name: str + menu_item_description: str + + +class CRI_OutputFormat(Enum): + """Supported certificate output formats.""" + + PEM_CRT_KEY = CRI_OutputFormat_Information( + "PEM CRT & KEY (default)", "Two separate PEM-encoded crt and key files." + ) + PEM_BUNDLE = CRI_OutputFormat_Information( + "PEM BUNDLE", "One PEM-encoded crt bundle file containing the crt and key." + ) + PFX_BUNDLE = CRI_OutputFormat_Information( + "PFX BUNDLE", "One PFX-encoded bundle file containing the crt and key." + ) + + +@dataclass +class CRI_KeyAlgorithm_Information: + """Information about supported certificate key algorithms which is used for questionary prompts.""" + + # The argument used for the actual step-cli command (should not be changed) + arg: str + menu_item_name: str + menu_item_description: str + + +class CRI_KeyAlgorithm(Enum): + """Supported certificate key algorithms.""" + + EC = CRI_KeyAlgorithm_Information( + "EC", + "EC (default)", + "Elliptic Curve; efficient and secure for most modern applications.", + ) + RSA = CRI_KeyAlgorithm_Information( + "RSA", "RSA", "Traditional RSA algorithm; widely supported." + ) + OKP = CRI_KeyAlgorithm_Information( + "OKP", + "OKP", + "Octet Key Pair; uses modern elliptic curves for high-performance cryptography.", + ) + + +@dataclass +class CRI_ECCurve_Information: + """Information about supported elliptic curves which is used for questionary prompts.""" + + # The argument used for the actual step-cli command (should not be changed) + arg: str + menu_item_name: str + menu_item_description: str + + +class CRI_ECCurve(Enum): + """Supported elliptic curves for EC keys.""" + + P256 = CRI_ECCurve_Information( + "P-256", + "P-256 (default)", + "Also known as secp256r1; widely supported and secure for most use cases.", + ) + P384 = CRI_ECCurve_Information( + "P-384", + "P-384", + "Provides higher security than P-256 with a longer key length.", + ) + P521 = CRI_ECCurve_Information( + "P-521", "P-521", "Very strong security, but slower and less widely supported." + ) + + +@dataclass +class CRI_RSAKeySize_Information: + """Information about supported RSA key sizes which is used for questionary prompts.""" + + # The argument used for the actual step-cli command (should not be changed) + arg: str + menu_item_name: str + menu_item_description: str + + +class CRI_RSAKeySize(Enum): + """Supported key sizes for RSA keys.""" + + RSA2048 = CRI_RSAKeySize_Information( + "2048", "2048 (default)", "Standard key size; secure for most applications." + ) + RSA3072 = CRI_RSAKeySize_Information( + "3072", + "3072", + "Stronger security than 2048-bit, with moderate performance cost.", + ) + RSA4096 = CRI_RSAKeySize_Information( + "4096", + "4096", + "Very strong security; slower but highly secure for sensitive use cases.", + ) + + +@dataclass +class CRI_OKPCurve_Information: + """Information about supported OKP elliptic curves which is used for questionary prompts.""" + + # The argument used for the actual step-cli command (should not be changed) + arg: str + menu_item_name: str + menu_item_description: str + + +class CRI_OKPCurve(Enum): + """Supported elliptic curves for OKP keys.""" + + Ed25519 = CRI_OKPCurve_Information( + "Ed25519", + "Ed25519 (default)", + "High-performance signature algorithm with strong security.", + ) + + +@dataclass +class CertificateRequestInfo: + """Data model for a basic certificate request.""" + + # Backing field for subject_name (do not access directly) + _subject_name: str = field(repr=False) + + output_format: CRI_OutputFormat + + # SAN entries (always available after __post_init__) + san_entries: list[str] = field(default_factory=list) + + # Key options + key_algorithm: CRI_KeyAlgorithm = CRI_KeyAlgorithm.EC + ecc_curve: CRI_ECCurve = CRI_ECCurve.P256 + rsa_size: CRI_RSAKeySize = CRI_RSAKeySize.RSA2048 + okp_curve: CRI_OKPCurve = CRI_OKPCurve.Ed25519 + + # Optional validity overrides + valid_since: datetime | None = None + valid_until: datetime | None = None + + # Derived fields (always set in __post_init__ / setter) + final_output_dir: Path = field(init=False) + final_crt_output_name_with_suffix: Path = field(init=False) + final_key_output_name_with_suffix: Path = field(init=False) + final_pem_bundle_output_name_with_suffix: Path = field(init=False) + final_pfx_bundle_output_name_with_suffix: Path = field(init=False) + + def __post_init__(self): + # Generate timestamp + self.timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S-%f") + + # Output folder with timestamp + self.final_output_dir = SCRIPT_CERT_DIR / self.timestamp + + # Initialize subject_name via setter to trigger synchronization logic + initial_subject = self._subject_name + self.subject_name = initial_subject + + # Ensure timezone awareness + if self.valid_since and self.valid_since.tzinfo is None: + self.valid_since = self.valid_since.replace(tzinfo=timezone.utc) + logger.warning("Validity start date is not timezone aware, assuming UTC.") + + if self.valid_until and self.valid_until.tzinfo is None: + self.valid_until = self.valid_until.replace(tzinfo=timezone.utc) + logger.warning("Validity end date is not timezone aware, assuming UTC.") + + # Normalize key options + if self.key_algorithm == CRI_KeyAlgorithm.RSA: + self.rsa_size = self.rsa_size or CRI_RSAKeySize.RSA2048 + else: + self.ecc_curve = self.ecc_curve or CRI_ECCurve.P256 + + @property + def subject_name(self) -> str: + """Return the current subject name.""" + + return self._subject_name + + @subject_name.setter + def subject_name(self, value: str): + """ + Update subject_name and keep SAN entries and derived filenames in sync. + """ + + old_value = getattr(self, "_subject_name", None) + self._subject_name = value + + # Update SAN entries + if old_value is not None: + # Replace old subject name occurrences + self.san_entries = [ + value if entry == old_value else entry for entry in self.san_entries + ] + + # Ensure subject_name is always present in SAN entries + if value not in self.san_entries: + self.san_entries.append(value) + + # Update derived filenames + self._update_derived_filenames(value) + + def _update_derived_filenames(self, subject_name: str): + """Update all filename fields derived from subject_name.""" + + self.final_crt_output_name_with_suffix = Path( + sanitize_filename(f"{subject_name}.crt") + ) + self.final_key_output_name_with_suffix = Path( + sanitize_filename(f"{subject_name}.key") + ) + self.final_pem_bundle_output_name_with_suffix = Path( + sanitize_filename(f"{subject_name}.pem") + ) + self.final_pfx_bundle_output_name_with_suffix = Path( + sanitize_filename(f"{subject_name}.pfx") + ) + + def validate(self): + if not self.subject_name: + raise ValueError("Subject name must not be empty") + + if ( + self.valid_since + and self.valid_until + and self.valid_since > self.valid_until + ): + raise ValueError("Validity start date must be before end date") + + if self.valid_until and self.valid_until < datetime.now().astimezone(): + raise ValueError("Validity end date must be in the future") + + # Used to determine the correct user prompt and step-cli argument to pass + def is_key_algorithm_ec(self) -> bool: + return self.key_algorithm == CRI_KeyAlgorithm.EC + + def is_key_algorithm_rsa(self) -> bool: + return self.key_algorithm == CRI_KeyAlgorithm.RSA + + def is_key_algorithm_okp(self) -> bool: + return self.key_algorithm == CRI_KeyAlgorithm.OKP + + def is_output_format_pem(self) -> bool: + return self.output_format in { + CRI_OutputFormat.PEM_CRT_KEY, + CRI_OutputFormat.PEM_BUNDLE, + } + + def is_output_format_pfx(self) -> bool: + return self.output_format == CRI_OutputFormat.PFX_BUNDLE + + +# --- Functions --- + + +# Used by convert_certificate() +@dataclass +class CertificateConversionResult: + format: CRI_OutputFormat + certificate: Path | None = None + private_key: Path | None = None + pem_bundle: Path | None = None + pfx: Path | None = None diff --git a/step_cli_tools/models/select.py b/step_cli_tools/models/select.py new file mode 100644 index 0000000..b61e39b --- /dev/null +++ b/step_cli_tools/models/select.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- + +import string +from typing import Any, Dict, Optional, Sequence, Union + +from prompt_toolkit.application import Application +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.keys import Keys +from prompt_toolkit.styles import Style +from questionary import utils +from questionary.constants import DEFAULT_QUESTION_PREFIX, DEFAULT_SELECTED_POINTER +from questionary.prompts import common +from questionary.prompts.common import Choice, InquirerControl, Separator +from questionary.question import Question +from questionary.styles import merge_styles_default + + +def CUSTOMIZED_select( + message: str, + choices: Sequence[Union[str, Choice, Dict[str, Any]]], + default: Optional[Union[str, Choice, Dict[str, Any]]] = None, + qmark: str = DEFAULT_QUESTION_PREFIX, + pointer: Optional[str] = DEFAULT_SELECTED_POINTER, + style: Optional[Style] = None, + use_shortcuts: bool = False, + use_arrow_keys: bool = True, + use_indicator: bool = False, + use_jk_keys: bool = True, + use_emacs_keys: bool = True, + use_search_filter: bool = False, + show_selected: bool = False, + show_description: bool = True, + instruction: Optional[str] = None, + **kwargs: Any, +) -> Question: + """A list of items to select **one** option from. + + The user can pick one option and confirm it (if you want to allow + the user to select multiple options, use :meth:`questionary.checkbox` instead). + + Example: + >>> import questionary + >>> questionary.select( + ... "What do you want to do?", + ... choices=[ + ... "Order a pizza", + ... "Make a reservation", + ... "Ask for opening hours" + ... ]).ask() + ? What do you want to do? Order a pizza + 'Order a pizza' + + .. image:: ../images/select.gif + + This is just a really basic example, the prompt can be customised using the + parameters. + + + Args: + message: Question text + + choices: Items shown in the selection, this can contain :class:`Choice` or + or :class:`Separator` objects or simple items as strings. Passing + :class:`Choice` objects, allows you to configure the item more + (e.g. preselecting it or disabling it). + + default: A value corresponding to a selectable item in the choices, + to initially set the pointer position to. + + qmark: Question prefix displayed in front of the question. + By default this is a ``?``. + + pointer: Pointer symbol in front of the currently highlighted element. + By default this is a ``ยป``. + Use ``None`` to disable it. + + instruction: A hint on how to navigate the menu. + It's ``(Use shortcuts)`` if only ``use_shortcuts`` is set + to True, ``(Use arrow keys or shortcuts)`` if ``use_arrow_keys`` + & ``use_shortcuts`` are set and ``(Use arrow keys)`` by default. + + style: A custom color and style for the question parts. You can + configure colors as well as font types for different elements. + + use_indicator: Flag to enable the small indicator in front of the + list highlighting the current location of the selection + cursor. + + use_shortcuts: Allow the user to select items from the list using + shortcuts. The shortcuts will be displayed in front of + the list items. Arrow keys, j/k keys and shortcuts are + not mutually exclusive. + + use_arrow_keys: Allow the user to select items from the list using + arrow keys. Arrow keys, j/k keys and shortcuts are not + mutually exclusive. + + use_jk_keys: Allow the user to select items from the list using + `j` (down) and `k` (up) keys. Arrow keys, j/k keys and + shortcuts are not mutually exclusive. + + use_emacs_keys: Allow the user to select items from the list using + `Ctrl+N` (down) and `Ctrl+P` (up) keys. Arrow keys, j/k keys, + emacs keys and shortcuts are not mutually exclusive. + + use_search_filter: Flag to enable search filtering. Typing some string will + filter the choices to keep only the ones that contain the + search string. + Note that activating this option disables "vi-like" + navigation as "j" and "k" can be part of a prefix and + therefore cannot be used for navigation + + show_selected: Display current selection choice at the bottom of list. + + show_description: Display description of current selection if available. + + Returns: + :class:`Question`: Question instance, ready to be prompted (using ``.ask()``). + """ + if not (use_arrow_keys or use_shortcuts or use_jk_keys or use_emacs_keys): + raise ValueError( + ( + "Some option to move the selection is required. " + "Arrow keys, j/k keys, emacs keys, or shortcuts." + ) + ) + + if use_jk_keys and use_search_filter: + raise ValueError( + "Cannot use j/k keys with prefix filter search, since j/k can be part of the prefix." + ) + + if use_shortcuts and use_jk_keys: + if any(getattr(c, "shortcut_key", "") in ["j", "k"] for c in choices): + raise ValueError( + "A choice is trying to register j/k as a " + "shortcut key when they are in use as arrow keys " + "disable one or the other." + ) + + if choices is None or len(choices) == 0: + raise ValueError("A list of choices needs to be provided.") + + if use_shortcuts: + real_len_of_choices = sum(1 for c in choices if not isinstance(c, Separator)) + if real_len_of_choices > len(InquirerControl.SHORTCUT_KEYS): + raise ValueError( + "A list with shortcuts supports a maximum of {} " + "choices as this is the maximum number " + "of keyboard shortcuts that are available. You " + "provided {} choices!" + "".format(len(InquirerControl.SHORTCUT_KEYS), real_len_of_choices) + ) + + merged_style = merge_styles_default([style]) + + # CUSTOMIZED: 'default' is not passed to InquirerControl anymore + # See https://github.com/tmbo/questionary/issues/473 for details + ic = InquirerControl( + choices, + pointer=pointer, + use_indicator=use_indicator, + use_shortcuts=use_shortcuts, + show_selected=show_selected, + show_description=show_description, + use_arrow_keys=use_arrow_keys, + initial_choice=default, + ) + + def get_prompt_tokens(): + # noinspection PyListCreation + tokens = [("class:qmark", qmark), ("class:question", " {} ".format(message))] + + if ic.is_answered: + if isinstance(ic.get_pointed_at().title, list): + tokens.append( + ( + "class:answer", + "".join([token[1] for token in ic.get_pointed_at().title]), # type: ignore + ) + ) + else: + tokens.append(("class:answer", ic.get_pointed_at().title)) # type: ignore + else: + if instruction: + tokens.append(("class:instruction", instruction)) + else: + if use_shortcuts and use_arrow_keys: + instruction_msg = f"(Use shortcuts or arrow keys{', type to filter' if use_search_filter else ''})" + elif use_shortcuts and not use_arrow_keys: + instruction_msg = f"(Use shortcuts{', type to filter' if use_search_filter else ''})" + else: + instruction_msg = f"(Use arrow keys{', type to filter' if use_search_filter else ''})" + tokens.append(("class:instruction", instruction_msg)) + + return tokens + + layout = common.create_inquirer_layout(ic, get_prompt_tokens, **kwargs) + + bindings = KeyBindings() + + @bindings.add(Keys.ControlQ, eager=True) + @bindings.add(Keys.ControlC, eager=True) + def _(event): + event.app.exit(exception=KeyboardInterrupt, style="class:aborting") + + if use_shortcuts: + # add key bindings for choices + for i, c in enumerate(ic.choices): + if c.shortcut_key is None and not c.disabled and not use_arrow_keys: + raise RuntimeError( + "{} does not have a shortcut and arrow keys " + "for movement are disabled. " + "This choice is not reachable.".format(c.title) + ) + if isinstance(c, Separator) or c.shortcut_key is None or c.disabled: + continue + + # noinspection PyShadowingNames + def _reg_binding(i, keys): + # trick out late evaluation with a "function factory": + # https://stackoverflow.com/a/3431699 + @bindings.add(keys, eager=True) + def select_choice(event): + ic.pointed_at = i + + _reg_binding(i, c.shortcut_key) + + def move_cursor_down(event): + ic.select_next() + while not ic.is_selection_valid(): + ic.select_next() + + def move_cursor_up(event): + ic.select_previous() + while not ic.is_selection_valid(): + ic.select_previous() + + if use_search_filter: + + def search_filter(event): + ic.add_search_character(event.key_sequence[0].key) + + for character in string.printable: + bindings.add(character, eager=True)(search_filter) + bindings.add(Keys.Backspace, eager=True)(search_filter) + + if use_arrow_keys: + bindings.add(Keys.Down, eager=True)(move_cursor_down) + bindings.add(Keys.Up, eager=True)(move_cursor_up) + + if use_jk_keys: + bindings.add("j", eager=True)(move_cursor_down) + bindings.add("k", eager=True)(move_cursor_up) + + if use_emacs_keys: + bindings.add(Keys.ControlN, eager=True)(move_cursor_down) + bindings.add(Keys.ControlP, eager=True)(move_cursor_up) + + @bindings.add(Keys.ControlM, eager=True) + def set_answer(event): + ic.is_answered = True + event.app.exit(result=ic.get_pointed_at().value) + + @bindings.add(Keys.Any) + def other(event): + """Disallow inserting other text.""" + + return Question( + Application( + layout=layout, + key_bindings=bindings, + style=merged_style, + **utils.used_kwargs(kwargs, Application.__init__), + ) + ) diff --git a/step_cli_tools/operations.py b/step_cli_tools/operations.py index 0e23528..0edfb70 100644 --- a/step_cli_tools/operations.py +++ b/step_cli_tools/operations.py @@ -1,43 +1,69 @@ # --- Standard library imports --- import platform import re +from datetime import datetime +from enum import Enum +from typing import TypeVar # --- Third-party imports --- from rich.panel import Panel # --- Local application imports --- -from .common import * -from .configuration import * -from .support_functions import * -from .validators import * - -__all__ = [ - "operation1", - "operation2", -] +from .common import DEFAULT_QY_STYLE, SCRIPT_CERT_DIR, STEP_BIN, console, logger, qy +from .configuration import config +from .models.data import CertificateRequestInfo, CRI_OutputFormat +from .models.select import CUSTOMIZED_select +from .utils.ca import ( + execute_certificate_request, + get_ca_root_info, + is_step_ca_server_healthy, +) +from .utils.certificates import ( + choose_cert_from_list, + convert_certificate, + delete_linux_cert_by_path, + delete_windows_cert_by_thumbprint, + find_linux_cert_by_sha256, + find_linux_certs_by_name, + find_windows_cert_by_sha256, + find_windows_certs_by_name, +) +from .utils.general import execute_step_command, parse_date_str +from .utils.network import is_host_available, is_server_certificate_trusted +from .utils.paths import join_safe_path +from .utils.validators import ( + CertificateSubjectNameValidator, + DateTimeValidator, + HostnameOrIPAddressAndOptionalPortValidator, + SHA256OrNameValidator, + SHA256Validator, +) + +# Type variable for objects like CRI_OutputFormat, CRI_KeyAlgorithm, etc. +TEnum = TypeVar("TEnum", bound=Enum) def operation1(): """ Install a root certificate in the system trust store. - Prompt the user for the CA server and (optionally) root CA fingerprint, then execute the step-ca bootstrap command. + Prompt the user for the step-ca server and (optionally) root CA fingerprint, then execute the step-ca bootstrap command. """ warning_text = ( - "You are about to install a root CA on your system.\n" + "You are about to install a root CA certificate on your system.\n" "This may pose a potential security risk to your device.\n" "Make sure you fully [bold]trust the CA before proceeding![/bold]" ) console.print(Panel.fit(warning_text, title="WARNING", border_style="#F9ED69")) - # Ask for CA hostname/IP and port + # Ask for step-ca hostname/IP and port default = config.get("ca_server_config.default_ca_server") console.print() ca_input = qy.text( - message="Enter step CA server hostname or IP (optionally with :port)", + message="Enter step step-ca server hostname or IP (optionally with :port)", default=default, - validate=HostnamePortValidator, + validate=HostnameOrIPAddressAndOptionalPortValidator, style=DEFAULT_QY_STYLE, ).ask() @@ -50,18 +76,22 @@ def operation1(): port = int(port_str) if port_str else 9000 ca_base_url = f"https://{ca_server}:{port}" - # Run the health check via helper - trust_unknown_default = config.get( - "ca_server_config.trust_unknow_certificates_by_default" - ) - if not check_ca_health(ca_base_url, trust_unknown_default): - # Either failed or user cancelled + if not is_host_available(ca_base_url): + logger.error( + f"step-ca server at '{ca_base_url}' is not available.\n\nAre address and port correct?" + ) + return + + # Trust unknown certificates because the root certificate will be installed from this server + if not is_step_ca_server_healthy( + ca_base_url=ca_base_url, trust_unknown_certificates=True + ): return use_fingerprint = False if config.get("ca_server_config.fetch_root_ca_certificate_automatically"): # Get root certificate info - ca_root_info = get_ca_root_info(ca_base_url, trust_unknown_default) + ca_root_info = get_ca_root_info(ca_base_url, trust_unknown_certificates=True) if ca_root_info is None: return @@ -87,8 +117,8 @@ def operation1(): # Ask for fingerprint console.print() fingerprint = qy.text( - message="Enter root certificate fingerprint (SHA256, 64 hex chars)", - validate=SHA256Validator, + message="Enter root certificate fingerprint (SHA256, 64 hex chars, blank to abort)", + validate=SHA256Validator(accept_blank=True), style=DEFAULT_QY_STYLE, ).ask() # Check for empty input @@ -138,8 +168,8 @@ def operation1(): "--force", ] - result = execute_step_command(bootstrap_args, STEP_BIN) - if isinstance(result, str): + success, _ = execute_step_command(bootstrap_args, STEP_BIN) + if success: logger.info( "You may need to restart your system for the changes to take full effect." ) @@ -206,7 +236,7 @@ def operation2(): cert_info = ( choose_cert_from_list( certs_info, - "Multiple certificates were found. Please select the one to remove:", + "Multiple certificates were found. Please select the one to remove", ) if len(certs_info) > 1 else certs_info[0] @@ -239,7 +269,7 @@ def operation2(): cert_info = ( choose_cert_from_list( certs_info, - "Multiple certificates were found. Please select the one to remove:", + "Multiple certificates were found. Please select the one to remove", ) if len(certs_info) > 1 else certs_info[0] @@ -254,3 +284,561 @@ def operation2(): else: logger.error(f"Unsupported platform for this operation: {system}") + + +def operation3(): + """ + Request a new certificate from a step-ca server. + + Prompt the user for various certificate request parameters, request a new certificate and convert it to the desired format. + """ + + def _get_choices(enum_class: type[Enum]) -> list[qy.Choice]: + """ + Convert an Enum with 'menu_item_name' and 'menu_item_description' into + a list of questionary.Choice objects for selection prompts. + + Args: + enum_class: The Enum class to convert. + + Returns: + List of questionary.Choice objects. + """ + + choices = [] + for item in enum_class: + # Skip items without a proper name or description + if getattr(item.value, "menu_item_name", None) and getattr( + item.value, "menu_item_description", None + ): + choices.append( + qy.Choice( + title=item.value.menu_item_name, + description=item.value.menu_item_description, + value=item, + ) + ) + else: + logger.debug( + f"Skipping enum item '{item}' as it lacks property 'menu_item_name' or 'menu_item_description'" + ) + return choices + + def _select_enum_option(cri_enum_obj: TEnum, message: str) -> TEnum | None: + """ + Display a questionary select menu for an Enum class (usually CRI_OutputFormat or similar) and return the selected value. + + Args: + cri_enum_obj: An instance of CRI_OutputFormat or a similar Enum class + message: Prompt message for the selection + + Returns: + The selected Enum member, or None if cancelled + """ + + choices = _get_choices(type(cri_enum_obj)) + default_choice = next((c for c in choices if c.value == cri_enum_obj), None) + console.print() + selection = CUSTOMIZED_select( + message=message, + choices=choices, + default=default_choice, + use_search_filter=True, + use_jk_keys=False, + style=DEFAULT_QY_STYLE, + ).ask() + return selection + + def _prompt_for_password( + message: str = "Enter password", + confirm_message: str = "Confirm password", + max_attempts: int = 10, + ) -> str | None: + """ + Prompt the user for a password and confirm it. + + Args: + message: The message to display to the user. + confirm_message: The message to display to the user to confirm the password. + max_attempts: The maximum number of attempts to get a valid password. + + Returns: + The password if successful, None otherwise. + """ + + for attempt in range(max_attempts): + console.print() + password = qy.password(message=message, style=DEFAULT_QY_STYLE).ask() + if not password: + return + + console.print() + confirm = qy.password(message=confirm_message, style=DEFAULT_QY_STYLE).ask() + if not confirm: + return + + if password == confirm: + return password + + # If they don't match, ask if they want to try again + console.print() + retry = qy.confirm( + message="Inputs did not match. Try again?", + style=DEFAULT_QY_STYLE, + ).ask() + if not retry: + return + + logger.error(f"Failed to get password after {max_attempts} attempts.") + return + + # Ask for CA hostname/IP and port + default = config.get("ca_server_config.default_ca_server") + console.print() + ca_input = qy.text( + message="Enter step step-ca server hostname or IP (optionally with :port)", + default=default, + validate=HostnameOrIPAddressAndOptionalPortValidator, + style=DEFAULT_QY_STYLE, + ).ask() + + if not ca_input or not ca_input.strip(): + logger.info("Operation cancelled by user.") + return + + # Parse host and port + ca_server, _, port_str = ca_input.partition(":") + port = int(port_str) if port_str else 9000 + ca_base_url = f"https://{ca_server}:{port}" + + if not is_host_available(ca_base_url): + logger.error( + f"step-ca server at '{ca_base_url}' is not available.\n\nAre address and port correct?" + ) + return + + # Trust unknown certificates here because the trust check is performed below and we need to verify that the server is a step-ca server + if not is_step_ca_server_healthy( + ca_base_url=ca_base_url, trust_unknown_certificates=True + ): + return + + if not is_server_certificate_trusted(ca_base_url): + logger.info( + "step-cli requires the root certificate of the step-ca server to be trusted. Please install it to the system trust store first." + ) + return + + # Initialize CertificateRequestInfo with default values + cri = CertificateRequestInfo( + _subject_name=config.get("certificate_request_config.default_subject_name"), + output_format=CRI_OutputFormat.PEM_CRT_KEY, + ) + + # ---------------------------- + # Start of review/edit section + # ---------------------------- + + def _edit_san_entries(cri_obj: CertificateRequestInfo): + """ + Allow adding, editing, and removing SAN entries interactively. + Remembers last selected menu item to restore cursor position. + """ + + last_selected: str | None = None # Store last selection + + while True: + san_choices = [ + qy.Choice( + title=f"{i + 1}: {v}", + value=i, + description=( + "Automatically derived from subject name and cannot be edited." + if i == 0 + else None + ), + ) + for i, v in enumerate(cri_obj.san_entries) + ] + san_choices.append( + qy.Choice(title="Add", description="Add a new SAN entry.", value="add") + ) + # Only show "Clear All" if there are multiple entries + if len(cri_obj.san_entries) > 1: + san_choices.append( + qy.Choice( + title="Clear All", + description="Clear all additional SAN entries.", + value="clear", + ) + ) + elif last_selected == "clear": + # Reset if "Clear All" was last selected but is no longer available + last_selected = None + san_choices.append(qy.Choice(title="Save", value="save")) + + console.print() + choice = CUSTOMIZED_select( + message="Edit SAN entries", + choices=san_choices, + default=last_selected, # Restore cursor position + use_search_filter=True, + use_jk_keys=False, + style=DEFAULT_QY_STYLE, + ).ask() + + if not choice or choice == "save": + return + + # Store last selected value + last_selected = choice + + if choice == "add": + console.print() + new_san = qy.text( + message="Enter SAN entry (blank to cancel)", + validate=CertificateSubjectNameValidator(accept_blank=True), + style=DEFAULT_QY_STYLE, + ).ask() + if new_san and not new_san.strip() == "": + cri_obj.san_entries.append(new_san.strip()) + + elif choice == "clear": + console.print() + confirm = qy.confirm( + message="Clear all SAN entries?", + style=DEFAULT_QY_STYLE, + ).ask() + if confirm: + cri_obj.san_entries.clear() + # Re-add subject name to generate default SAN entry + cri_obj.subject_name = cri_obj.subject_name + + elif isinstance(choice, int): + current = cri_obj.san_entries[choice] + console.print() + edited = qy.text( + message="Edit SAN entry (blank to delete)", + default=current, + validate=CertificateSubjectNameValidator(accept_blank=True), + style=DEFAULT_QY_STYLE, + ).ask() + # User cancelled (allow empty input for deletion) + if edited is None: + continue + + edited_stripped = edited.strip() + # Delete if blank + if edited_stripped == "": + cri_obj.san_entries.pop(choice) + # Reset last selected if it was deleted + if last_selected == choice: + last_selected = None + else: + cri_obj.san_entries[choice] = edited_stripped + + def _review_and_edit(cri_obj: CertificateRequestInfo) -> bool: + """ + Review and optionally edit CertificateRequestInfo fields. + Remembers last selected position for main menu and submenus + """ + + last_selected_main: str | None = None + + while True: + # Build main menu + choices = [ + qy.Choice( + title=f"Subject Name: {cri_obj.subject_name}", + description="The main subject name for the certificate.", + value="subject_name", + ), + qy.Choice( + title=f"Output Format: {cri_obj.output_format.value.menu_item_name}", + description="Output format for the certificate file(s).", + value="output_format", + ), + qy.Choice( + title=f"SAN Entries: {len(cri_obj.san_entries)}", + description="The subject alternative names for the certificate.", + value="san_entries", + ), + qy.Choice( + title=f"Key Algorithm: {cri_obj.key_algorithm.value.menu_item_name}", + description="The algorithm used for the private key.", + value="key_algorithm", + ), + ] + + if cri_obj.is_key_algorithm_ec(): + choices.append( + qy.Choice( + title=f"EC Curve: {cri_obj.ecc_curve.value.menu_item_name}", + value="ecc_curve", + ) + ) + if cri_obj.is_key_algorithm_rsa(): + choices.append( + qy.Choice( + title=f"RSA Key Size: {cri_obj.rsa_size.value.menu_item_name}", + value="rsa_size", + ) + ) + if cri_obj.is_key_algorithm_okp(): + choices.append( + qy.Choice( + title=f"OKP Curve: {cri_obj.okp_curve.value.menu_item_name}", + value="okp_curve", + ) + ) + choices.extend( + [ + qy.Choice( + title=f"Valid Since: {cri_obj.valid_since if cri_obj.valid_since else 'CA default'}", + description="Start date/time for certificate validity.", + value="valid_since", + ), + qy.Choice( + title=f"Valid Until: {cri_obj.valid_until if cri_obj.valid_until else 'CA default'}", + description="End date/time for certificate validity.", + value="valid_until", + ), + qy.Choice(title="Proceed", value="proceed"), + qy.Choice(title="Exit", value="exit"), + ] + ) + + # Main menu selection + console.print() + answer = CUSTOMIZED_select( + message="Review and edit certificate request", + choices=choices, + default=last_selected_main, + use_search_filter=True, + use_jk_keys=False, + style=DEFAULT_QY_STYLE, + ).ask() + + if answer: + last_selected_main = answer + + if not answer or answer == "exit": + logger.info("Operation cancelled by user.") + return False + + if answer == "proceed": + try: + cri_obj.validate() + except Exception as e: + logger.error(f"Invalid certificate request configuration: {e}") + continue + return True + + # Submenu: Subject Name + if answer == "subject_name": + console.print() + value = qy.text( + message="Enter subject name", + default=cri_obj.subject_name, + validate=CertificateSubjectNameValidator, + style=DEFAULT_QY_STYLE, + ).ask() + if value: + cri_obj.subject_name = value.strip() + + # Submenu: Output Format + elif answer == "output_format": + value = _select_enum_option( + cri_obj.output_format, "Select output format" + ) + if value: + cri_obj.output_format = value + + # Submenu: SAN Entries + elif answer == "san_entries": + _edit_san_entries(cri_obj) + + # Submenu: Key Algorithm + elif answer == "key_algorithm": + value = _select_enum_option( + cri_obj.key_algorithm, "Select key algorithm" + ) + if value: + cri_obj.key_algorithm = value + + # Submenu: EC Curve + elif answer == "ecc_curve" and cri_obj.is_key_algorithm_ec(): + value = _select_enum_option(cri_obj.ecc_curve, "Select EC curve") + if value: + cri_obj.ecc_curve = value + + # Submenu: RSA Size + elif answer == "rsa_size" and cri_obj.is_key_algorithm_rsa(): + value = _select_enum_option(cri_obj.rsa_size, "Select RSA key size") + if value: + cri_obj.rsa_size = value + + # Submenu: OKP Curve + elif answer == "okp_curve" and cri_obj.is_key_algorithm_okp(): + value = _select_enum_option(cri_obj.okp_curve, "Select OKP curve") + if value: + cri_obj.okp_curve = value + + # Submenu: Valid Since + elif answer == "valid_since": + console.print() + value = qy.text( + message="Enter validity start date/time (blank for CA default)", + default=( + cri_obj.valid_since.strftime("%Y-%m-%d %H:%M:%S") + if cri_obj.valid_since + else "" + ), + validate=DateTimeValidator( + # Start date must be in the future (only date part is considered) + not_before=datetime.now() + .astimezone() + .replace(hour=0, minute=0, second=0, microsecond=0), + not_after=cri_obj.valid_until, + accept_blank=True, + ), + style=DEFAULT_QY_STYLE, + ).ask() + if value: + # User wants to clear the date + if value.strip() == "": + cri_obj.valid_since = None + else: + # The string is already validated, so we can directly parse it + cri_obj.valid_since = parse_date_str(value).astimezone() + + # Submenu: Valid Until + elif answer == "valid_until": + console.print() + value = qy.text( + message="Enter validity end date/time (blank for CA default)", + default=( + cri_obj.valid_until.strftime("%Y-%m-%d %H:%M:%S") + if cri_obj.valid_until + else "" + ), + validate=DateTimeValidator( + not_before=cri_obj.valid_since, accept_blank=True + ), + style=DEFAULT_QY_STYLE, + ).ask() + if value: + # User wants to clear the date + if value.strip() == "": + cri_obj.valid_until = None + else: + # The string is already validated, so we can directly parse it + cri_obj.valid_until = parse_date_str(value).astimezone() + + # Start review/edit loop + proceed = _review_and_edit(cri) + if not proceed: + return + + # After finalizing options, ask for optional private key password depending on output format + key_password = None + if cri.is_output_format_pem(): + key_password = _prompt_for_password( + message="Enter private key password (blank for no password)", + confirm_message="Confirm private key password", + ) + + # Optional PFX Encryption + pfx_password = None + if cri.is_output_format_pfx(): + pfx_password = _prompt_for_password( + message="Enter PFX password (blank for no password)", + confirm_message="Confirm PFX password", + ) + + # -------------------------- + # End of review/edit section + # -------------------------- + + result = execute_certificate_request(cri, ca_base_url) + if not result: + logger.info("Operation cancelled.") + return + + crt_path, key_path = result + + try: + result = convert_certificate( + crt_path=crt_path, + key_path=key_path, + output_dir=SCRIPT_CERT_DIR, + output_format=cri.output_format, + key_output_encryption_password=key_password, + pfx_output_encryption_password=pfx_password, + ) + except Exception as e: + logger.error(f"Failed to convert certificate: {e}") + return + + # Make sure the final output directory exists + cri.final_output_dir.mkdir(exist_ok=True, parents=True) + + try: + if result.certificate and result.private_key: + final_crt_path = join_safe_path( + target_dir=cri.final_output_dir, + target_file_name_with_suffix=cri.final_crt_output_name_with_suffix, + ) + final_key_path = join_safe_path( + target_dir=cri.final_output_dir, + target_file_name_with_suffix=cri.final_key_output_name_with_suffix, + ) + # Move the files to their final destination + result.certificate.rename(final_crt_path) + result.private_key.rename(final_key_path) + logger.info(f"Certificate saved to '{final_crt_path}'.") + logger.info(f"Private key saved to '{final_key_path}'.") + + elif result.pem_bundle: + final_pem_bundle_path = join_safe_path( + target_dir=cri.final_output_dir, + target_file_name_with_suffix=cri.final_pem_bundle_output_name_with_suffix, + ) + # Move the file to its final destination + result.pem_bundle.rename(final_pem_bundle_path) + logger.info(f"PEM bundle saved to '{final_pem_bundle_path}'.") + + elif result.pfx: + final_pfx_path = join_safe_path( + target_dir=cri.final_output_dir, + target_file_name_with_suffix=cri.final_pfx_bundle_output_name_with_suffix, + ) + # Move the file to its final destination + result.pfx.rename(final_pfx_path) + logger.info(f"PFX saved to '{final_pfx_path}'.") + + # This should never happen but just in case + else: + logger.error( + "Failed to save certificate because of an invalid CertificateConversionResult object." + ) + return + + except Exception as e: + logger.error(f"Failed to save certificate: {e}") + return + + # Delete the key and crt from the cache + if crt_path.exists(): + try: + crt_path.unlink() + logger.debug(f"Deleted certificate '{crt_path}' from cache") + except Exception as e: + logger.warning(f"Failed to delete certificate '{crt_path}' from cache: {e}") + + if key_path.exists(): + try: + key_path.unlink() + logger.debug(f"Deleted key '{key_path}' from cache") + except Exception as e: + logger.warning(f"Failed to delete key '{key_path}' from cache: {e}") diff --git a/step_cli_tools/utils/ca.py b/step_cli_tools/utils/ca.py new file mode 100644 index 0000000..5d1c44f --- /dev/null +++ b/step_cli_tools/utils/ca.py @@ -0,0 +1,262 @@ +# --- Standard library imports --- # +import re +import ssl +from pathlib import Path +from urllib.error import URLError +from urllib.request import urlopen + +# --- Third-party imports --- # +from cryptography import x509 +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes +from cryptography.x509.oid import NameOID + +# --- Local application imports --- # +from ..common import SCRIPT_CACHE_DIR, STEP_BIN, get_masked_url_for_logging, logger +from ..models.data import CertificateRequestInfo, RootCAInfo +from ..utils.network import is_server_certificate_trusted +from .general import execute_step_command + + +def execute_ca_request( + url: str, + trust_unknown_certificates: bool = False, + timeout: int = 10, +) -> str | None: + """ + Perform an HTTPS request to a step-ca server, respecting trust settings for certificates. + + Args: + url: URL to request. + trust_unknown_certificates: If True, trust unverified SSL certificates. + timeout: Timeout in seconds. + + Returns: + Response body as string, or None on failure. + """ + + masked_url = get_masked_url_for_logging(url) + logger.debug( + f"url={masked_url}, trust_unknown_certificates={trust_unknown_certificates}, timeout={timeout}" + ) + + 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() + + # Determine SSL context based on trust settings and server certificate status + if trust_unknown_certificates or not is_server_certificate_trusted(url): + context = ssl._create_unverified_context() + if not trust_unknown_certificates: + logger.warning( + f"Server certificate for '{masked_url}' is untrusted. Proceeding with unverified SSL context." + ) + else: + context = ssl.create_default_context() + + try: + return do_request(context) + + except URLError as e: + logger.error(f"Connection failed: {e}\n\nAre address and port correct?") + return + + except Exception as e: + logger.error(f"Request failed: {e}\n\nAre address and port correct?") + return + + +def execute_certificate_request( + request_parameters: CertificateRequestInfo, + ca_base_url: str, +) -> tuple[Path, Path] | None: + """ + Request a new certificate from a step-ca server. + + Args: + request_parameters: Certificate request parameters. + ca_base_url: Base URL of the step-ca server, including protocol and port. + + Returns: + Tuple of certificate and key paths on success, None on error or user cancel. + """ + + logger.debug(locals()) + + try: + request_parameters.validate() + except ValueError as e: + logger.error(f"Invalid certificate request parameters: {e}") + return + + # The step-ca server can only return the certificate and key in this format + tmp_crt_file_path = SCRIPT_CACHE_DIR / f"{request_parameters.timestamp}.crt" + tmp_key_file_path = SCRIPT_CACHE_DIR / f"{request_parameters.timestamp}.key" + + args = [ + "ca", + "certificate", + request_parameters.subject_name, + tmp_crt_file_path, + tmp_key_file_path, + "--ca-url", + ca_base_url, + "--force", + ] + + # Just a safety measure as the entries should contain the subject name at least + if request_parameters.san_entries: + for san_entry in request_parameters.san_entries: + args.extend(["--san", san_entry]) + + # The logic is already handled in the data class + if request_parameters.key_algorithm: + args.extend(["--kty", request_parameters.key_algorithm.value.arg]) + + if request_parameters.is_key_algorithm_ec(): + args.extend(["--curve", request_parameters.ecc_curve.value.arg]) + + if request_parameters.is_key_algorithm_okp(): + args.extend(["--curve", request_parameters.okp_curve.value.arg]) + + if request_parameters.is_key_algorithm_rsa(): + args.extend(["--size", request_parameters.rsa_size.value.arg]) + + if request_parameters.valid_since: + args.extend( + [ + "--not-before", + request_parameters.valid_since.isoformat(timespec="seconds"), + ] + ) + + if request_parameters.valid_until: + args.extend( + [ + "--not-after", + request_parameters.valid_until.isoformat(timespec="seconds"), + ] + ) + + # Interactive because the the user will be asked for the JWK password + success, _ = execute_step_command(args=args, step_bin=STEP_BIN, interactive=True) + if not success: + logger.error("Certificate request failed.") + return + + return tmp_crt_file_path, tmp_key_file_path + + +def is_step_ca_server_healthy( + ca_base_url: str, trust_unknown_certificates: bool = False +) -> bool: + """ + Check the health endpoint of a step-ca server via HTTPS. + + Args: + ca_base_url: Base URL of the step-ca server, including protocol and port. + trust_unknown_certificates: 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" + + response = execute_ca_request( + url=health_url, + trust_unknown_certificates=trust_unknown_certificates, + ) + + 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(): + logger.info(f"step-ca server at '{ca_base_url}' is healthy.") + return True + + logger.error(f"CA health check failed for '{ca_base_url}'.") + return False + + +def get_ca_root_info( + ca_base_url: str, + trust_unknown_certificates: bool = False, +) -> RootCAInfo | None: + """ + Fetch the first root certificate from a step-ca server and return its name and SHA256 fingerprint. + + Args: + ca_base_url: Base URL of the step-ca server (e.g. https://my-ca-host:9000). + trust_unknown_certificates: If True, trust unverified SSL certificates. + + Returns: + RootCAInfo on success, None on error. + """ + + logger.debug(locals()) + + roots_url = ca_base_url.rstrip("/") + "/roots.pem" + + pem_bundle = execute_ca_request( + url=roots_url, + trust_unknown_certificates=trust_unknown_certificates, + ) + + if pem_bundle is None: + logger.debug("Failed to retrieve roots.pem") + return + + try: + # Extract first PEM certificate + match = re.search( + "-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", + pem_bundle, + re.S, + ) + if not match: + logger.error("No certificate found in roots.pem") + return + + logger.debug("Loading PEM certificate") + cert = x509.load_pem_x509_certificate( + match.group(0).encode(), + default_backend(), + ) + + # Compute SHA256 fingerprint + fingerprint_hex = cert.fingerprint(hashes.SHA256()).hex().upper() + fingerprint = ":".join( + fingerprint_hex[i : i + 2] for i in range(0, len(fingerprint_hex), 2) + ) + + # 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 = ( + str(cn[0].value) + if cn and cn[0].value is not None + else str(cert.subject.rfc4514_string()) + ) + except Exception as e: + logger.warning(f"Unable to retrieve CA name: {e}") + ca_name = "Unknown CA" + + logger.info("Root CA information retrieved successfully.") + + return RootCAInfo( + ca_name=ca_name, + fingerprint_sha256=fingerprint.replace(":", ""), + ) + + except Exception as e: + logger.error(f"Failed to process CA root certificate: {e}") + return diff --git a/step_cli_tools/support_functions.py b/step_cli_tools/utils/certificates.py similarity index 53% rename from step_cli_tools/support_functions.py rename to step_cli_tools/utils/certificates.py index 6b74941..14f243e 100644 --- a/step_cli_tools/support_functions.py +++ b/step_cli_tools/utils/certificates.py @@ -1,445 +1,27 @@ -# --- Standard library imports --- +# --- Standard library imports --- # import base64 import json -import os -import platform import re -import shutil -import ssl import subprocess -import tarfile -import tempfile -import time import warnings from pathlib import Path -from urllib.request import urlopen -import urllib.error -from zipfile import ZipFile +from typing import TypeVar -# --- Third-party imports --- +# --- Third-party imports --- # from cryptography import x509 from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import dsa, ec, ed448, ed25519, rsa +from cryptography.hazmat.primitives.asymmetric.types import PublicKeyTypes +from cryptography.hazmat.primitives.serialization import load_pem_private_key, pkcs12 from cryptography.utils import CryptographyDeprecationWarning -from cryptography.x509.oid import NameOID -from packaging import version - -# --- Local application imports --- -from .common import * -from .configuration import * -from .data_classes import * - -__all__ = [ - "check_for_update", - "install_step_cli", - "execute_step_command", - "check_ca_health", - "get_ca_root_info", - "find_windows_cert_by_sha256", - "find_windows_certs_by_name", - "find_linux_cert_by_sha256", - "find_linux_certs_by_name", - "delete_windows_cert_by_thumbprint", - "delete_linux_cert_by_path", - "choose_cert_from_list", -] - - -def check_for_update( - pkg_name: str, current_pkg_version: str, include_prerelease: bool = False -) -> str | None: - """ - Check PyPI for newer releases of the package. - - Args: - 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. - """ - - 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: - 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) > current_parsed_version - ): - logger.debug("Returning newer version from cache") - return latest_version - except (json.JSONDecodeError, OSError) as e: - logger.debug(f"Failed to read update cache: {e}") +# --- Local application imports --- # +from ..common import DEFAULT_QY_STYLE, console, logger, qy +from ..models.data import CertificateConversionResult, CRI_OutputFormat - # 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) - latest_parsed_version = version.parse(latest_version) - - logger.debug(f"Latest available version on PyPI: {latest_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 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. - - Args: - step_bin: Path to the step binary. - """ - - system = platform.system() - arch = platform.machine() - 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" - archive_type = "zip" - elif system == "Linux": - url = "https://github.com/smallstep/cli/releases/latest/download/step_linux_amd64.tar.gz" - archive_type = "tar.gz" - elif system == "Darwin": - url = "https://github.com/smallstep/cli/releases/latest/download/step_darwin_amd64.tar.gz" - archive_type = "tar.gz" - else: - logger.error(f"Unsupported platform: {system}") - return - - tmp_dir = tempfile.mkdtemp() - tmp_path = os.path.join(tmp_dir, os.path.basename(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()) - - 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) - else: - with tarfile.open(tmp_path, "r:gz") as tar_ref: - tar_ref.extractall(tmp_dir) - - step_bin_name = os.path.basename(step_bin) - - # Search recursively for the binary - matches = [] - for root, _, files in os.walk(tmp_dir): - if step_bin_name in files: - found_path = os.path.join(root, step_bin_name) - matches.append(found_path) - - if not matches: - 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) - os.makedirs(binary_dir, exist_ok=True) - - # 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) - - logger.info(f"step-cli installed: {step_bin}") - - try: - result = subprocess.run([step_bin, "version"], capture_output=True, text=True) - logger.info(f"Installed step version:\n{result.stdout.strip()}") - except Exception as e: - logger.error(f"Failed to run step-cli: {e}") - - -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. - step_bin: Path to the step binary. - interactive: If True, run the command interactively without capturing output. - - Returns: - Command output as a string if successful, otherwise None. - """ - - logger.debug(locals()) - - if not step_bin or not os.path.exists(step_bin): - 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: - 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: - logger.error(f"step-cli command failed: {result.stderr.strip()}") - return None - - return result.stdout.strip() - - except Exception as e: - logger.error(f"Failed to execute step-cli command: {e}") - return None - - -def execute_ca_request( - url: str, - trust_unknown_default: bool = False, - timeout: int = 10, -) -> str | None: - """ - 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. - """ - - 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 = ( - ssl._create_unverified_context() - if trust_unknown_default - else ssl.create_default_context() - ) - - try: - return do_request(context) - - except urllib.error.URLError as e: - reason = getattr(e, "reason", None) - - logger.debug(f"URLError: {e}") - - if isinstance(reason, ssl.SSLCertVerificationError): - logger.warning("Server provided an unknown or self-signed certificate.") - - console.print() - answer = qy.confirm( - message=f"Do you want to trust '{url}' this time?", - default=False, - style=DEFAULT_QY_STYLE, - ).ask() - - if not answer: - 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: - logger.error( - f"Retry failed: {retry_error}\n\nIs the port correct and the server available?" - ) - return None - - logger.error( - f"Connection failed: {e}\n\nIs the port correct and the server available?" - ) - return None - - except Exception as e: - 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. - - 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" - - response = execute_ca_request( - health_url, - trust_unknown_default=trust_unknown_default, - ) - - 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(): - logger.info(f"CA at '{ca_base_url}' is healthy.") - return True - - logger.error(f"CA health check failed for '{ca_base_url}'.") - return False - - -def get_ca_root_info( - ca_base_url: str, - trust_unknown_default: bool = False, -) -> CARootInfo | None: - """ - Fetch the first root certificate from a Smallstep CA and return its name - 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. - - Returns: - 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( - roots_url, - trust_unknown_default=trust_unknown_default, - ) - - if pem_bundle is None: - logger.debug("Failed to retrieve roots.pem") - return None - - try: - # Extract first PEM certificate - match = re.search( - "-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", - pem_bundle, - re.S, - ) - if not match: - 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(), - ) - - # Compute SHA256 fingerprint - fingerprint_hex = cert.fingerprint(hashes.SHA256()).hex().upper() - fingerprint = ":".join( - fingerprint_hex[i : i + 2] for i in range(0, len(fingerprint_hex), 2) - ) - - # 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 = ( - str(cn[0].value) - if cn and cn[0].value is not None - else str(cert.subject.rfc4514_string()) - ) - except Exception as e: - logger.warning(f"Unable to retrieve CA name: {e}") - ca_name = "Unknown CA" - - logger.info("Root CA information retrieved successfully.") - - return CARootInfo( - ca_name=ca_name, - fingerprint_sha256=fingerprint.replace(":", ""), - ) - - except Exception as e: - logger.error(f"Failed to process CA root certificate: {e}") - return None +# Used for the choose_cert_from_list() function +TPathOrStr = TypeVar("TPathOrStr", Path, str) def find_windows_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None: @@ -489,7 +71,7 @@ def find_windows_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | No if result.returncode != 0: logger.error(f"Failed to query certificates: {result.stderr.strip()}") - return None + return normalized_fp = sha256_fingerprint.lower().replace(":", "") @@ -505,7 +87,7 @@ def find_windows_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | No continue logger.debug("No matching Windows certificate found") - return None + return def find_windows_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: @@ -583,7 +165,7 @@ def find_windows_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: return matches -def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None: +def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[Path, str] | None: """ Search the Linux system trust store for a certificate matching a given SHA256 fingerprint. @@ -600,47 +182,47 @@ def find_linux_cert_by_sha256(sha256_fingerprint: str) -> tuple[str, str] | None logger.debug(f"Starting Linux certificate search by SHA256: {sha256_fingerprint}") - cert_dir = "/etc/ssl/certs" + cert_dir = Path("/etc/ssl/certs") fingerprint = sha256_fingerprint.lower().replace(":", "") - if not os.path.isdir(cert_dir): + if not cert_dir.is_dir(): logger.error(f"Cert directory not found: {cert_dir}") - return None + return # Ignore deprecation warnings about non-positive serial numbers with warnings.catch_warnings(): warnings.filterwarnings("ignore", category=CryptographyDeprecationWarning) - for cert_file in os.listdir(cert_dir): - path = os.path.join(cert_dir, cert_file) - if os.path.isfile(path): + for cert_file in cert_dir.iterdir(): + if cert_file.is_file(): try: - logger.debug(f"Reading certificate file: {path}") - with open(path, "rb") as f: - cert_data = f.read() - try: - # Try PEM first - cert = x509.load_pem_x509_certificate( - cert_data, default_backend() - ) - except ValueError: - # Fallback to DER - cert = x509.load_der_x509_certificate( - cert_data, default_backend() - ) - fp = cert.fingerprint(hashes.SHA256()).hex() - if fp.lower() == fingerprint: - logger.debug("Matching Linux certificate found") - return (path, cert.subject.rfc4514_string()) + logger.debug(f"Reading certificate file: {cert_file}") + cert_data = cert_file.read_bytes() + try: + # Try PEM first + cert = x509.load_pem_x509_certificate( + cert_data, default_backend() + ) + except ValueError: + # Fallback to DER + cert = x509.load_der_x509_certificate( + cert_data, default_backend() + ) + fp = cert.fingerprint(hashes.SHA256()).hex() + if fp.lower() == fingerprint: + logger.debug("Matching Linux certificate found") + return cert_file, cert.subject.rfc4514_string() except Exception as e: - logger.debug(f"Failed to process certificate file '{path}': {e}") + logger.debug( + f"Failed to process certificate file '{cert_file}': {e}" + ) continue logger.debug("No matching Linux certificate found") - return None + return -def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: +def find_linux_certs_by_name(name_pattern: str) -> list[tuple[Path, str]]: """ Search Linux trust store for certificates by name. Supports simple wildcard '*' and matches separately against @@ -651,13 +233,13 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: 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): + cert_dir = Path("/etc/ssl/certs") + if not cert_dir.is_dir(): logger.error(f"Cert directory not found: {cert_dir}") return [] @@ -665,19 +247,18 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: escaped_pattern = re.escape(name_pattern).replace(r"\*", ".*") pattern_re = re.compile(f"^{escaped_pattern}$", re.IGNORECASE) - matches = [] - seen_real_paths: set[str] = set() + matches: list[tuple[Path, str]] = [] + seen_real_paths: set[Path] = set() with warnings.catch_warnings(): warnings.filterwarnings("ignore", category=CryptographyDeprecationWarning) - for cert_file in os.listdir(cert_dir): - path = os.path.join(cert_dir, cert_file) - if not os.path.isfile(path): + for cert_file in cert_dir.iterdir(): + if not cert_file.is_file(): continue try: - real_path = os.path.realpath(path) + real_path = cert_file.resolve() # Skip duplicate certificates pointing to the same real file if real_path in seen_real_paths: @@ -685,35 +266,29 @@ def find_linux_certs_by_name(name_pattern: str) -> list[tuple[str, str]]: 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: - # PEM support - cert = x509.load_pem_x509_certificate( - cert_data, default_backend() - ) - except ValueError: - # Fallback to DER - cert = x509.load_der_x509_certificate( - cert_data, default_backend() - ) - subject_str = cert.subject.rfc4514_string() - components = [comp.strip() for comp in subject_str.split(",")] + logger.debug(f"Processing certificate file: {cert_file}") - for comp in components: - 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((path, subject_str)) - break + cert_data = cert_file.read_bytes() + try: + # PEM support + cert = x509.load_pem_x509_certificate(cert_data, default_backend()) + except ValueError: + # Fallback to DER + cert = x509.load_der_x509_certificate(cert_data, default_backend()) + + subject_str = cert.subject.rfc4514_string() + components = [comp.strip() for comp in subject_str.split(",")] + + for comp in components: + 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((cert_file, subject_str)) + break except Exception as e: - logger.debug(f"Failed to process certificate file '{path}': {e}") + logger.debug(f"Failed to process certificate file '{cert_file}': {e}") continue logger.debug(f"Total matching Linux certificates found: {len(matches)}") @@ -734,7 +309,7 @@ def delete_windows_cert_by_thumbprint(thumbprint: str, cn: str, elevated: bool = console.print() answer = qy.confirm( - message=f"Do you really want to remove the certificate: '{cn}'?", + message=f"Do you really want to remove the certificate '{cn}'?", default=False, style=DEFAULT_QY_STYLE, ).ask() @@ -824,7 +399,7 @@ def delete_windows_cert_by_thumbprint(thumbprint: str, cn: str, elevated: bool = ).ask() if not retry_with_admin_privileges: logger.info("Operation cancelled by user.") - return None + return delete_windows_cert_by_thumbprint(thumbprint, cn, elevated=True) return @@ -836,7 +411,7 @@ def delete_windows_cert_by_thumbprint(thumbprint: str, cn: str, elevated: bool = logger.error(f"Failed to remove certificate with thumbprint '{thumbprint}'") -def delete_linux_cert_by_path(cert_path: str, cn: str, elevated: bool = False): +def delete_linux_cert_by_path(cert_path: Path, cn: str, elevated: bool = False): """ Delete a certificate from the Linux system trust store. @@ -846,7 +421,6 @@ def delete_linux_cert_by_path(cert_path: str, cn: str, elevated: bool = False): 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") @@ -876,7 +450,7 @@ def confirm_retry(message: str) -> bool: console.print() answer = qy.confirm( - message=f"Do you really want to remove the certificate: '{cn}'?", + message=f"Do you really want to remove the certificate '{cn}'?", default=False, style=DEFAULT_QY_STYLE, ).ask() @@ -884,11 +458,11 @@ def confirm_retry(message: str) -> bool: logger.info("Operation cancelled by user.") return - if not cert_path_obj.is_symlink(): + if not cert_path.is_symlink(): logger.warning(f"'{cert_path}' is not a symlink, skipping.") return - target_path = cert_path_obj.resolve() + target_path = cert_path.resolve() logger.debug(f"Resolved symlink target: {target_path}") try: @@ -979,14 +553,14 @@ def confirm_retry(message: str) -> bool: def choose_cert_from_list( - certs: list[tuple[str, str]], message: str = "Select a certificate:" -) -> tuple[str, str] | None: + certs: list[tuple[TPathOrStr, str]], message: str = "Select a certificate:" +) -> tuple[TPathOrStr, str] | None: """ - Presents an 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 (Path/thumbprint, subject). Args: - certs: List of tuples (id, subject) to choose from. - message: message text for the questionary select. + certs: List of tuples (Path/thumbprint, subject) to choose from. + message: A message text for the questionary select. Returns: The selected tuple or None if user cancels. @@ -996,7 +570,7 @@ def choose_cert_from_list( if not certs: logger.debug("No certificates available for selection") - return None + return # Sort certificates alphabetically by subject (case-insensitive) sorted_certs = sorted(certs, key=lambda cert: cert[1].lower()) @@ -1015,7 +589,7 @@ def choose_cert_from_list( if selected_subject is None: logger.debug("User cancelled certificate selection") - return None + return # Return the full tuple matching the selected subject for cert in sorted_certs: @@ -1026,4 +600,160 @@ def choose_cert_from_list( return cert logger.debug("Selected certificate not found in internal list") - return None + return + + +def convert_certificate( + crt_path: Path, + key_path: Path, + output_dir: Path, + output_format: CRI_OutputFormat, + key_output_encryption_password: str | None = None, + pfx_output_encryption_password: str | None = None, +) -> CertificateConversionResult: + """ + Convert or bundle a CRT and KEY file into the desired output format. + + Args: + crt_path: Path to the PEM-encoded certificate (.crt). + key_path: Path to the PEM-encoded private key (.key). + output_dir: Directory where output file(s) will be written. + output_format: Desired output format (PEM_CRT_KEY, PEM_BUNDLE, PFX_BUNDLE). + key_output_encryption_password: Optional password for KEY output. + pfx_output_encryption_password: Optional password for PFX output. + + Returns: + CertificateConversionResult with relevant paths set. + """ + + # Mask sensitive values for logging + debug_locals = { + key: ( + "***" + if key + in {"key_output_encryption_password", "pfx_output_encryption_password"} + and value is not None + else value + ) + for key, value in locals().items() + } + logger.debug(debug_locals) + + def _public_key_bytes(public_key: "PublicKeyTypes") -> bytes: + # Serialize public key to a canonical DER representation for comparison + return public_key.public_bytes( + encoding=serialization.Encoding.DER, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + + output_dir.mkdir(parents=True, exist_ok=True) + result = CertificateConversionResult(format=output_format) + + # --- PEM_CRT_KEY --- + if output_format == CRI_OutputFormat.PEM_CRT_KEY: + crt_out = output_dir / crt_path.name + key_out = output_dir / key_path.name + + # Copy certificate + if crt_path != crt_out: + crt_out.write_bytes(crt_path.read_bytes()) + else: + logger.debug( + f"Certificate output path '{crt_out}' is the same as input, skipping copy." + ) + + # Write (optionally encrypted) private key + if key_output_encryption_password: + private_key_obj = load_pem_private_key(key_path.read_bytes(), password=None) + key_bytes = private_key_obj.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.BestAvailableEncryption( + key_output_encryption_password.encode() + ), + ) + key_out.write_bytes(key_bytes) + else: + if key_path != key_out: + key_out.write_bytes(key_path.read_bytes()) + else: + logger.debug( + f"Private key output path '{key_out}' is the same as input, skipping copy." + ) + + result.certificate = crt_out + result.private_key = key_out + return result + + crt_data = crt_path.read_bytes() + key_data = key_path.read_bytes() + + # --- PEM_BUNDLE --- + if output_format == CRI_OutputFormat.PEM_BUNDLE: + bundle_path = output_dir / f"{crt_path.stem}_bundle.pem" + + # Encrypt key first if password provided + if key_output_encryption_password: + private_key_obj = load_pem_private_key(key_data, password=None) + key_bytes = private_key_obj.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.BestAvailableEncryption( + key_output_encryption_password.encode() + ), + ) + else: + key_bytes = key_data + + # Key first, certificate second + combined = key_bytes.rstrip() + b"\n\n" + crt_data.lstrip() + bundle_path.write_bytes(combined) + + result.pem_bundle = bundle_path + return result + + # --- PFX_BUNDLE --- + if output_format == CRI_OutputFormat.PFX_BUNDLE: + cert = x509.load_pem_x509_certificate(crt_data) + private_key = load_pem_private_key(key_data, password=None) + + if not isinstance( + private_key, + ( + rsa.RSAPrivateKey, + dsa.DSAPrivateKey, + ec.EllipticCurvePrivateKey, + ed25519.Ed25519PrivateKey, + ed448.Ed448PrivateKey, + ), + ): + raise TypeError( + f"Unsupported private key type for PKCS#12 serialization: {type(private_key).__name__}" + ) + + if _public_key_bytes(cert.public_key()) != _public_key_bytes( + private_key.public_key() + ): + raise ValueError( + f"The provided private key '{key_path}' does not match the certificate '{crt_path}'" + ) + + pfx_path = output_dir / f"{crt_path.stem}.pfx" + pfx_bytes = pkcs12.serialize_key_and_certificates( + name=crt_path.stem.encode(), + key=private_key, + cert=cert, + cas=None, + encryption_algorithm=( + serialization.BestAvailableEncryption( + pfx_output_encryption_password.encode() + ) + if pfx_output_encryption_password + else serialization.NoEncryption() + ), + ) + pfx_path.write_bytes(pfx_bytes) + result.pfx = pfx_path + return result + + raise ValueError(f"Unsupported output format: {output_format}") diff --git a/step_cli_tools/utils/general.py b/step_cli_tools/utils/general.py new file mode 100644 index 0000000..905aef1 --- /dev/null +++ b/step_cli_tools/utils/general.py @@ -0,0 +1,259 @@ +# --- Standard library imports --- +import json +import platform +import shutil +import subprocess +import tarfile +import tempfile +import time +from datetime import datetime +from pathlib import Path +from urllib.request import urlopen +from zipfile import ZipFile + +# --- Third-party imports --- +from packaging import version + +# --- Local application imports --- +from ..common import SCRIPT_CACHE_DIR, logger +from ..configuration import config +from ..utils.validators import DateTimeValidator + + +def check_for_update( + pkg_name: str, current_pkg_version: str, include_prerelease: bool = False +) -> str | None: + """ + Check PyPI for newer releases of the package. + + Args: + 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. + """ + + cache = SCRIPT_CACHE_DIR / f"{pkg_name}_update_check_cache.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: + 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) > current_parsed_version + ): + logger.debug("Returning newer version from cache") + return latest_version + + 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 + + latest_version = max(releases, key=version.parse) + latest_parsed_version = version.parse(latest_version) + + logger.debug(f"Latest available version on PyPI: {latest_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 as e: + logger.debug(f"Update check failed: {e}") + return + + +def install_step_cli(step_bin: Path): + """ + 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() + logger.debug(f"Detected platform: {system} {arch}") + logger.debug(f"Target installation path: {step_bin}") + + if system == "Windows": + url = "https://github.com/smallstep/cli/releases/latest/download/step_windows_amd64.zip" + archive_type = "zip" + elif system == "Linux": + url = "https://github.com/smallstep/cli/releases/latest/download/step_linux_amd64.tar.gz" + archive_type = "tar.gz" + elif system == "Darwin": + url = "https://github.com/smallstep/cli/releases/latest/download/step_darwin_amd64.tar.gz" + archive_type = "tar.gz" + else: + logger.error(f"Unsupported platform: {system}") + return + + tmp_dir = Path(tempfile.mkdtemp()) + tmp_path = tmp_dir / Path(url).name + logger.debug(f"Downloading step-cli from '{url}'...") + + with urlopen(url) as response, tmp_path.open("wb") as out_file: + out_file.write(response.read()) + + logger.debug(f"Archive downloaded to temporary path: {tmp_path}") + + logger.debug(f"Extracting '{archive_type}' archive...") + if archive_type == "zip": + with ZipFile(tmp_path, "r") as zip_ref: + zip_ref.extractall(tmp_dir) + else: + with tarfile.open(tmp_path, "r:gz") as tar_ref: + tar_ref.extractall(tmp_dir) + + # Search recursively for the binary + matches = [p for p in tmp_dir.rglob(step_bin.name)] + + if not matches: + logger.error(f"Could not find '{step_bin.name}' in the extracted archive.") + return + + extracted_path = matches[0] + logger.debug(f"Using extracted binary: {extracted_path}") + + # Prepare installation path + step_bin.parent.mkdir(parents=True, exist_ok=True) + + # Delete old binary if exists + if step_bin.exists(): + logger.debug("Removing existing step binary") + step_bin.unlink() + + shutil.move(str(extracted_path), str(step_bin)) + step_bin.chmod(0o755) + + logger.debug(f"step-cli binary: {step_bin}") + + try: + result = subprocess.run( + [str(step_bin), "version"], capture_output=True, text=True + ) + logger.info(result.stdout.strip()) + except Exception as e: + logger.error(f"Failed to run step-cli: {e}") + + +def execute_step_command( + args: list[str], step_bin: Path, interactive: bool = False +) -> tuple[bool, str | None]: + """ + Execute a step-cli command and return output or log errors. + + Args: + args: List of command-line arguments for step-cli. + step_bin: Path to the step-cli binary. + interactive: Whether to run the command interactively. + + Returns: + Tuple of (success, output) + - success: True if the command executed successfully + - output: Captured stdout if non-interactive, otherwise None + """ + + logger.debug(locals()) + + if not step_bin.exists(): + logger.error("step-cli not found. Please install it first.") + return False, None + + try: + result = subprocess.run( + [step_bin] + args, + capture_output=not interactive, + text=True, + ) + + logger.debug(f"step-cli command exit code: {result.returncode}") + if not interactive: + if result.stdout: + logger.debug(f"step-cli command stdout: {result.stdout.strip()}") + if result.stderr: + logger.debug(f"step-cli command stderr: {result.stderr.strip()}") + + if result.returncode != 0: + error_msg = ( + f"step-cli command exit code: {result.returncode}" + if interactive + else f"step-cli command failed: {result.stderr.strip()}" + ) + logger.error(error_msg) + return False, ( + result.stdout.strip() if not interactive and result.stdout else None + ) + + return True, None if interactive else result.stdout.strip() + + except Exception as e: + logger.error(f"Failed to execute step-cli command: {e}") + return False, None + + +def parse_date_str(date_str: str) -> datetime: + """ + Parse a date string into a datetime object. + + Args: + date_str: The date string to parse. + + Returns: + A datetime object representing the parsed date. + + Raises: + ValueError: If the date string is not in a supported format. + """ + + for fmt in DateTimeValidator.SUPPORTED_FORMATS: + try: + return datetime.strptime(date_str, fmt) + except ValueError: + pass + + raise ValueError(f"Invalid date format: {date_str}") diff --git a/step_cli_tools/utils/network.py b/step_cli_tools/utils/network.py new file mode 100644 index 0000000..db32356 --- /dev/null +++ b/step_cli_tools/utils/network.py @@ -0,0 +1,141 @@ +# --- Standard library imports --- +import socket +import ssl +from urllib.parse import urlparse + +# --- Third-party imports --- +import requests + +# --- Local application imports --- +from ..common import get_masked_url_for_logging, logger + + +def is_server_certificate_trusted( + server_url: str, use_system_store: bool = True, timeout_seconds: int = 5 +) -> bool: + """ + Check if the server TLS certificate is trusted by the system. + + Args: + server_url: The URL of the server to check. + use_system_store: If True, use the system trust store. If False, use the default certifi bundle. + timeout_seconds: Timeout in seconds. + + Returns: + True if the server certificate is trusted, False otherwise. + """ + + masked_server_url = get_masked_url_for_logging(server_url) + logger.debug( + f"server_url={masked_server_url}, use_system_store={use_system_store}, timeout_seconds={timeout_seconds}" + ) + + if use_system_store: + return _is_certificate_trusted_via_system_store(server_url, timeout_seconds) + + try: + requests.get(server_url, timeout=timeout_seconds) + logger.debug(f"Server certificate from '{masked_server_url}' is trusted") + return True + except requests.exceptions.SSLError as e: + logger.debug( + f"Server certificate from '{masked_server_url}' is not trusted: {e}" + ) + return False + except requests.exceptions.RequestException as e: + logger.error( + f"Failed to check server certificate from '{masked_server_url}': {e}" + ) + return False + + +def _is_certificate_trusted_via_system_store( + server_url: str, + timeout_seconds: int, +) -> bool: + """ + Perform a TLS handshake using the OS trust store. + + Args: + server_url: The URL of the server to check. + timeout_seconds: Timeout in seconds. + + Returns: + True if the server certificate is trusted (system trust store), False otherwise. + """ + + masked_server_url = get_masked_url_for_logging(server_url) + + parsed_url = urlparse(server_url) + hostname = parsed_url.hostname + port = parsed_url.port or 443 + + if hostname is None: + logger.error( + f"Invalid server URL provided: '{masked_server_url}'.", + ) + return False + + context = ssl.create_default_context() + + try: + with socket.create_connection( + (hostname, port), timeout=timeout_seconds + ) as sock: + with context.wrap_socket(sock, server_hostname=hostname): + logger.debug( + f"Server certificate from '{masked_server_url}' is trusted (system trust store)", + ) + return True + except ssl.SSLError as e: + logger.debug( + f"Server certificate from '{masked_server_url}' is not trusted (system trust store): {e}", + ) + return False + except OSError as e: + logger.error( + f"Failed to check server certificate from '{masked_server_url}': {e}", + ) + return False + + +def is_host_available(server_url: str, timeout_seconds: int = 5) -> bool: + """ + Check if a host is reachable on the given port via TCP. + + Args: + server_url: The server URL containing hostname and optional port. + timeout_seconds: Timeout in seconds. + + Returns: + True if the host is reachable on the specified port, False otherwise. + """ + + masked_server_url = get_masked_url_for_logging(server_url) + + parsed_url = urlparse(server_url) + hostname = parsed_url.hostname + # Use default port if not specified + if parsed_url.port: + port = parsed_url.port + elif parsed_url.scheme == "http": + port = 80 + else: + port = 443 + + if hostname is None: + logger.error(f"Invalid server URL provided: '{masked_server_url}'") + return False + + logger.debug(f"host='{hostname}', port={port}, timeout_seconds={timeout_seconds}") + + try: + with socket.create_connection( + (hostname, port), + timeout=timeout_seconds, + ): + logger.debug(f"Host '{hostname}' is reachable on port {port}") + return True + except (OSError, socket.timeout) as e: + logger.debug(f"Host '{hostname}' is not reachable on port {port}: {e}") + return False diff --git a/step_cli_tools/utils/paths.py b/step_cli_tools/utils/paths.py new file mode 100644 index 0000000..475ba8b --- /dev/null +++ b/step_cli_tools/utils/paths.py @@ -0,0 +1,127 @@ +# --- Standard library imports --- +import re +from datetime import datetime +from pathlib import Path + +# --- Local application imports --- +from ..common import logger + +WINDOWS_RESERVED_NAMES = { + "CON", + "PRN", + "AUX", + "NUL", + *(f"COM{i}" for i in range(1, 10)), + *(f"LPT{i}" for i in range(1, 10)), +} + +UMLAUT_MAP = { + "ร„": "Ae", + "ร–": "Oe", + "รœ": "Ue", + "รค": "ae", + "รถ": "oe", + "รผ": "ue", + "รŸ": "ss", +} + + +def sanitize_filename(value: str) -> str: + """ + Convert a string into a filesystem-safe filename. + + Args: + value: The string to sanitize + + Returns: + The sanitized filename + """ + + original_value = value + path = Path(value) + + stem = path.stem + suffix = path.suffix + + # Replace German umlauts + for umlaut, replacement in UMLAUT_MAP.items(): + stem = stem.replace(umlaut, replacement) + + # Replace asterisk with a readable token + stem = stem.replace("*", "wildcard") + + # Replace path separators and whitespace + stem = re.sub(r"[\\/]+", "_", stem) + stem = re.sub(r"\s+", "_", stem) + + # Allow only safe characters + stem = re.sub(r"[^A-Za-z0-9._-]", "_", stem) + + # Collapse multiple underscores + stem = re.sub(r"_+", "_", stem) + + # Collapse multiple dots + stem = re.sub(r"\.{2,}", ".", stem) + + # Delete leading/trailing dots and underscores + stem = stem.strip("._") + + if not stem: + stem = f"sanitized_filename_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S-%f')}" + logger.warning( + f"Filename '{original_value}' is empty after sanitization. Using '{stem}' instead." + ) + + if stem.upper() in WINDOWS_RESERVED_NAMES: + logger.warning( + f"Filename '{stem}' is a reserved name on Windows. Appending underscore." + ) + stem = f"{stem}_" + + filename = f"{stem}{suffix}" + + if len(filename) > 255: + logger.warning( + f"Filename '{filename[:255]}...' is too long. Truncating to 255 characters." + ) + filename = filename[:255] + + logger.debug(f"Sanitized filename: {filename}") + return filename + + +def join_safe_path( + target_dir: Path, target_file_name_with_suffix: Path, ensure_target_dir_exists=True +) -> Path: + """ + Join a target directory with a sanitized filename and return the final path for the file. + Ensures no filename collisions by appending a timestamp before the extension if necessary. + + Args: + target_dir: The base directory for the file. + target_file_name_with_suffix: The desired filename with suffix which will be sanitized. + ensure_target_dir_exists: If True, the target directory will be created if it doesn't exist. + + Returns: + The final path for the file. + """ + + # Sanitize the filename + sanitized_filename = sanitize_filename(target_file_name_with_suffix.name) + stem = Path(sanitized_filename).stem + suffix = "".join(Path(sanitized_filename).suffixes) + + # Ensure target directory exists if necessary + if ensure_target_dir_exists: + logger.debug(f"Creating target directory: {target_dir}") + target_dir.mkdir(parents=True, exist_ok=True) + + final_path = target_dir / sanitized_filename + + # If file already exists, append timestamp before the suffix + if final_path.exists(): + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S-%f") + final_path = target_dir / f"{stem}_{timestamp}{suffix}" + + logger.debug(f"Final path: {final_path}") + return final_path diff --git a/step_cli_tools/utils/validators.py b/step_cli_tools/utils/validators.py new file mode 100644 index 0000000..2519ff7 --- /dev/null +++ b/step_cli_tools/utils/validators.py @@ -0,0 +1,504 @@ +# --- Standard library imports --- +import ipaddress +import re +from datetime import datetime, timezone + +# --- Third-party imports --- +from prompt_toolkit.document import Document +from questionary import ValidationError, Validator + +# --- Local application imports --- +from ..common import logger + + +class CertificateSubjectNameValidator(Validator): + """ + Validates a certificate Subject Name or SAN entry. + Supports: + - CN/DN (escaped) + - DNS hostnames with optional wildcard in first label + - IPv4 and IPv6 addresses + + Args: + accept_blank: If True, accept blank input. + + Returns: + None if valid, otherwise a string describing the validation error. + """ + + def __init__(self, accept_blank: bool = False): + self.accept_blank = accept_blank + super().__init__() + + # Characters that must be escaped in DN (RFC 4514) + DN_ESCAPE_CHARS = r",+\"\\<>;=" + + # DNS regex allowing optional wildcard in first label + DNS_WILDCARD_REGEX = re.compile( + r"^(?:\*\.)?(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(? datetime | None: + if value is None: + return + + if value.tzinfo is None: + logger.warning("Datetime is not timezone aware, assuming UTC.") + return value.replace(tzinfo=timezone.utc) + + return value + + def _parse_datetime(self, value: str) -> datetime | None: + # First try ISO-8601 (fast path) + try: + return datetime.fromisoformat(value) + except ValueError: + pass + + # Fallback to supported legacy formats + for fmt in self.SUPPORTED_FORMATS: + try: + return datetime.strptime(value, fmt) + except ValueError: + continue + + return + + def validate(self, document): + if document.text is None or document.text.strip() == "": + # Accept blank input if configured + if self.accept_blank: + return + raise ValidationError( + message="Value cannot be blank", + cursor_position=len(document.text), + ) + + value = document.text.strip() + + now_recommended = datetime.now(timezone.utc).strftime(self.recommended_format) + parsed_datetime = self._parse_datetime(value) + if parsed_datetime is None: + raise ValidationError( + message=( + f"Invalid date/time value. " + f"Recommended format: {now_recommended}" + ), + cursor_position=len(document.text), + ) + + # Normalize parsed datetime + if parsed_datetime.tzinfo is None: + parsed_datetime = parsed_datetime.replace(tzinfo=timezone.utc) + + # Validate lower bound + if self.not_before is not None and parsed_datetime < self.not_before: + raise ValidationError( + message=( + f"'{value}' is earlier than allowed minimum " + f"'{self.not_before.strftime(self.recommended_format)}'" + ), + cursor_position=len(document.text), + ) + + # Validate upper bound + if self.not_after is not None and parsed_datetime > self.not_after: + raise ValidationError( + message=( + f"'{value}' is later than allowed maximum " + f"'{self.not_after.strftime(self.recommended_format)}'" + ), + cursor_position=len(document.text), + ) + + +# --- Validators used by the configuration class --- + + +def int_range_validator(min_value: int, max_value: int): + """ + Returns a validator function that ensures an integer is within [min_value, max_value]. + + Args: + min_value: Minimum allowed integer value (inclusive). + max_value: Maximum allowed integer value (inclusive). + + Returns: + A function(value) -> Optional[str]: + - Returns None if valid. + - Returns a string describing the problem if invalid. + """ + + def validator(value): + if not isinstance(value, int): + return f"Invalid type: expected int, got {type(value).__name__}" + if value < min_value or value > max_value: + return f"Value {value} out of range ({min_value}โ€“{max_value})" + return + + return validator + + +def str_allowed_validator(allowed: list[str]): + """ + Returns a validator function that ensures a string value is one of the allowed values. + + Args: + allowed: List of allowed string values. + + Returns: + A function(value) -> Optional[str]: + - Returns None if valid. + - Returns a descriptive string if invalid. + """ + + def validator(value): + if not isinstance(value, str): + return f"Invalid type: expected str, got {type(value).__name__}" + if value not in allowed: + allowed_str = ", ".join(map(repr, allowed)) + return f"Invalid value '{value}'. Allowed values: {allowed_str}" + return + + return validator + + +def bool_validator(value) -> str | None: + """ + Validates that a value is of type bool. + + Args: + value: The value to validate. + + Returns: + None if valid, otherwise a descriptive string. + """ + + if not isinstance(value, bool): + return f"Invalid type: expected bool, got {type(value).__name__}" + return + + +def hostname_or_ip_address_and_optional_port_validator( + value: str, accept_blank: bool = False, accept_port: bool = True +) -> str | None: + """ + Wrapper for HostnameOrIPAddressAndOptionalPortValidator to adapt to config schema validation. + + Args: + value: The string value to validate. + + Returns: + None if valid, otherwise a string describing the validation error. + """ + + if value is None or value.strip() == "": + if accept_blank: + return + return "Value cannot be blank" + + validator = HostnameOrIPAddressAndOptionalPortValidator(accept_port=accept_port) + document = Document(text=value) + + try: + validator.validate(document) + return + except ValidationError as exc: + # Return the validation message as string + return exc.message + + +def certificate_subject_name_validator( + value: str, accept_blank: bool = False +) -> str | None: + """ + Wrapper for CertificateSubjectNameValidator to adapt to config schema validation. + + Args: + value: The string value to validate. + + Returns: + None if valid, otherwise a string describing the validation error. + """ + + if value is None or value.strip() == "": + if accept_blank: + return + return "Value cannot be blank" + + validator = CertificateSubjectNameValidator() + document = Document(text=value) + + try: + validator.validate(document) + return + except ValidationError as exc: + # Return the validation message as string + return exc.message diff --git a/step_cli_tools/validators.py b/step_cli_tools/validators.py deleted file mode 100644 index d3136ad..0000000 --- a/step_cli_tools/validators.py +++ /dev/null @@ -1,195 +0,0 @@ -# --- Standard library imports --- -import ipaddress -import re - -# --- Third-party imports --- -from questionary import ValidationError, Validator - - -__all__ = [ - "HostnamePortValidator", - "SHA256Validator", - "SHA256OrNameValidator", - "int_range_validator", - "str_allowed_validator", - "bool_validator", - "server_validator", -] - - -class HostnamePortValidator(Validator): - def validate(self, document): - value = document.text.strip() - - # Check if port is specified - if ":" in value: - host_part, port_part = value.rsplit(":", 1) - if not port_part.isdigit() or not (1 <= int(port_part) <= 65535): - raise ValidationError( - message=f"Invalid port: {port_part}. Must be between 1 and 65535.", - cursor_position=len(document.text), - ) - else: - host_part = value - - # Check if host is a valid IP address - try: - ipaddress.ip_address(host_part) - return - except ValueError: - pass - - # Check hostname validity - hostname_regex = re.compile( - r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(? Optional[str]: - - Returns None if valid. - - Returns a string describing the problem if invalid. - """ - - def validator(value): - if not isinstance(value, int): - return f"Invalid type: expected int, got {type(value).__name__}" - if value < min_value or value > max_value: - return f"Value {value} out of range ({min_value}โ€“{max_value})" - return None - - return validator - - -def str_allowed_validator(allowed: list[str]): - """ - Returns a validator function that ensures a string value is one of the allowed values. - - Args: - allowed: List of allowed string values. - - Returns: - A function(value) -> Optional[str]: - - Returns None if valid. - - Returns a descriptive string if invalid. - """ - - def validator(value): - if not isinstance(value, str): - return f"Invalid type: expected str, got {type(value).__name__}" - if value not in allowed: - allowed_str = ", ".join(map(repr, allowed)) - return f"Invalid value '{value}'. Allowed values: {allowed_str}" - return None - - return validator - - -def bool_validator(value) -> str | None: - """ - Validates that a value is of type bool. - - Args: - value: The value to validate. - - Returns: - None if valid, otherwise a descriptive string. - """ - - if not isinstance(value, bool): - return f"Invalid type: expected bool, got {type(value).__name__}" - return None - - -def server_validator(value: str) -> str | None: - """ - Validate a server string with optional port. - - Args: - value: A string like "hostname" or "hostname:port" or "127.0.0.1:8080". - - Returns: - None if valid, otherwise a descriptive string. - """ - - if not isinstance(value, str): - return f"Invalid type: expected string, got {type(value).__name__}" - - value = value.strip() - if not value: - # An empty string is allowed in the config file - return None - - # Split host and optional port - if ":" in value: - host_part, port_part = value.rsplit(":", 1) - if not port_part.isdigit() or not (1 <= int(port_part) <= 65535): - return f"Invalid port: {port_part}. Must be between 1 and 65535." - else: - host_part = value - - # Check if host is a valid IP address - try: - ipaddress.ip_address(host_part) - return None # valid IP - except ValueError: - pass - - # Validate hostname format - hostname_regex = re.compile( - r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?