""" Utility functions for the Dependabot Terraform module version updater. """ import json import logging import os import re import subprocess import time from typing import Dict, List, Tuple import requests import yaml from packaging.version import InvalidVersion, Version from . import config def execute_shell_command(command: str, cwd: str) -> Tuple[bool, str, str]: """ Execute a shell command. :param command: Command to execute. :param cwd: Working directory. :return: Tuple (success, stdout, stderr). """ try: result = subprocess.run( command, check=True, shell=True, cwd=cwd, timeout=15, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) return True, result.stdout, result.stderr except subprocess.CalledProcessError as e: return False, e.stdout, e.stderr except subprocess.TimeoutExpired as e: return False, "", f"Command timed out: {e}" except Exception as e: # pylint: disable=broad-exception-caught return False, "", f"Unexpected error: {e}" def get_terraform_modules(config_file: str) -> Tuple[bool, List[str]]: """ Get the list of Terraform modules from the configuration file. This function reads a YAML configuration file and extracts the list of Terraform modules specified in it. If no modules are found, a warning is logged. If an error occurs while reading the file, an error is logged. :param config_file: Path to the YAML configuration file. :return: Tuple containing a boolean indicating success, and a list of modules. """ try: with open(config_file, "r", encoding="UTF-8") as fp: dependabot_config = yaml.safe_load(fp) modules = dependabot_config.get("modules", []) if not modules: logging.warning( "No modules found in the configuration file %s.", config_file ) return True, modules except yaml.YAMLError as e: logging.error("Error reading configuration file %s: %s", config_file, e) return False, [] except IOError as e: logging.error("Error reading configuration file %s: %s", config_file, e) return False, [] def is_cache_expired(cache_file: str, cache_duration: int) -> bool: """ Check if the cache is expired based on CACHE_DURATION. :return: True if cache is expired, False otherwise. """ if os.path.exists(cache_file): cache_mtime = os.path.getmtime(cache_file) if time.time() - cache_mtime < cache_duration: return False logging.info("Cache expired. It will be refreshed automatically.") return True logging.info("Cache file not found. It will be created.") return True def load_module_versions_cache(cache_file: str) -> Tuple[bool, Dict[str, str]]: """ Load module versions from cache. :return: Cached module versions or empty dict if cache doesn't exist. """ if os.path.exists(cache_file): try: with open(cache_file, "r", encoding="UTF-8") as fp: return True, json.load(fp) except OSError as e: logging.error("Error reading cache file: %s", e) except json.JSONDecodeError: logging.warning("Cache file is corrupted. It needs to be refreshed.") return False, {} return False, {} def save_module_versions_cache( cache_file: str, module_versions: Dict[str, str] ) -> bool: """ Save module versions to cache. :param cache_file: Path to the cache file. :param module_versions: Module versions to cache. """ try: json_string = json.dumps(module_versions) with open(cache_file, "w", encoding="UTF-8") as fp: fp.write(json_string) return True except TypeError as e: logging.error("Error saving cache file: %s", e) return False except IOError as e: logging.error("Error saving cache file: %s", e) return False def get_all_module_versions(tf_modules: List[str]) -> Dict[str, str]: """ Get the latest version of all modules with retries in case of timeouts. :param tf_modules: List of module names. :return: Dict of module versions. """ module_versions = {} max_retries = ( 3 # Number of times to retry on timeout (or other exceptions if desired) ) backoff_seconds = 2 # Time to wait between retries logging.info("Fetching latest module versions from GitHub...") for tf_module in tf_modules: repo_url = f"{config.GITHUB_REPOS_API_BASE_URL}/{tf_module}/releases/latest" # We'll try up to max_retries times in case of request failures for attempt in range(1, max_retries + 1): try: response = requests.get( repo_url, headers=config.GITHUB_REPOS_API_HEADERS, timeout=10 ) # Raise an exception for non-200 HTTP status codes response.raise_for_status() # If everything is fine, parse the version and break latest_version = response.json().get("tag_name", "unknown") module_versions[tf_module] = latest_version logging.debug( "Fetched latest version for %s: %s (attempt %d)", tf_module, latest_version, attempt, ) break # We have our version, no need for further retries except requests.exceptions.Timeout: logging.warning( "Timeout occurred for %s on attempt %d/%d", tf_module, attempt, max_retries, ) # If we've exhausted our retries, set as unknown if attempt == max_retries: logging.error("Max retries reached for %s, giving up.", tf_module) module_versions[tf_module] = "unknown" else: # Wait a bit before retrying time.sleep(backoff_seconds) except requests.exceptions.RequestException as e: # This handles any other request-related errors (DNS, SSL, etc.) logging.error( "Request failed for %s on attempt %d/%d: %s", tf_module, attempt, max_retries, str(e), ) module_versions[tf_module] = "unknown" # No point in retrying if it's a hard error (e.g., 404 or bad SSL) break # If the loop ended without success and wasn't set, mark as unknown if tf_module not in module_versions: module_versions[tf_module] = "unknown" return module_versions def get_latest_terraform_version() -> str: """ Get the latest Terraform version. :return: Latest Terraform version. """ try: response = requests.get( "https://releases.hashicorp.com/terraform/index.json", timeout=10, ) response.raise_for_status() releases = response.json() versions = releases.get("versions", {}) valid_versions = [] for v in versions.keys(): try: parsed = Version(v) if not parsed.is_prerelease: valid_versions.append(v) except InvalidVersion: logging.warning("Invalid version format: %s", v) continue if not valid_versions: logging.error("No valid versions found in the response.") return "1.11.4" latest_version = str(max(valid_versions, key=Version)) return latest_version except requests.exceptions.RequestException as e: logging.error("Error fetching latest Terraform version: %s", e) return "1.11.4" # Fallback to a known version if the request fails def parse_source_url(source: str) -> Tuple[str, str]: """ Extracts the Git repository URL and version (ref) from a Terraform module source URL. Supports only SSH GitHub URLs in the form: - git@github.com:org/repo.git?ref=v1.0.0 - git@github.com:org/repo.git - git@github.com:org/repo?ref=v1.0.0 - Includes Terraform module paths which are ignored. :param source: The source URL. :return: A tuple of (normalized_git_url, version) """ # git@github.com:org/repo.git?ref=v1.0.0 if not source: raise ValueError("Source URL is empty.") # Extract ref query parameter (if present) ref_match = re.search(r"\?ref=([^&]+)", source) ref = ref_match.group(1) if ref_match else "master" source_url = source.split("?ref")[0] or source if "//" in source_url: parts = source_url.split("//") source_url = parts[0] module_path = parts[1] if not source_url.endswith(".git"): source_url += ".git" tmp_source = f"{source_url}//{module_path}" else: tmp_source = source_url if not tmp_source.endswith(".git"): tmp_source += ".git" tmp_source += "//" return tmp_source, ref def extract_module_name(source_url: str) -> str: """ Extracts module name from the source URL. :param source_url: The source URL. :return: Module name. """ logging.debug(source_url) match = re.search(r"git@github\.com:[^/]+/([^/]+).git", source_url) logging.debug(match) return match.group(1) if match else "" def parse_and_update_terraform_content( content: str, module_versions: Dict[str, str], latest_terraform_version: str ) -> Tuple[str, bool]: """ Parse and update Terraform content with the latest module versions. :param content: The content of a Terraform file. :param module_versions: Dictionary of module names and their latest versions. :return: Tuple of updated content and a boolean indicating if updates were made. """ lines = content.splitlines(keepends=True) updated_content = [] file_updated = False for line in lines: # pylint: disable=too-many-nested-blocks if "source" in line and config.TERRAFORM_MODULE_SOURCE_PREFIX in line: match = re.search(r'source\s*=\s*"([^"]+)"', line) if match: source = match.group(1) if source.startswith(config.TERRAFORM_MODULE_SOURCE_PREFIX): source_url, current_version = parse_source_url(source) module_name = extract_module_name(source_url) logging.debug( "Source URL: %s, Module name: %s, Current version: %s", source_url, module_name, current_version, ) if module_name and module_name in module_versions: latest_version = module_versions[module_name] logging.debug("Latest version: %s", latest_version) if latest_version not in (current_version, "unknown"): line = line.replace( source, f"{source_url}?ref={latest_version}" ) file_updated = True logging.info( "Updating module %s from %s to %s", module_name, current_version, latest_version, ) else: logging.warning( "Could not extract module name or version for source URL: %s", source_url, ) elif "required_version" in line: match = re.search(r"required_version\s*=\s*\"(.*)\"", line) if match: current_version = match.group(1) if latest_terraform_version not in (current_version, "unknown"): line = line.replace(current_version, latest_terraform_version) file_updated = True logging.info( "Updating Terraform version from %s to %s", current_version, latest_terraform_version, ) updated_content.append(line) return "".join(updated_content), file_updated def parse_and_update_terraform_file( file_path: str, module_versions: Dict[str, str], latest_terraform_version: str, dry_run: bool, ) -> bool: """ Parse and update a Terraform file with the latest module versions. :param file_path: :param module_versions: :param dry_run: :return: """ try: # Read the file content with open(file_path, "r", encoding="UTF-8") as file: content = file.read() # Update the content based on module versions updated_content, file_updated = parse_and_update_terraform_content( content, module_versions, latest_terraform_version ) # If changes were made, write the updated content back to the file if file_updated and not dry_run: with open(file_path, "w", encoding="UTF-8") as file: file.write(updated_content) format_terraform_file(file_path) return file_updated except IOError as e: logging.error("Error processing file %s: %s", file_path, e) return False def format_terraform_file(file_path: str) -> None: """Formats the given Terraform file using `terraform fmt`.""" execute_shell_command(f"terraform fmt {file_path}", os.path.dirname(file_path))