From 37bead7623462809683a2e720a5c35a11424fd0f Mon Sep 17 00:00:00 2001 From: Buba Date: Tue, 27 Jan 2026 18:35:35 +0000 Subject: [PATCH 1/2] Configure and enable ruff formatter --- .pre-commit-config.yaml | 3 +- ruff.toml | 17 +- scripts/github-star-repo.py | 6 +- scripts/infra-mcp/server.py | 16 +- .../tools/collections/container_tools.py | 30 ++- .../infra-mcp/tools/collections/task_tools.py | 16 +- scripts/infra-mcp/tools/get_app_icon.py | 28 +-- .../tools/get_container_categories.py | 8 +- scripts/infra-mcp/tools/get_container_tags.py | 202 +++++++++--------- .../infra-mcp/tools/get_dashboard_groups.py | 2 +- scripts/labctl.py | 110 ++++++---- scripts/proxy-request-log.py | 6 +- scripts/restructure-services.py | 4 +- scripts/update-example-env.py | 44 ++-- 14 files changed, 257 insertions(+), 235 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b9fb174c..57b7d4ed 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -49,9 +49,8 @@ repos: # Linter - https://docs.astral.sh/ruff/linter/ - id: ruff-check args: ["--fix"] - # TODO Enable after pending MRs are merged # Formatter - https://docs.astral.sh/ruff/formatter/ - # - id: ruff-format + - id: ruff-format # Dockerfile linter, validate inline bash, written in Haskell - repo: https://github.com/hadolint/hadolint diff --git a/ruff.toml b/ruff.toml index 958403ee..5ecdc256 100644 --- a/ruff.toml +++ b/ruff.toml @@ -1,7 +1,13 @@ # https://docs.astral.sh/ruff/settings/ target-version = "py313" -line-length = 120 +line-length = 150 + +# Exclude third-party scripts +exclude = [ + "scripts/git-filter-repo.py", + "scripts/test-colors.py", +] [lint] select = [ @@ -27,11 +33,10 @@ ignore = [ "TRY003", # Avoid specifying long messages outside the exception class ] -# Exclude third-party scripts -exclude = [ - "scripts/git-filter-repo.py", - "scripts/test-colors.py", -] +[lint.pylint] +max-returns = 10 # Increased from default 6 +max-branches = 20 # Increased from default 12 +max-statements = 75 # Increased from default 50 [lint.isort] known-first-party = [ diff --git a/scripts/github-star-repo.py b/scripts/github-star-repo.py index 4b85dba7..229ea27b 100755 --- a/scripts/github-star-repo.py +++ b/scripts/github-star-repo.py @@ -19,8 +19,8 @@ def star_github_repo(repo_url: str) -> None: """ try: parsed_url = urlparse(repo_url) - repo_path = unquote(parsed_url.path.strip('/')) - owner, repo_name = repo_path.split('/') + repo_path = unquote(parsed_url.path.strip("/")) + owner, repo_name = repo_path.split("/") # Construct the API endpoint URL api_url = f"https://api.github.com/user/starred/{owner}/{repo_name}" @@ -31,7 +31,7 @@ def star_github_repo(repo_url: str) -> None: # Make the PUT request to star the repository headers = { "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json" + "Accept": "application/vnd.github+json", } response = requests.put(api_url, headers=headers, timeout=30) diff --git a/scripts/infra-mcp/server.py b/scripts/infra-mcp/server.py index 0f4ddccc..ce71ef22 100755 --- a/scripts/infra-mcp/server.py +++ b/scripts/infra-mcp/server.py @@ -24,13 +24,13 @@ from utils.security import validate_url_for_ssrf # Configure logging -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') +logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger("infra-mcp") logger.info("Starting Infra MCP server") mcp = FastMCP( name="infra-mcp", - instructions="Use these tools to configure the homelab infrastructure and interact with the services." + instructions="Use these tools to configure the homelab infrastructure and interact with the services.", ) @@ -121,7 +121,7 @@ class Args: args.registry = None tags, _, _, _ = tag_finder.get_image_tags(args) - return [tag['name'] for tag in tags[:limit]] if tags else [] + return [tag["name"] for tag in tags[:limit]] if tags else [] except Exception: logger.exception("list-container-tags failed for image=%r", image) return [] @@ -156,7 +156,7 @@ class Args: # Get same hash tags but don't output to stdout same_hash_tags = tag_finder.list_same_hash_tags(args, suppress_output=True) - return [tag['name'] for tag in same_hash_tags] if same_hash_tags else [] + return [tag["name"] for tag in same_hash_tags] if same_hash_tags else [] except Exception: logger.exception("list-same-hash-container-tags failed for image=%r tag=%r", image, tag) return [] @@ -194,9 +194,9 @@ class Args: most_specific = tag_finder.get_most_specific_tag(args) same_hash = tag_finder.list_same_hash_tags(args, suppress_output=True) if most_specific: - return most_specific['name'] + return most_specific["name"] elif same_hash: - return same_hash[0]['name'] + return same_hash[0]["name"] else: return tag or "latest" except Exception: @@ -212,8 +212,8 @@ class Args: logger.info(f"Repository root path: {repository_root_path}") # Check environment variables to enable/disable tools - enable_task_tools = os.environ.get('ENABLE_TASK_TOOLS', 'true').lower() != 'false' - enable_container_tools = os.environ.get('ENABLE_CONTAINER_TOOLS', 'true').lower() != 'false' + enable_task_tools = os.environ.get("ENABLE_TASK_TOOLS", "true").lower() != "false" + enable_container_tools = os.environ.get("ENABLE_CONTAINER_TOOLS", "true").lower() != "false" # Add tools based on environment variable settings if enable_task_tools: diff --git a/scripts/infra-mcp/tools/collections/container_tools.py b/scripts/infra-mcp/tools/collections/container_tools.py index 435d78d3..c09e562b 100644 --- a/scripts/infra-mcp/tools/collections/container_tools.py +++ b/scripts/infra-mcp/tools/collections/container_tools.py @@ -25,12 +25,12 @@ def get_container_operations(): A list of dictionaries with operation name and description """ return [ - {'name': 'pull', 'description': "Pull the latest container image for the specified service"}, - {'name': 'up', 'description': "Start the specified service containers"}, - {'name': 'down', 'description': "Stop the specified service containers"}, - {'name': 'restart', 'description': "Restart the specified service containers"}, - {'name': 'recreate', 'description': "Recreate the specified service containers"}, - {'name': 'config', 'description': "Show the docker-compose configuration for the specified service"}, + {"name": "pull", "description": "Pull the latest container image for the specified service"}, + {"name": "up", "description": "Start the specified service containers"}, + {"name": "down", "description": "Stop the specified service containers"}, + {"name": "restart", "description": "Restart the specified service containers"}, + {"name": "recreate", "description": "Recreate the specified service containers"}, + {"name": "config", "description": "Show the docker-compose configuration for the specified service"}, ] @@ -47,12 +47,12 @@ def execute_container_operation(operation: str, service_name: str, repository_ro The command output as a string """ # Validate operation - valid_operations = {op['name'] for op in get_container_operations()} + valid_operations = {op["name"] for op in get_container_operations()} if operation not in valid_operations: return f"Invalid operation: {operation}" # Validate service_name format - if not re.match(r'^[a-zA-Z0-9_/-]+$', service_name): + if not re.match(r"^[a-zA-Z0-9_/-]+$", service_name): return f"Invalid service name format: {service_name}" cmd = [ @@ -60,15 +60,12 @@ def execute_container_operation(operation: str, service_name: str, repository_ro os.path.join(repository_root_path, "scripts", "labctl.py"), "service", operation, - service_name + service_name, ] try: result = subprocess.run( # noqa: S603 - cmd, - capture_output=True, - text=True, - check=True + cmd, capture_output=True, text=True, check=True ) except subprocess.CalledProcessError as e: return f"Error running operation: {e.stderr or str(e)}" @@ -87,6 +84,7 @@ def create_operation_function(op: str, repository_root_path: str) -> Callable[[s Returns: A callable function that executes the operation on a given service """ + def operation_fn(service_name: str) -> str: """ Execute one operation on the specified service and return the output @@ -113,15 +111,15 @@ def add_container_operation_tools(mcp_server: FastMCP, repository_root_path: str operations = get_container_operations() for op in operations: - operation_fn = create_operation_function(op['name'], repository_root_path) + operation_fn = create_operation_function(op["name"], repository_root_path) tool_name = f"container-service-{op['name']}" - description = op['description'] + description = op["description"] tool = Tool.from_function( fn=operation_fn, name=tool_name, title=tool_name, - description=description + description=description, ) mcp_server.add_tool(tool) diff --git a/scripts/infra-mcp/tools/collections/task_tools.py b/scripts/infra-mcp/tools/collections/task_tools.py index 23788527..73a606eb 100644 --- a/scripts/infra-mcp/tools/collections/task_tools.py +++ b/scripts/infra-mcp/tools/collections/task_tools.py @@ -36,7 +36,7 @@ def get_task_list(repository_root_path: str) -> list[dict[str, str]]: [task_bin, "--list-all", "--dir", repository_root_path], capture_output=True, text=True, - check=True + check=True, ) except subprocess.CalledProcessError: logger.exception("Error getting task list") @@ -46,14 +46,13 @@ def get_task_list(repository_root_path: str) -> list[dict[str, str]]: # Parse output lines for line in result.stdout.splitlines(): # Match lines like "* task_name: task description" - match = re.match(r'^\*\s+(.+?):\s+(.+)$', line.strip()) + match = re.match(r"^\*\s+(.+?):\s+(.+)$", line.strip()) if match: task_name = match.group(1).strip() description = match.group(2).strip() - tasks.append({ - "name": task_name, - "description": description - }) + tasks.append( + {"name": task_name, "description": description}, + ) logger.debug(f"Found {len(tasks)} tasks") return tasks @@ -80,7 +79,7 @@ def execute_task(task_name: str, repository_root_path: str) -> str: [task_bin, task_name, "--dir", repository_root_path], capture_output=True, text=True, - check=True + check=True, ).stdout.strip() except subprocess.CalledProcessError as e: logger.exception(f"Error executing task {task_name}") @@ -98,6 +97,7 @@ def create_task_function(task_name: str, repository_root_path: str) -> Callable[ Returns: A callable function that executes the task """ + def task_fn() -> str: return execute_task(task_name, repository_root_path) @@ -124,7 +124,7 @@ def add_task_tools(mcp_server: FastMCP, repository_root_path: str) -> None: fn=task_fn, name=tool_name, title=tool_name, - description=description + description=description, ) mcp_server.add_tool(tool) diff --git a/scripts/infra-mcp/tools/get_app_icon.py b/scripts/infra-mcp/tools/get_app_icon.py index fb0ac754..c4ebd461 100755 --- a/scripts/infra-mcp/tools/get_app_icon.py +++ b/scripts/infra-mcp/tools/get_app_icon.py @@ -20,7 +20,7 @@ def __init__(self): Initialize the AppIconFinder with default headers for HTTP requests. """ self.headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } def get_app_icon(self, app_name, homepage_url): @@ -57,7 +57,7 @@ def _find_dashboard_icon(self, app_name): url, headers=self.headers, timeout=10, - allow_redirects=True + allow_redirects=True, ) if response.ok: return icon_name @@ -81,42 +81,42 @@ def _find_favicon_url(self, homepage_url): # Make sure URL has a scheme homepage_url = homepage_url.strip() - if not homepage_url.startswith(('http://', 'https://')): - homepage_url = 'https://' + homepage_url + if not homepage_url.startswith(("http://", "https://")): + homepage_url = "https://" + homepage_url # Fetch the homepage response = requests.get(homepage_url, headers=self.headers, timeout=10) response.raise_for_status() # Parse the HTML - soup = BeautifulSoup(response.text, 'html.parser') + soup = BeautifulSoup(response.text, "html.parser") # Look for favicon in different ways # 1. Check for link tags with rel="icon" or rel="shortcut icon" - icon_links = soup.find_all('link', rel=re.compile(r'(shortcut icon|icon|apple-touch-icon)', re.I)) + icon_links = soup.find_all("link", rel=re.compile(r"(shortcut icon|icon|apple-touch-icon)", re.I)) if icon_links: # Sort by preference: apple-touch-icon > icon > shortcut icon def get_priority(link): - rel_attr = link.get('rel', []) + rel_attr = link.get("rel", []) if isinstance(rel_attr, str): rel_attr = [rel_attr] - rel_lower = ' '.join(rel_attr).lower() - if 'apple-touch-icon' in rel_lower: + rel_lower = " ".join(rel_attr).lower() + if "apple-touch-icon" in rel_lower: return 3 - elif 'icon' in rel_lower and 'shortcut' not in rel_lower: + elif "icon" in rel_lower and "shortcut" not in rel_lower: return 2 else: return 1 icon_links = sorted(icon_links, key=get_priority, reverse=True) for link in icon_links: - if 'href' in link.attrs: + if "href" in link.attrs: # Make relative URLs absolute - favicon_url = urljoin(homepage_url, link['href']) + favicon_url = urljoin(homepage_url, link["href"]) return favicon_url # 2. Check for the default location - default_favicon = urljoin(homepage_url, '/favicon.ico') + default_favicon = urljoin(homepage_url, "/favicon.ico") favicon_response = requests.head(default_favicon, headers=self.headers, timeout=5) if favicon_response.status_code == 200: return default_favicon @@ -143,7 +143,7 @@ def test_icon_finder(): ("Radarr", "radarr.video"), ("Grafana", "grafana.com"), ("Jellyfin", "jellyfin.org"), - ("HUP", "hup.hu") + ("HUP", "hup.hu"), ] for app_name, homepage in test_cases: diff --git a/scripts/infra-mcp/tools/get_container_categories.py b/scripts/infra-mcp/tools/get_container_categories.py index 7aa96435..c4b883a4 100755 --- a/scripts/infra-mcp/tools/get_container_categories.py +++ b/scripts/infra-mcp/tools/get_container_categories.py @@ -9,7 +9,7 @@ from utils.git import get_git_root except ModuleNotFoundError: # If that fails, try relative import (when run as a standalone script) - sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from utils.git import get_git_root @@ -57,16 +57,16 @@ def get_container_categories(self) -> list[str]: has_readme = readme_path.exists() and readme_path.is_file() # Check for any yaml files in the current directory - has_yaml = any(file.lower().endswith(('.yaml', '.yml')) for file in files) + has_yaml = any(file.lower().endswith((".yaml", ".yml")) for file in files) # Only include directories with both README.md and at least one yaml file if has_readme and has_yaml: # Get the path relative to the docker directory rel_path = root_path.relative_to(self.docker_path) # Convert to string and use forward slashes for consistency - rel_path_str = str(rel_path).replace('\\', '/') + rel_path_str = str(rel_path).replace("\\", "/") # Add the directory to categories if it's not the root docker directory - if rel_path_str != '.': + if rel_path_str != ".": categories.append(rel_path_str) return sorted(categories) diff --git a/scripts/infra-mcp/tools/get_container_tags.py b/scripts/infra-mcp/tools/get_container_tags.py index 419c9b61..317b97fc 100755 --- a/scripts/infra-mcp/tools/get_container_tags.py +++ b/scripts/infra-mcp/tools/get_container_tags.py @@ -30,9 +30,9 @@ def _parse_arch(self, arch: str) -> tuple[str, str]: Returns: tuple: A tuple containing (os_part, arch_part) """ - parts = arch.split('/') - os_part = parts[0] if parts else 'linux' - arch_part = parts[1] if len(parts) > 1 else 'amd64' + parts = arch.split("/") + os_part = parts[0] if parts else "linux" + arch_part = parts[1] if len(parts) > 1 else "amd64" return os_part, arch_part def _parse_version(self, tag_name: str) -> tuple[int, ...] | None: @@ -53,7 +53,7 @@ def _parse_version(self, tag_name: str) -> tuple[int, ...] | None: """ # Handle common prefixes normalized = tag_name.lower() - if normalized.startswith('v') and len(normalized) > 1 and normalized[1].isdigit(): + if normalized.startswith("v") and len(normalized) > 1 and normalized[1].isdigit(): normalized = normalized[1:] # Skip non-version tags @@ -65,12 +65,12 @@ def _parse_version(self, tag_name: str) -> tuple[int, ...] | None: return None # Extract the version part (before any '-', '_', or non-numeric suffix) - version_part = normalized.split('-')[0].split('_')[0] + version_part = normalized.split("-")[0].split("_")[0] # Split by '.' and try to parse as integers try: version_numbers = [] - for part in version_part.split('.'): + for part in version_part.split("."): # Only take numeric parts if part.isdigit(): version_numbers.append(int(part)) @@ -100,7 +100,7 @@ def _parse_datetime(self, datetime_str: str | None) -> datetime: if not datetime_str: return datetime.min try: - return datetime.fromisoformat(datetime_str.replace('Z', '+00:00')) + return datetime.fromisoformat(datetime_str.replace("Z", "+00:00")) except (ValueError, AttributeError): return datetime.min @@ -116,9 +116,9 @@ def _extract_arch_digest(self, tag: dict[str, Any], architecture: str) -> str | str: The digest for the specified architecture, or None if not found """ arch_os, arch_variant = self._parse_arch(architecture) - for image in tag.get('images', []): - if image.get('architecture') == arch_variant and image.get('os') == arch_os: - return image.get('digest') + for image in tag.get("images", []): + if image.get("architecture") == arch_variant and image.get("os") == arch_os: + return image.get("digest") return None def _create_tag_data_dict(self, tag: dict[str, Any], architecture: str) -> dict[str, Any]: @@ -133,10 +133,10 @@ def _create_tag_data_dict(self, tag: dict[str, Any], architecture: str) -> dict[ dict: Standardized tag data dictionary with name, last_updated, size, and digest """ return { - 'name': tag['name'], - 'last_updated': tag.get('last_updated'), - 'size': tag.get('full_size', 0), - 'digest': self._extract_arch_digest(tag, architecture) + "name": tag["name"], + "last_updated": tag.get("last_updated"), + "size": tag.get("full_size", 0), + "digest": self._extract_arch_digest(tag, architecture), } def _version_sort_key(self, tag: dict[str, Any]) -> tuple: @@ -154,22 +154,22 @@ def _version_sort_key(self, tag: dict[str, Any]) -> tuple: Returns: tuple: A sort key that can be used with sorted() or list.sort() """ - tag_name = tag['name'].lower() + tag_name = tag["name"].lower() # Priority 1: 'latest' tag - if tag_name == 'latest': + if tag_name == "latest": return (2, (999, 999, 999, 999), datetime.max) # Priority 2: Version tags - version = self._parse_version(tag['name']) + version = self._parse_version(tag["name"]) if version: # Pad version tuple to 4 elements for consistent comparison padded_version = version + (0,) * (4 - len(version)) - updated = self._parse_datetime(tag.get('last_updated')) + updated = self._parse_datetime(tag.get("last_updated")) return (1, padded_version[:4], updated) # Priority 3: Non-version tags - updated = self._parse_datetime(tag.get('last_updated')) + updated = self._parse_datetime(tag.get("last_updated")) return (0, (0, 0, 0, 0), updated) def _sort_tags(self, tag_data: list[dict[str, Any]], sort_by: str) -> None: @@ -184,10 +184,12 @@ def _sort_tags(self, tag_data: list[dict[str, Any]], sort_by: str) -> None: tag_data.sort(key=self._version_sort_key, reverse=True) elif sort_by == "updated": # Sort by last_updated timestamp - tag_data.sort(key=lambda x: self._parse_datetime(x.get('last_updated')), reverse=True) + tag_data.sort(key=lambda x: self._parse_datetime(x.get("last_updated")), reverse=True) # else: sort_by == "default", keep original order - def get_docker_hub_tags(self, image_name: str, limit: int = 10, architecture: str = "linux/amd64", sort_by: str = "version") -> list[dict[str, Any]]: + def get_docker_hub_tags( + self, image_name: str, limit: int = 10, architecture: str = "linux/amd64", sort_by: str = "version" + ) -> list[dict[str, Any]]: """Query Docker Hub for image tags with timestamp information. Args: @@ -200,10 +202,10 @@ def get_docker_hub_tags(self, image_name: str, limit: int = 10, architecture: st list: List of tag dictionaries sorted according to sort_by parameter """ # Parse repository name - if '/' in image_name: - namespace, repo = image_name.split('/', 1) + if "/" in image_name: + namespace, repo = image_name.split("/", 1) else: - namespace = 'library' # Official images are in the 'library' namespace + namespace = "library" # Official images are in the 'library' namespace repo = image_name url: str = f"https://hub.docker.com/v2/repositories/{namespace}/{repo}/tags?page_size=100" @@ -213,16 +215,16 @@ def get_docker_hub_tags(self, image_name: str, limit: int = 10, architecture: st data = response.json() tag_data: list[dict[str, Any]] = [] - for tag in data.get('results', []): + for tag in data.get("results", []): tag_data.append(self._create_tag_data_dict(tag, architecture)) # Handle pagination if there are more tags - while 'next' in data and data['next'] and len(tag_data) < 1000: # Limit to avoid too many requests - response = requests.get(data['next'], timeout=30) + while "next" in data and data["next"] and len(tag_data) < 1000: # Limit to avoid too many requests + response = requests.get(data["next"], timeout=30) response.raise_for_status() data = response.json() - for tag in data.get('results', []): + for tag in data.get("results", []): tag_data.append(self._create_tag_data_dict(tag, architecture)) # Sort based on sort_by parameter @@ -233,7 +235,14 @@ def get_docker_hub_tags(self, image_name: str, limit: int = 10, architecture: st else: return tag_data - def get_registry_tags(self, registry_url: str, image_name: str, limit: int = 10, architecture: str = "linux/amd64", sort_by: str = "version") -> list[dict[str, Any]]: + def get_registry_tags( + self, + registry_url: str, + image_name: str, + limit: int = 10, + architecture: str = "linux/amd64", + sort_by: str = "version", + ) -> list[dict[str, Any]]: """Query a registry API v2 for image tags and attempt to get creation time. Args: @@ -251,7 +260,7 @@ def get_registry_tags(self, registry_url: str, image_name: str, limit: int = 10, response = requests.get(url, timeout=30) response.raise_for_status() data = response.json() - tags: list[str] = data.get('tags', []) + tags: list[str] = data.get("tags", []) # For Docker Registry API v2, we need to make additional requests to get manifest and timestamps tag_data: list[dict[str, Any]] = [] @@ -259,7 +268,7 @@ def get_registry_tags(self, registry_url: str, image_name: str, limit: int = 10, manifest_url = f"{registry_url}/v2/{image_name}/manifests/{tag}" try: # Try to get the manifest to extract creation time - headers = {'Accept': 'application/vnd.docker.distribution.manifest.v2+json'} + headers = {"Accept": "application/vnd.docker.distribution.manifest.v2+json"} manifest_response = requests.get(manifest_url, headers=headers, timeout=30) manifest_response.raise_for_status() @@ -269,31 +278,25 @@ def get_registry_tags(self, registry_url: str, image_name: str, limit: int = 10, # For multi-arch images, we need to find the digest for the specific architecture digest = None # Try to parse the architecture from the manifest if it's a multi-arch image - if 'manifests' in manifest: - for m in manifest.get('manifests', []): - if m.get('platform', {}).get('architecture') == architecture.split('/')[1] and \ - m.get('platform', {}).get('os') == architecture.split('/')[0]: - digest = m.get('digest') + if "manifests" in manifest: + for m in manifest.get("manifests", []): + if ( + m.get("platform", {}).get("architecture") == architecture.split("/")[1] + and m.get("platform", {}).get("os") == architecture.split("/")[0] + ): + digest = m.get("digest") break else: # If it's not a multi-arch manifest, just use the digest directly - digest = manifest_response.headers.get('Docker-Content-Digest') + digest = manifest_response.headers.get("Docker-Content-Digest") # Most registry implementations don't expose creation time directly in the API # We'll use the response headers as a rough proxy for recency - last_modified = manifest_response.headers.get('Last-Modified') - tag_data.append({ - 'name': tag, - 'last_updated': last_modified, - 'digest': digest - }) + last_modified = manifest_response.headers.get("Last-Modified") + tag_data.append({"name": tag, "last_updated": last_modified, "digest": digest}) except requests.exceptions.RequestException: # If we can't get detailed info, just use the tag name - tag_data.append({ - 'name': tag, - 'last_updated': None, - 'digest': None - }) + tag_data.append({"name": tag, "last_updated": None, "digest": None}) # Sort based on sort_by parameter self._sort_tags(tag_data, sort_by) @@ -309,13 +312,13 @@ def _format_datetime(self, datetime_str: str | None) -> str: return "Unknown" try: # Parse ISO format - dt = datetime.fromisoformat(datetime_str.replace('Z', '+00:00')) - return dt.strftime('%Y-%m-%d %H:%M:%S UTC') + dt = datetime.fromisoformat(datetime_str.replace("Z", "+00:00")) + return dt.strftime("%Y-%m-%d %H:%M:%S UTC") except (TypeError, ValueError): # Try to parse HTTP date format try: dt = parsedate_to_datetime(datetime_str) - return dt.strftime('%Y-%m-%d %H:%M:%S UTC') + return dt.strftime("%Y-%m-%d %H:%M:%S UTC") except Exception: return datetime_str @@ -323,7 +326,7 @@ def _format_size(self, size_bytes: int | None) -> str: """Format size in bytes to human-readable format.""" if size_bytes is None: return "Unknown" - for unit in ['B', 'KB', 'MB', 'GB']: + for unit in ["B", "KB", "MB", "GB"]: if size_bytes < 1024.0: return f"{size_bytes:.2f} {unit}" size_bytes /= 1024.0 @@ -344,7 +347,7 @@ def get_tags_by_digest(self, tags: list[dict[str, Any]], target_digest: str | No if not target_digest: return [] - return [tag for tag in tags if tag.get('digest') == target_digest] + return [tag for tag in tags if tag.get("digest") == target_digest] def _determine_tag_specificity(self, tag: str) -> int: """Determine how specific a version tag is, higher is more specific. @@ -368,8 +371,8 @@ def _determine_tag_specificity(self, tag: str) -> int: # Skip tags that start with non-digits and aren't version tags # Allow common 'v' prefix before digits (e.g., v1.2.3) - normalized = tag[1:] if tag.startswith('v') and len(tag) > 1 else tag - if not normalized[0].isdigit() and '-' not in normalized: + normalized = tag[1:] if tag.startswith("v") and len(tag) > 1 else tag + if not normalized[0].isdigit() and "-" not in normalized: return -1 tag = normalized @@ -377,7 +380,7 @@ def _determine_tag_specificity(self, tag: str) -> int: score = 0 # Split by common separators - parts = tag.replace('-', '.').replace('_', '.').split('.') + parts = tag.replace("-", ".").replace("_", ".").split(".") # Count numeric segments (major.minor.patch get higher scores) numeric_parts = [] @@ -401,7 +404,7 @@ def _determine_tag_specificity(self, tag: str) -> int: score += len(parts) * 5 # Penalize 'latest' and 'stable' tags - if tag.lower() in ['latest', 'stable']: + if tag.lower() in ["latest", "stable"]: score = -100 return score @@ -417,11 +420,11 @@ def _parse_image_reference(self, image: str, registry: str | None = None) -> tup tuple: A tuple containing (registry_url, image_name, is_docker_hub) """ # Parse the image name to determine if it includes a registry - if '/' in image and ('.' in image.split('/')[0] or ':' in image.split('/')[0]): + if "/" in image and ("." in image.split("/")[0] or ":" in image.split("/")[0]): # This looks like a hostname with a port or domain name - parts = image.split('/') + parts = image.split("/") registry_host = parts[0] - image_name = '/'.join(parts[1:]) + image_name = "/".join(parts[1:]) registry_url = registry or f"https://{registry_host}" return registry_url, image_name, False else: @@ -430,7 +433,7 @@ def _parse_image_reference(self, image: str, registry: str | None = None) -> tup def _get_output_flags(self, args: argparse.Namespace, suppress_output: bool = False) -> tuple[bool, bool]: """Get output flag settings from args.""" - quiet = args.quiet if hasattr(args, 'quiet') else False + quiet = args.quiet if hasattr(args, "quiet") else False should_output = not suppress_output and not quiet return quiet, should_output @@ -438,7 +441,7 @@ def get_image_tags(self, args: argparse.Namespace, limit: int | None = None) -> """Get tags for the image based on the provided arguments.""" _, should_output = self._get_output_flags(args) fetch_limit = limit if limit is not None else args.limit - sort_by = getattr(args, 'sort', 'version') # Default to 'version' if not specified + sort_by = getattr(args, "sort", "version") # Default to 'version' if not specified registry_url, image_name, is_docker_hub = self._parse_image_reference(args.image, args.registry) @@ -457,19 +460,19 @@ def get_image_tags(self, args: argparse.Namespace, limit: int | None = None) -> def list_recent_tags(self, args: argparse.Namespace) -> None: """List recent tags for an image.""" # Get output flags - quiet = args.quiet if hasattr(args, 'quiet') else False + quiet = args.quiet if hasattr(args, "quiet") else False # Get tags for the image tags, _, _, _ = self.get_image_tags(args) if tags: # Only take up to limit - tags = tags[:args.limit] + tags = tags[: args.limit] if quiet: # In quiet mode, just output the tag names, one per line for tag in tags: - print(tag['name']) + print(tag["name"]) else: # Detailed output print(f"\nMost recent {len(tags)} tags for {args.image} ({args.architecture}):") @@ -477,9 +480,9 @@ def list_recent_tags(self, args: argparse.Namespace) -> None: print("-" * 95) for tag in tags: - updated = self._format_datetime(tag.get('last_updated')) - size = self._format_size(tag.get('size')) if 'size' in tag else 'N/A' - digest = self._format_digest(tag.get('digest')) + updated = self._format_datetime(tag.get("last_updated")) + size = self._format_size(tag.get("size")) if "size" in tag else "N/A" + digest = self._format_digest(tag.get("digest")) print(f"{tag['name']:<30} {updated:<30} {size:<15} {digest:<20}") elif not quiet: print(f"No tags found for {args.image}") @@ -503,15 +506,15 @@ def list_same_hash_tags(self, args: argparse.Namespace, suppress_output: bool = return [] # Find the target tag to get its digest - tag_name = args.tag if args.tag else all_tags[0]['name'] # Use first tag if none specified - target_tag = next((t for t in all_tags if t['name'] == tag_name), None) + tag_name = args.tag if args.tag else all_tags[0]["name"] # Use first tag if none specified + target_tag = next((t for t in all_tags if t["name"] == tag_name), None) if not target_tag: if should_output: print(f"Tag '{tag_name}' not found for {args.image}") return [] - target_digest = target_tag.get('digest') + target_digest = target_tag.get("digest") if not target_digest: if should_output: print(f"No digest found for tag '{tag_name}'") @@ -524,7 +527,7 @@ def list_same_hash_tags(self, args: argparse.Namespace, suppress_output: bool = if not suppress_output and quiet: # In quiet mode (but not suppressed), output the tag names, one per line for tag in same_hash_tags: - print(tag['name']) + print(tag["name"]) elif should_output: # Detailed output print(f"\nTags with the same digest as '{tag_name}' ({self._format_digest(target_digest)}) for {args.image}:") @@ -532,8 +535,8 @@ def list_same_hash_tags(self, args: argparse.Namespace, suppress_output: bool = print("-" * 75) for tag in same_hash_tags: - updated = self._format_datetime(tag.get('last_updated')) - size = self._format_size(tag.get('size')) if 'size' in tag else 'N/A' + updated = self._format_datetime(tag.get("last_updated")) + size = self._format_size(tag.get("size")) if "size" in tag else "N/A" print(f"{tag['name']:<30} {updated:<30} {size:<15}") elif should_output: print(f"No tags found with the same digest as '{tag_name}'") @@ -543,7 +546,7 @@ def list_same_hash_tags(self, args: argparse.Namespace, suppress_output: bool = def get_most_specific_tag(self, args: argparse.Namespace) -> dict[str, Any] | None: """Find the most specific version tag from a set of tags with the same hash.""" # Store quiet flag to local variable for easier access - quiet = args.quiet if hasattr(args, 'quiet') else False + quiet = args.quiet if hasattr(args, "quiet") else False # First, get all tags with the same hash # When in quiet mode, suppress output from list_same_hash_tags @@ -556,7 +559,7 @@ def get_most_specific_tag(self, args: argparse.Namespace) -> dict[str, Any] | No # Calculate the specificity score for each tag tag_scores: list[tuple[dict[str, Any], int]] = [] for tag in same_hash_tags: - score = self._determine_tag_specificity(tag['name']) + score = self._determine_tag_specificity(tag["name"]) tag_scores.append((tag, score)) # Sort by specificity score (highest first) @@ -567,14 +570,14 @@ def get_most_specific_tag(self, args: argparse.Namespace) -> dict[str, Any] | No if quiet: # Only output the final recommended tag - print(most_specific['name']) + print(most_specific["name"]) else: # Detailed output print("\nMost specific tag:") print(f"{'TAG':<30} {'SPECIFICITY SCORE':<20} {'LAST UPDATED':<30}") print("-" * 80) - updated = self._format_datetime(most_specific.get('last_updated')) + updated = self._format_datetime(most_specific.get("last_updated")) print(f"{most_specific['name']:<30} {tag_scores[0][1]:<20} {updated:<30}") # Show honorable mentions (other high scoring tags) @@ -582,7 +585,7 @@ def get_most_specific_tag(self, args: argparse.Namespace) -> dict[str, Any] | No print("\nOther version tags (sorted by specificity):") for tag, score in tag_scores[1:6]: # Show at most 5 other tags if score > 0: # Only show actual version tags - updated = self._format_datetime(tag.get('last_updated')) + updated = self._format_datetime(tag.get("last_updated")) print(f"{tag['name']:<30} {score:<20} {updated:<30}") print(f"\nRecommended tag to use: {most_specific['name']}") @@ -591,32 +594,35 @@ def get_most_specific_tag(self, args: argparse.Namespace) -> dict[str, Any] | No def main() -> None: - parser = argparse.ArgumentParser(description='Operations on container image tags') - parser.add_argument('--registry', help='Registry URL (defaults to Docker Hub if not specified)') - parser.add_argument('--architecture', default='linux/amd64', - help='Architecture to query for (default: linux/amd64)') - parser.add_argument('--quiet', action='store_true', help='Only output final results, no status or progress messages') + parser = argparse.ArgumentParser(description="Operations on container image tags") + parser.add_argument("--registry", help="Registry URL (defaults to Docker Hub if not specified)") + parser.add_argument("--architecture", default="linux/amd64", help="Architecture to query for (default: linux/amd64)") + parser.add_argument("--quiet", action="store_true", help="Only output final results, no status or progress messages") - subparsers = parser.add_subparsers(dest='command', help='Command to execute', required=True) + subparsers = parser.add_subparsers(dest="command", help="Command to execute", required=True) # Create the parser for the "list-recent" command - recent_parser = subparsers.add_parser('list-recent', help='List recent tags for an image') - recent_parser.add_argument('image', help='Image name (e.g., nginx or registry.example.com/nginx)') - recent_parser.add_argument('--limit', type=int, default=10, help='Maximum number of tags to display (default: 10)') - recent_parser.add_argument('--sort', choices=['version', 'updated', 'default'], default='version', - help='Sort order: version (by version number, default), updated (by last_updated timestamp), default (registry order)') + recent_parser = subparsers.add_parser("list-recent", help="List recent tags for an image") + recent_parser.add_argument("image", help="Image name (e.g., nginx or registry.example.com/nginx)") + recent_parser.add_argument("--limit", type=int, default=10, help="Maximum number of tags to display (default: 10)") + recent_parser.add_argument( + "--sort", + choices=["version", "updated", "default"], + default="version", + help="Sort order: version (by version number, default), updated (by last_updated timestamp), default (registry order)", + ) # Create the parser for the "list-same-hash" command - hash_parser = subparsers.add_parser('list-same-hash', help='List tags with the same hash') - hash_parser.add_argument('image', help='Image name (e.g., nginx or registry.example.com/nginx)') - hash_parser.add_argument('--tag', help='Tag to use as reference (default: latest or first tag found)') - hash_parser.add_argument('--limit', type=int, default=100, help='Maximum number of tags to search through (default: 100)') + hash_parser = subparsers.add_parser("list-same-hash", help="List tags with the same hash") + hash_parser.add_argument("image", help="Image name (e.g., nginx or registry.example.com/nginx)") + hash_parser.add_argument("--tag", help="Tag to use as reference (default: latest or first tag found)") + hash_parser.add_argument("--limit", type=int, default=100, help="Maximum number of tags to search through (default: 100)") # Create the parser for the "get-most-specific-tag" command - specific_parser = subparsers.add_parser('get-most-specific-tag', help='Find the most specific version tag from tags with the same hash') - specific_parser.add_argument('image', help='Image name (e.g., nginx or registry.example.com/nginx)') - specific_parser.add_argument('--tag', help='Tag to use as reference (default: latest or first tag found)') - specific_parser.add_argument('--limit', type=int, default=100, help='Maximum number of tags to search through (default: 100)') + specific_parser = subparsers.add_parser("get-most-specific-tag", help="Find the most specific version tag from tags with the same hash") + specific_parser.add_argument("image", help="Image name (e.g., nginx or registry.example.com/nginx)") + specific_parser.add_argument("--tag", help="Tag to use as reference (default: latest or first tag found)") + specific_parser.add_argument("--limit", type=int, default=100, help="Maximum number of tags to search through (default: 100)") # Initialize the container tag finder finder = ContainerTagFinder() diff --git a/scripts/infra-mcp/tools/get_dashboard_groups.py b/scripts/infra-mcp/tools/get_dashboard_groups.py index d9925f19..bfbd9d58 100755 --- a/scripts/infra-mcp/tools/get_dashboard_groups.py +++ b/scripts/infra-mcp/tools/get_dashboard_groups.py @@ -11,7 +11,7 @@ from utils.git import get_git_root except ModuleNotFoundError: # If that fails, try relative import (when run as a standalone script) - sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from utils.git import get_git_root diff --git a/scripts/labctl.py b/scripts/labctl.py index 8225cb4f..dd2700e1 100755 --- a/scripts/labctl.py +++ b/scripts/labctl.py @@ -24,6 +24,7 @@ @dataclass class DockerOptions: """Configuration options for Docker operations.""" + pull_before_start: bool = False quiet: bool = False # Log options @@ -35,7 +36,7 @@ class DockerOptions: # Global variables docker_stacks_dir: Path = (Path(__file__).resolve().parent.parent / "docker").resolve() -ALLOWED_STATES: tuple[str, ...] = ('pull', 'up', 'down', 'restart', 'recreate', 'config', 'logs') +ALLOWED_STATES: tuple[str, ...] = ("pull", "up", "down", "restart", "recreate", "config", "logs") def create_network_if_missing(network_name: str) -> None: @@ -137,9 +138,9 @@ def has_build_directive(compose_file: Path) -> bool: """ with open(compose_file) as f: yaml_content = yaml.safe_load(f) - if yaml_content and 'services' in yaml_content: - for service_config in yaml_content['services'].values(): - if 'build' in service_config: + if yaml_content and "services" in yaml_content: + for service_config in yaml_content["services"].values(): + if "build" in service_config: return True return False @@ -147,10 +148,10 @@ def has_build_directive(compose_file: Path) -> bool: def get_env_file_args(host_config_dir: Path, service_name: str) -> list[str]: """Get environment file arguments for Docker Compose with normalized paths.""" env_paths = [ - host_config_dir.parent / ".env", # Common .env file in config/docker - host_config_dir / ".env", # Host-specific .env file in config/docker/ + host_config_dir.parent / ".env", # Common .env file in config/docker + host_config_dir / ".env", # Host-specific .env file in config/docker/ host_config_dir.parent / f".env.{service_name}", # Common service-specific .env file in config/docker - host_config_dir / f".env.{service_name}" # Host- and service-specific .env file in config/docker/ + host_config_dir / f".env.{service_name}", # Host- and service-specific .env file in config/docker/ ] args = [] @@ -177,7 +178,13 @@ def docker(cmd: list[str], env=None, stdin=None, stdout=None, stderr=None) -> No subprocess.run([docker_bin, *cmd], env=env, stdin=stdin, stdout=stdout, stderr=stderr, check=True) # noqa: S603 -def docker_pull(stack_dir: Path, service_name: str, compose_file: Path, env_file_args: list[str], quiet: bool = False) -> None: +def docker_pull( + stack_dir: Path, + service_name: str, + compose_file: Path, + env_file_args: list[str], + quiet: bool = False, +) -> None: """Pull Docker images for a service. Args: @@ -224,7 +231,13 @@ def build_log_command_flags(options: DockerOptions) -> list[str]: return flags -def docker_command(host_config_dir: Path, stack_dir: Path, service_name: str, action: str, options: DockerOptions = None) -> None: +def docker_command( + host_config_dir: Path, + stack_dir: Path, + service_name: str, + action: str, + options: DockerOptions = None, +) -> None: """Execute Docker Compose command for a service. Args: @@ -298,7 +311,13 @@ def load_services_config(config_file: str) -> dict: sys.exit(1) -def process_services(host_config_dir: Path, config: dict, state_override: str | None = None, pull_before_start: bool = False, quiet: bool = False) -> None: +def process_services( + host_config_dir: Path, + config: dict, + state_override: str | None = None, + pull_before_start: bool = False, + quiet: bool = False, +) -> None: """Process services based on the configuration. Args: @@ -308,11 +327,11 @@ def process_services(host_config_dir: Path, config: dict, state_override: str | pull_before_start: Whether to pull images before starting services quiet: Whether to use quiet mode for docker operations """ - if not config or 'services' not in config: + if not config or "services" not in config: logger.error("Error: Invalid configuration format. 'services' key not found.") return - services = config['services'] + services = config["services"] # Process services using the new structure for category_entry in services: @@ -326,12 +345,12 @@ def process_services(host_config_dir: Path, config: dict, state_override: str | # Process each service in this category for service in service_list: - name = service.get('name', '') + name = service.get("name", "") if not name: logger.warning(f"Skipping invalid service entry in category {category}: missing name") continue - state = (state_override or service.get('state', 'up')).lower() + state = (state_override or service.get("state", "up")).lower() if state not in ALLOWED_STATES: logger.warning(f"Unknown state '{state}' for service {category}/{name}") continue @@ -371,7 +390,7 @@ def cmd_service(args) -> None: sys.exit(1) # Parse service name in format category/subcategory/name - name_parts = args.name.split('/') + name_parts = args.name.split("/") if len(name_parts) < 2: logger.error("Service name must be in format category/name or category/subcategory/name") sys.exit(1) @@ -379,62 +398,59 @@ def cmd_service(args) -> None: # The last part is always the service name service_name = name_parts[-1] # Everything before the last part is the category path - category_path = '/'.join(name_parts[:-1]) + category_path = "/".join(name_parts[:-1]) # Create options with all parameters - options = DockerOptions( - pull_before_start=args.pull_before_start, - quiet=args.quiet - ) + options = DockerOptions(pull_before_start=args.pull_before_start, quiet=args.quiet) # Add log options if they exist in args and operation is 'logs' - if args.operation == 'logs': - if hasattr(args, 'follow'): + if args.operation == "logs": + if hasattr(args, "follow"): options.follow = args.follow - if hasattr(args, 'tail'): + if hasattr(args, "tail"): options.tail = args.tail - if hasattr(args, 'since'): + if hasattr(args, "since"): options.since = args.since - if hasattr(args, 'timestamps'): + if hasattr(args, "timestamps"): options.timestamps = args.timestamps docker_command(get_host_config_dir(), docker_stacks_dir / category_path, service_name, args.operation, options) def main() -> None: - parser = argparse.ArgumentParser(description='Manage Docker services using YAML configuration.') - subparsers = parser.add_subparsers(dest='command', help='Commands', required=True) + parser = argparse.ArgumentParser(description="Manage Docker services using YAML configuration.") + subparsers = parser.add_subparsers(dest="command", help="Commands", required=True) # Config command - config_parser = subparsers.add_parser('config', help='Manage service configurations') - config_subparsers = config_parser.add_subparsers(dest='config_command', help='Config subcommands', required=True) + config_parser = subparsers.add_parser("config", help="Manage service configurations") + config_subparsers = config_parser.add_subparsers(dest="config_command", help="Config subcommands", required=True) # Config apply command - config_apply_parser = config_subparsers.add_parser('apply', help='Apply service configurations') - config_apply_parser.add_argument('--config', '-c', help='Path to the YAML configuration file') - config_apply_parser.add_argument('--mode', '-m', choices=list(ALLOWED_STATES), help='Override state for all services') - config_apply_parser.add_argument('--pull-before-start', action='store_true', default=False, help='Pull images before starting services') - config_apply_parser.add_argument('--quiet', action='store_true', default=False, help='Use quiet mode for docker operations') + config_apply_parser = config_subparsers.add_parser("apply", help="Apply service configurations") + config_apply_parser.add_argument("--config", "-c", help="Path to the YAML configuration file") + config_apply_parser.add_argument("--mode", "-m", choices=list(ALLOWED_STATES), help="Override state for all services") + config_apply_parser.add_argument("--pull-before-start", action="store_true", default=False, help="Pull images before starting services") + config_apply_parser.add_argument("--quiet", action="store_true", default=False, help="Use quiet mode for docker operations") # Service command - service_parser = subparsers.add_parser('service', help='Manage individual services') - service_parser.add_argument('operation', choices=list(ALLOWED_STATES), help='Operation to perform on the service') - service_parser.add_argument('name', help='Service name in format category/name or category/subcategory/name') - service_parser.add_argument('--pull-before-start', action='store_true', default=False, help='Pull images before starting the service') - service_parser.add_argument('--quiet', action='store_true', default=False, help='Use quiet mode for docker operations') + service_parser = subparsers.add_parser("service", help="Manage individual services") + service_parser.add_argument("operation", choices=list(ALLOWED_STATES), help="Operation to perform on the service") + service_parser.add_argument("name", help="Service name in format category/name or category/subcategory/name") + service_parser.add_argument("--pull-before-start", action="store_true", default=False, help="Pull images before starting the service") + service_parser.add_argument("--quiet", action="store_true", default=False, help="Use quiet mode for docker operations") # Log-specific options - service_parser.add_argument('--follow', '-f', action='store_true', help='Follow log output (like tail -f)') - service_parser.add_argument('--tail', '-n', default="all", help='Number of lines to show from the end of logs (default: all)') - service_parser.add_argument('--since', '-s', help='Show logs since timestamp (e.g., "10m" for last 10 minutes)') - service_parser.add_argument('--timestamps', '-t', action='store_true', help='Show timestamps with log entries') + service_parser.add_argument("--follow", "-f", action="store_true", help="Follow log output (like tail -f)") + service_parser.add_argument("--tail", "-n", default="all", help="Number of lines to show from the end of logs (default: all)") + service_parser.add_argument("--since", "-s", help='Show logs since timestamp (e.g., "10m" for last 10 minutes)') + service_parser.add_argument("--timestamps", "-t", action="store_true", help="Show timestamps with log entries") args = parser.parse_args() # Handle command structure - if args.command == 'config': - if args.config_command == 'apply': + if args.command == "config": + if args.config_command == "apply": cmd_config_apply(args) - elif args.command == 'service': + elif args.command == "service": cmd_service(args) else: parser.print_help() @@ -444,6 +460,6 @@ def main() -> None: try: main() except KeyboardInterrupt: - print('Interrupted') + print("Interrupted") # Exit Code 130: Script terminated by Control-C sys.exit(130) diff --git a/scripts/proxy-request-log.py b/scripts/proxy-request-log.py index ada2e189..7e1d62fe 100755 --- a/scripts/proxy-request-log.py +++ b/scripts/proxy-request-log.py @@ -23,13 +23,13 @@ def proxy_request(self): # Parse the URL parsed_url = urlparse(self.path) target_host = parsed_url.hostname - target_port = parsed_url.port or (80 if parsed_url.scheme == 'http' else 443) + target_port = parsed_url.port or (80 if parsed_url.scheme == "http" else 443) print("\nParsed URL:") print(parsed_url) # Read the content length - content_length = int(self.headers.get('Content-Length', 0)) + content_length = int(self.headers.get("Content-Length", 0)) post_data = self.rfile.read(content_length) if content_length > 0 else None # Create headers for the target request @@ -48,7 +48,7 @@ def proxy_request(self): conn = http.client.HTTPConnection(target_host, target_port) # Make the request to the target server - conn.request(self.command, urlunparse(parsed_url._replace(scheme='', netloc='')), body=post_data, headers=headers) + conn.request(self.command, urlunparse(parsed_url._replace(scheme="", netloc="")), body=post_data, headers=headers) target_response = conn.getresponse() # Send the response back to the client diff --git a/scripts/restructure-services.py b/scripts/restructure-services.py index e3ca11ec..4daa9141 100755 --- a/scripts/restructure-services.py +++ b/scripts/restructure-services.py @@ -271,9 +271,7 @@ def _perform_dry_run( f"→ Would move config: {paths.config_dir.relative_to(paths.docker_dir)}/ → {paths.stack_dir.relative_to(paths.docker_dir)}/config/" ) - messages.append( - f"→ Would rename: {paths.stack_dir.relative_to(paths.docker_dir)}/ → {final_dir.relative_to(paths.docker_dir)}/" - ) + messages.append(f"→ Would rename: {paths.stack_dir.relative_to(paths.docker_dir)}/ → {final_dir.relative_to(paths.docker_dir)}/") return True, "\n ".join(messages) diff --git a/scripts/update-example-env.py b/scripts/update-example-env.py index 77618596..4c5a2bc3 100755 --- a/scripts/update-example-env.py +++ b/scripts/update-example-env.py @@ -3,25 +3,25 @@ import sys SENSITIVE_VARS_TO_MASK = [ - 'KEY', - 'USERNAME', - 'PASSWORD', - 'PASSPHRASE', - 'TOKEN', - 'SECRET', - 'SENSITIVE', + "KEY", + "USERNAME", + "PASSWORD", + "PASSPHRASE", + "TOKEN", + "SECRET", + "SENSITIVE", ] SENSITIVE_VARS_TO_GENERALIZE = { - 'TIMEZONE': 'Etc/UTC', - 'MYDOMAIN': 'example.com', - 'LOCATION_CITY': 'Greenwich', - 'LOCATION_LATITUDE': '51.48', - 'LOCATION_LONGITUDE': '0.00', - 'ADMIN_USER': 'admin', - 'ADMIN_EMAIL': 'root@localhost', - 'ADMIN_DISPLAYNAME': 'AdminUser', - 'IP': 'xxx.xxx.xxx.xxx' + "TIMEZONE": "Etc/UTC", + "MYDOMAIN": "example.com", + "LOCATION_CITY": "Greenwich", + "LOCATION_LATITUDE": "51.48", + "LOCATION_LONGITUDE": "0.00", + "ADMIN_USER": "admin", + "ADMIN_EMAIL": "root@localhost", + "ADMIN_DISPLAYNAME": "AdminUser", + "IP": "xxx.xxx.xxx.xxx", } @@ -37,7 +37,7 @@ def get_generalized_value(variable_name: str) -> str | None: """ for key, value in SENSITIVE_VARS_TO_GENERALIZE.items(): # Use custom boundaries: not preceded/followed by alphanumeric (treats _ as boundary) - pattern = rf'(? str: output_lines = [] for line in lines: - if '=' in line: - variable, value = line.strip().split('=', 1) + if "=" in line: + variable, value = line.strip().split("=", 1) generalized_value = get_generalized_value(variable) if generalized_value is not None: output_lines.append(f"{variable}={generalized_value}") elif contains_any_substring(variable, SENSITIVE_VARS_TO_MASK): - output_lines.append(f"{variable}=\"use-some-very-secure-value-here\"") + output_lines.append(f'{variable}="use-some-very-secure-value-here"') else: output_lines.append(line.strip()) else: output_lines.append(line.strip()) - return '\n'.join(output_lines) + '\n' + return "\n".join(output_lines) + "\n" def main() -> None: @@ -74,7 +74,7 @@ def main() -> None: if len(sys.argv) == 3: output_file = sys.argv[2] - with open(output_file, 'w') as f: + with open(output_file, "w") as f: f.write(output) else: print(output) From 058afe9c9ef80dcc5120ee0e78c7dc015f03d731 Mon Sep 17 00:00:00 2001 From: Buba Date: Tue, 27 Jan 2026 18:38:30 +0000 Subject: [PATCH 2/2] Simplify code, add more safeguards --- scripts/github-extract-links.py | 58 ++-- scripts/infra-mcp/tools/get_app_icon.py | 4 +- scripts/infra-mcp/utils/git.py | 20 +- scripts/labctl.py | 356 ++++++++++-------------- scripts/update-example-env.py | 106 ++++--- 5 files changed, 276 insertions(+), 268 deletions(-) diff --git a/scripts/github-extract-links.py b/scripts/github-extract-links.py index 1ec3e2b7..c3af40a4 100755 --- a/scripts/github-extract-links.py +++ b/scripts/github-extract-links.py @@ -1,40 +1,50 @@ #!/usr/bin/env python3 +"""Extract GitHub repository links from files in a directory.""" -import os import re import sys +from pathlib import Path +SCANNABLE_EXTENSIONS: tuple[str, ...] = (".md", ".yml", ".yaml", ".sh") +# Usernames for user accounts on GitHub can only contain alphanumeric characters and dashes ( - ). +GITHUB_REPO_PATTERN = re.compile(r"https://github\.com/([\w.\-_]+/[\w.\-_]+)") -def extract_github_links(directory: str) -> list[str]: + +def extract_github_links(directory: Path) -> list[str]: + """Extract unique GitHub repository links from files in a directory. + + Scans markdown, YAML, shell scripts, and Dockerfiles for GitHub URLs. + """ github_links: set[str] = set() - for root, _dirs, files in os.walk(directory): - for file in files: - if file.endswith(".md") or file.endswith(".yml") or file.endswith(".yaml") or file.endswith(".sh") or file.startswith("Dockerfile"): - file_path = os.path.join(root, file) - with open(file_path) as f: - content = f.read() - # Usernames for user accounts on GitHub can only contain alphanumeric characters and dashes ( - ). - links = re.findall(r"https://github.com/([\w.\-\_]+/[\w.\-\_]+)", content) - links = trim_git_ending(links) - github_links.update(links) + + for file_path in directory.rglob("*"): + if not file_path.is_file(): + continue + + if not (file_path.suffix in SCANNABLE_EXTENSIONS or file_path.name.startswith("Dockerfile")): + continue + + try: + content = file_path.read_text() + except (OSError, UnicodeDecodeError): + continue + + links = GITHUB_REPO_PATTERN.findall(content) + github_links.update(trim_git_suffix(link) for link in links) + return list(github_links) -def trim_git_ending(links: list[str]) -> list[str]: - trimmed_links: list[str] = [] - for link in links: - if link.endswith(".git"): - trimmed_links.append(link[:-4]) - else: - trimmed_links.append(link) - return trimmed_links +def trim_git_suffix(link: str) -> str: + """Remove .git suffix from a repository path if present.""" + return link.removesuffix(".git") -def main(): - directory = sys.argv[1] if len(sys.argv) > 1 else os.getcwd() +def main() -> None: + """Extract and print GitHub links from the specified directory.""" + directory = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd() - links = extract_github_links(directory) - for link in links: + for link in extract_github_links(directory): print(f"https://github.com/{link}") diff --git a/scripts/infra-mcp/tools/get_app_icon.py b/scripts/infra-mcp/tools/get_app_icon.py index c4ebd461..b6ef994d 100755 --- a/scripts/infra-mcp/tools/get_app_icon.py +++ b/scripts/infra-mcp/tools/get_app_icon.py @@ -117,8 +117,8 @@ def get_priority(link): # 2. Check for the default location default_favicon = urljoin(homepage_url, "/favicon.ico") - favicon_response = requests.head(default_favicon, headers=self.headers, timeout=5) - if favicon_response.status_code == 200: + favicon_response = requests.head(default_favicon, headers=self.headers, timeout=5, allow_redirects=True) + if favicon_response.ok: return default_favicon else: return None diff --git a/scripts/infra-mcp/utils/git.py b/scripts/infra-mcp/utils/git.py index 073487e5..f94872c9 100644 --- a/scripts/infra-mcp/utils/git.py +++ b/scripts/infra-mcp/utils/git.py @@ -4,13 +4,20 @@ import shutil import subprocess +from pathlib import Path -def get_git_root() -> str: +def get_git_root(reference_path: Path | str | None = None) -> Path: """Get the git repository root directory. + Args: + reference_path: Optional path (file or directory) to resolve git root from. + If a directory, the git command runs from that directory. + If a file, the git command runs from the file's parent directory. + If None, uses CWD. + Returns: - str: The absolute path to the git repository root directory. + Path: The absolute path to the git repository root directory. Raises: RuntimeError: If git executable is not found or not in a git repository. @@ -18,15 +25,22 @@ def get_git_root() -> str: git_cmd = shutil.which("git") if git_cmd is None: raise RuntimeError("Git not found on PATH") from None + + cwd = None + if reference_path: + resolved = Path(reference_path).resolve() + cwd = resolved if resolved.is_dir() else resolved.parent + try: result = subprocess.run( # noqa: S603 [git_cmd, "rev-parse", "--show-toplevel"], stdout=subprocess.PIPE, check=True, text=True, + cwd=cwd, ) except FileNotFoundError: raise RuntimeError("Git executable not found. Please install Git and ensure it is on your PATH.") from None except subprocess.CalledProcessError: raise RuntimeError("Unable to locate git repository. Are you running this inside a Git repo?") from None - return result.stdout.strip() + return Path(result.stdout.strip()) diff --git a/scripts/labctl.py b/scripts/labctl.py index dd2700e1..107e9100 100755 --- a/scripts/labctl.py +++ b/scripts/labctl.py @@ -1,8 +1,5 @@ #!/usr/bin/env python3 -""" -Docker services management script. -Uses YAML configuration to manage Docker services. -""" +"""Docker services management script using YAML configuration.""" import argparse import logging @@ -16,10 +13,12 @@ import yaml -# Configure logging logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger(__name__) +DOCKER_STACKS_DIR: Path = (Path(__file__).resolve().parent.parent / "docker").resolve() +ALLOWED_OPERATIONS: tuple[str, ...] = ("pull", "up", "down", "restart", "recreate", "config", "logs") + @dataclass class DockerOptions: @@ -34,17 +33,8 @@ class DockerOptions: timestamps: bool = False -# Global variables -docker_stacks_dir: Path = (Path(__file__).resolve().parent.parent / "docker").resolve() -ALLOWED_STATES: tuple[str, ...] = ("pull", "up", "down", "restart", "recreate", "config", "logs") - - def create_network_if_missing(network_name: str) -> None: - """Create Docker network if it doesn't exist. - - Args: - network_name: The name of the Docker network to check/create - """ + """Create Docker network if it doesn't exist.""" try: docker(["network", "inspect", network_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except subprocess.CalledProcessError: @@ -53,128 +43,115 @@ def create_network_if_missing(network_name: str) -> None: def get_external_networks(compose_file: Path) -> list[str]: - """Extract external networks from a Docker Compose file.""" - networks: list[str] = [] + """Extract external networks from a Docker Compose file. + + Supports formats: external: true | external: {name: "..."} | name: "..." + """ try: with open(compose_file) as f: yaml_content = yaml.safe_load(f) or {} - networks_def = yaml_content.get("networks") or {} - if isinstance(networks_def, dict): - for key, cfg in networks_def.items(): - if not isinstance(cfg, dict): - continue - ext = cfg.get("external", False) - # Support: external: true | external: {name: "..."} | name: "..." - name_override = cfg.get("name") - if ext is True: - networks.append(name_override or key) - elif isinstance(ext, dict): - networks.append(ext.get("name") or name_override or key) except (FileNotFoundError, yaml.YAMLError, OSError) as e: logger.warning(f"Error extracting networks from {compose_file}: {e}") + return [] + + networks_def = yaml_content.get("networks") or {} + if not isinstance(networks_def, dict): + return [] + + networks: list[str] = [] + for key, cfg in networks_def.items(): + if not isinstance(cfg, dict): + continue + ext = cfg.get("external", False) + name_override = cfg.get("name") + if ext is True: + networks.append(name_override or key) + elif isinstance(ext, dict): + networks.append(ext.get("name") or name_override or key) return list(dict.fromkeys(networks)) def create_service_networks(compose_file: Path) -> None: - """Create all external networks required by a service. - - Args: - compose_file: Path to the Docker Compose file - """ - networks = get_external_networks(compose_file) - for network_name in networks: + """Create all external networks required by a service.""" + for network_name in get_external_networks(compose_file): create_network_if_missing(network_name) def create_localhost_link(docker_config_dir: Path) -> None: - """Create 'localhost' symlink in the parent directory. - - Creates a symbolic link named 'localhost' that points to the current hostname directory. - - Args: - docker_config_dir: The Docker configuration directory containing hostname subdirectories - """ - hostname = socket.gethostname() + """Create 'localhost' symlink pointing to the current hostname directory.""" + hostname = socket.gethostname().lower() localhost_link = docker_config_dir / "localhost" hostname_dir = docker_config_dir / hostname - if hostname_dir.exists() and hostname_dir.is_dir(): - # Create or update the localhost symlink - if localhost_link.exists(): - if localhost_link.is_symlink(): - localhost_link.unlink() - else: - logger.error(f"Error: {localhost_link} exists but is not a symlink. Cannot create link.") - return + if not (hostname_dir.exists() and hostname_dir.is_dir()): + return - try: - os.symlink(f"{hostname}/", localhost_link, target_is_directory=True) - except Exception: - logger.exception("Error creating localhost symlink") + if localhost_link.exists() and not localhost_link.is_symlink(): + logger.error(f"Error: {localhost_link} exists but is not a symlink. Cannot create link.") + return + if localhost_link.is_symlink(): + localhost_link.unlink() -def get_compose_file(stack_dir: Path, service_name: str) -> Path: - """Get the yaml file path for a service. + try: + os.symlink(f"{hostname}/", localhost_link, target_is_directory=True) + except OSError: + logger.exception("Error creating localhost symlink") - Args: - stack_dir: Directory containing service definitions - service_name: Name of the service - Returns: - Path: The path to the service's compose file - """ +def get_compose_file(stack_dir: Path, service_name: str) -> Path: + """Get the yaml file path for a service.""" return stack_dir / service_name / f"{service_name}.yaml" def has_build_directive(compose_file: Path) -> bool: - """Check if the service uses a build directive. - - Args: - compose_file: Path to the Docker Compose file - - Returns: - bool: True if the compose file contains any build directives - """ + """Check if the compose file contains any build directives.""" with open(compose_file) as f: yaml_content = yaml.safe_load(f) - if yaml_content and "services" in yaml_content: - for service_config in yaml_content["services"].values(): - if "build" in service_config: - return True - return False + if not yaml_content: + return False + services = yaml_content.get("services") + if not isinstance(services, dict): + return False + return any("build" in svc for svc in services.values()) def get_env_file_args(host_config_dir: Path, service_name: str) -> list[str]: - """Get environment file arguments for Docker Compose with normalized paths.""" + """Get environment file arguments for Docker Compose. + + Searches for .env files in order of precedence: + 1. Common .env in config/docker + 2. Host-specific .env in config/docker/ + 3. Common service-specific .env. in config/docker + 4. Host- and service-specific .env. in config/docker/ + """ env_paths = [ - host_config_dir.parent / ".env", # Common .env file in config/docker - host_config_dir / ".env", # Host-specific .env file in config/docker/ - host_config_dir.parent / f".env.{service_name}", # Common service-specific .env file in config/docker - host_config_dir / f".env.{service_name}", # Host- and service-specific .env file in config/docker/ + host_config_dir.parent / ".env", + host_config_dir / ".env", + host_config_dir.parent / f".env.{service_name}", + host_config_dir / f".env.{service_name}", ] - args = [] + args: list[str] = [] for file in env_paths: - absolute_path = file.resolve() - if absolute_path.is_file(): - args.extend(["--env-file", str(absolute_path)]) + resolved = file.resolve() + if resolved.is_file(): + args.extend(["--env-file", str(resolved)]) return args -def docker(cmd: list[str], env=None, stdin=None, stdout=None, stderr=None) -> None: - """Execute a docker command with the given arguments. - - Args: - cmd: List of command arguments to pass to docker - env: Environment variables for the subprocess - stdin: Standard input for the subprocess - stdout: Standard output for the subprocess - stderr: Standard error for the subprocess - """ +def docker( + cmd: list[str], + env: dict[str, str] | None = None, + stdin: int | None = None, + stdout: int | None = None, + stderr: int | None = None, +) -> None: + """Execute a docker command with the given arguments.""" docker_bin = shutil.which("docker") if docker_bin is None: - raise RuntimeError("Docker executable not found on PATH.") from None + raise RuntimeError("Docker executable not found on PATH.") subprocess.run([docker_bin, *cmd], env=env, stdin=stdin, stdout=stdout, stderr=stderr, check=True) # noqa: S603 @@ -185,41 +162,27 @@ def docker_pull( env_file_args: list[str], quiet: bool = False, ) -> None: - """Pull Docker images for a service. - - Args: - stack_dir: Directory containing the service definition - service_name: Name of the service - compose_file: Path to the Docker Compose file - env_file_args: List of environment file arguments - quiet: Whether to use quiet mode (default: False) - """ + """Pull Docker images for a service, using build if the service has a build directive.""" logger.info(f">>> Pulling {stack_dir}/{service_name}") + if has_build_directive(compose_file): # Bake: https://docs.docker.com/guides/compose-bake/ env = os.environ.copy() env["COMPOSE_BAKE"] = "true" - build_cmd = ["compose", "-f", compose_file, *env_file_args, "build", "--pull"] + cmd = ["compose", "-f", compose_file, *env_file_args, "build", "--pull"] if quiet: - build_cmd.append("--quiet") - docker(build_cmd, env=env) + cmd.append("--quiet") + docker(cmd, env=env) else: - pull_cmd = ["compose", "-f", compose_file, *env_file_args, "pull"] + cmd = ["compose", "-f", compose_file, *env_file_args, "pull"] if quiet: - pull_cmd.append("--quiet") - docker(pull_cmd) + cmd.append("--quiet") + docker(cmd) def build_log_command_flags(options: DockerOptions) -> list[str]: - """Build Docker Compose log command flags based on options. - - Args: - options: Docker operation options containing log-specific settings - - Returns: - list[str]: List of command flags for docker compose logs - """ - flags = [] + """Build Docker Compose log command flags based on options.""" + flags: list[str] = [] if options.follow: flags.append("--follow") if options.tail and options.tail != "all": @@ -236,21 +199,13 @@ def docker_command( stack_dir: Path, service_name: str, action: str, - options: DockerOptions = None, + options: DockerOptions | None = None, ) -> None: - """Execute Docker Compose command for a service. - - Args: - host_config_dir: Path to the host-specific configuration directory - stack_dir: Path to the service category directory - service_name: Name of the service to operate on - action: The action to perform (pull, up, down, restart, recreate, config, logs) - options: Docker operation options (default: None) - """ + """Execute Docker Compose command for a service.""" if options is None: options = DockerOptions() - logger.info("") # separation + logger.info("") compose_file = get_compose_file(stack_dir, service_name) if not compose_file.exists(): logger.error(f"Compose file not found: {compose_file}") @@ -261,8 +216,8 @@ def docker_command( create_service_networks(compose_file) env_file_args = get_env_file_args(host_config_dir, service_name) + base_cmd = ["compose", "-f", compose_file, *env_file_args] - # Handle other operations match action: case "pull": docker_pull(stack_dir, service_name, compose_file, env_file_args, options.quiet) @@ -270,43 +225,38 @@ def docker_command( case "up": if options.pull_before_start: docker_pull(stack_dir, service_name, compose_file, env_file_args, options.quiet) - logger.info(f">>> Starting {stack_dir}/{service_name}") - docker(["compose", "-f", compose_file, *env_file_args, "up", "--detach"]) + docker([*base_cmd, "up", "--detach"]) case "down": logger.info(f">>> Stopping {stack_dir}/{service_name}") - docker(["compose", "-f", compose_file, *env_file_args, "down"]) + docker([*base_cmd, "down"]) case "restart": logger.info(f">>> Restarting {stack_dir}/{service_name}") - docker(["compose", "-f", compose_file, *env_file_args, "restart"]) + docker([*base_cmd, "restart"]) case "recreate": if options.pull_before_start: docker_pull(stack_dir, service_name, compose_file, env_file_args, options.quiet) - logger.info(f">>> Recreating {stack_dir}/{service_name}") - docker(["compose", "-f", compose_file, *env_file_args, "up", "--detach", "--force-recreate"]) + docker([*base_cmd, "up", "--detach", "--force-recreate"]) case "config": logger.info(f">>> Checking {stack_dir}/{service_name}") - docker(["compose", "-f", compose_file, *env_file_args, "config"]) + docker([*base_cmd, "config"]) case "logs": logger.info(f">>> Showing logs for {stack_dir}/{service_name}") - log_cmd = ["compose", "-f", compose_file, *env_file_args, "logs"] - log_cmd.extend(build_log_command_flags(options)) - docker(log_cmd) + docker([*base_cmd, "logs", *build_log_command_flags(options)]) -def load_services_config(config_file: str) -> dict: +def load_services_config(config_file: str | Path) -> dict: """Load services configuration from YAML file.""" try: - with open(config_file) as file: - config = yaml.safe_load(file) - return config - except Exception: + with open(config_file) as f: + return yaml.safe_load(f) + except (OSError, yaml.YAMLError): logger.exception(f"Error loading configuration file {config_file}") sys.exit(1) @@ -318,58 +268,65 @@ def process_services( pull_before_start: bool = False, quiet: bool = False, ) -> None: - """Process services based on the configuration. - - Args: - host_config_dir: Path to the host-specific configuration directory - config: Dictionary containing service configurations - state_override: State to override for all services (pull, up, down, etc.) - pull_before_start: Whether to pull images before starting services - quiet: Whether to use quiet mode for docker operations - """ - if not config or "services" not in config: + """Process services based on the configuration.""" + if not isinstance(config, dict): + logger.error("Error: Invalid configuration format. Config must be a dict.") + return + + if "services" not in config: logger.error("Error: Invalid configuration format. 'services' key not found.") return - services = config["services"] + if not isinstance(config["services"], list): + logger.error("Error: Invalid configuration format. 'services' must be a list.") + return + + options = DockerOptions(pull_before_start=pull_before_start, quiet=quiet) - # Process services using the new structure - for category_entry in services: + for category_entry in config["services"]: # Each entry should have a single key (the category name) and a list of services - if not category_entry or len(category_entry) != 1: + if not isinstance(category_entry, dict) or len(category_entry) != 1: logger.warning(f"Skipping invalid category entry: {category_entry}") continue - category = list(category_entry.keys())[0] + category = next(iter(category_entry.keys())) service_list = category_entry[category] + if not isinstance(service_list, list): + logger.warning(f"Skipping category '{category}': services value is not a list") + continue + # Process each service in this category for service in service_list: + if not isinstance(service, dict): + logger.warning(f"Skipping invalid service entry in category {category}: expected dict, got {type(service).__name__}") + continue + name = service.get("name", "") if not name: logger.warning(f"Skipping invalid service entry in category {category}: missing name") continue state = (state_override or service.get("state", "up")).lower() - if state not in ALLOWED_STATES: + if state not in ALLOWED_OPERATIONS: logger.warning(f"Unknown state '{state}' for service {category}/{name}") continue - docker_command(host_config_dir, docker_stacks_dir / category, name, state, DockerOptions(pull_before_start, quiet)) + docker_command(host_config_dir, DOCKER_STACKS_DIR / category, name, state, options) def get_host_config_dir() -> Path: + """Get the host-specific Docker configuration directory.""" hostname = socket.gethostname().lower() script_dir = Path(__file__).parent.absolute() - docker_config_dir = script_dir.parent / "config" / "docker" - return docker_config_dir / hostname + return script_dir.parent / "config" / "docker" / hostname -def cmd_config_apply(args) -> None: +def cmd_config_apply(args: argparse.Namespace) -> None: """Apply configuration to Docker services.""" if args.config: - config_file = args.config - host_config_dir = Path(config_file).parent + config_file = Path(args.config) + host_config_dir = config_file.parent else: # Use the default location based on hostname host_config_dir = get_host_config_dir() @@ -378,66 +335,56 @@ def cmd_config_apply(args) -> None: logger.info("Init...") config = load_services_config(config_file) create_localhost_link(host_config_dir.parent) - - # Process services with optional mode override process_services(host_config_dir, config, args.mode, args.pull_before_start, args.quiet) -def cmd_service(args) -> None: +def cmd_service(args: argparse.Namespace) -> None: """Manage individual Docker services.""" if not args.name: logger.error("Service name is required") sys.exit(1) - # Parse service name in format category/subcategory/name name_parts = args.name.split("/") if len(name_parts) < 2: logger.error("Service name must be in format category/name or category/subcategory/name") sys.exit(1) - # The last part is always the service name service_name = name_parts[-1] - # Everything before the last part is the category path category_path = "/".join(name_parts[:-1]) - # Create options with all parameters - options = DockerOptions(pull_before_start=args.pull_before_start, quiet=args.quiet) - - # Add log options if they exist in args and operation is 'logs' - if args.operation == "logs": - if hasattr(args, "follow"): - options.follow = args.follow - if hasattr(args, "tail"): - options.tail = args.tail - if hasattr(args, "since"): - options.since = args.since - if hasattr(args, "timestamps"): - options.timestamps = args.timestamps + options = DockerOptions( + pull_before_start=args.pull_before_start, + quiet=args.quiet, + follow=getattr(args, "follow", False), + tail=getattr(args, "tail", "all"), + since=getattr(args, "since", None), + timestamps=getattr(args, "timestamps", False), + ) - docker_command(get_host_config_dir(), docker_stacks_dir / category_path, service_name, args.operation, options) + docker_command(get_host_config_dir(), DOCKER_STACKS_DIR / category_path, service_name, args.operation, options) def main() -> None: + """Main entry point for the Docker services management CLI.""" parser = argparse.ArgumentParser(description="Manage Docker services using YAML configuration.") subparsers = parser.add_subparsers(dest="command", help="Commands", required=True) - # Config command + # Config command with apply subcommand config_parser = subparsers.add_parser("config", help="Manage service configurations") config_subparsers = config_parser.add_subparsers(dest="config_command", help="Config subcommands", required=True) - # Config apply command config_apply_parser = config_subparsers.add_parser("apply", help="Apply service configurations") config_apply_parser.add_argument("--config", "-c", help="Path to the YAML configuration file") - config_apply_parser.add_argument("--mode", "-m", choices=list(ALLOWED_STATES), help="Override state for all services") - config_apply_parser.add_argument("--pull-before-start", action="store_true", default=False, help="Pull images before starting services") - config_apply_parser.add_argument("--quiet", action="store_true", default=False, help="Use quiet mode for docker operations") + config_apply_parser.add_argument("--mode", "-m", choices=list(ALLOWED_OPERATIONS), help="Override state for all services") + config_apply_parser.add_argument("--pull-before-start", action="store_true", help="Pull images before starting services") + config_apply_parser.add_argument("--quiet", action="store_true", help="Use quiet mode for docker operations") # Service command service_parser = subparsers.add_parser("service", help="Manage individual services") - service_parser.add_argument("operation", choices=list(ALLOWED_STATES), help="Operation to perform on the service") + service_parser.add_argument("operation", choices=list(ALLOWED_OPERATIONS), help="Operation to perform on the service") service_parser.add_argument("name", help="Service name in format category/name or category/subcategory/name") - service_parser.add_argument("--pull-before-start", action="store_true", default=False, help="Pull images before starting the service") - service_parser.add_argument("--quiet", action="store_true", default=False, help="Use quiet mode for docker operations") + service_parser.add_argument("--pull-before-start", action="store_true", help="Pull images before starting the service") + service_parser.add_argument("--quiet", action="store_true", help="Use quiet mode for docker operations") # Log-specific options service_parser.add_argument("--follow", "-f", action="store_true", help="Follow log output (like tail -f)") service_parser.add_argument("--tail", "-n", default="all", help="Number of lines to show from the end of logs (default: all)") @@ -446,14 +393,13 @@ def main() -> None: args = parser.parse_args() - # Handle command structure - if args.command == "config": - if args.config_command == "apply": + match args.command: + case "config" if args.config_command == "apply": cmd_config_apply(args) - elif args.command == "service": - cmd_service(args) - else: - parser.print_help() + case "service": + cmd_service(args) + case _: + parser.print_help() if __name__ == "__main__": diff --git a/scripts/update-example-env.py b/scripts/update-example-env.py index 4c5a2bc3..ae6e53a9 100755 --- a/scripts/update-example-env.py +++ b/scripts/update-example-env.py @@ -1,8 +1,20 @@ #!/usr/bin/env python3 +"""Mask sensitive variables in environment files for safe sharing. + +This script reads an environment file and replaces sensitive values with +placeholder values, making it safe to share as an example configuration. +""" + +import argparse +import logging import re import sys +from pathlib import Path + +logging.basicConfig(level=logging.INFO, format="%(message)s") +logger = logging.getLogger(__name__) -SENSITIVE_VARS_TO_MASK = [ +SENSITIVE_VARS_TO_MASK: tuple[str, ...] = ( "KEY", "USERNAME", "PASSWORD", @@ -10,9 +22,9 @@ "TOKEN", "SECRET", "SENSITIVE", -] +) -SENSITIVE_VARS_TO_GENERALIZE = { +SENSITIVE_VARS_TO_GENERALIZE: dict[str, str] = { "TIMEZONE": "Etc/UTC", "MYDOMAIN": "example.com", "LOCATION_CITY": "Greenwich", @@ -24,13 +36,16 @@ "IP": "xxx.xxx.xxx.xxx", } +MASKED_VALUE: str = '"use-some-very-secure-value-here"' -def contains_any_substring(string: str, substrings: list[str]) -> bool: + +def contains_any_substring(string: str, substrings: tuple[str, ...]) -> bool: + """Check if the string contains any of the given substrings.""" return any(substring in string for substring in substrings) def get_generalized_value(variable_name: str) -> str | None: - """Check if any generalization key appears as a discrete word in the variable name. + """Find a generalized value for a variable if it matches a known pattern. Treats underscore as a word boundary, so SERVER_IP will match IP. Returns the generalized value if a match is found, None otherwise. @@ -43,42 +58,65 @@ def get_generalized_value(variable_name: str) -> str | None: return None -def mask_sensitive_variables(input_file: str) -> str: - with open(input_file) as f: - lines = f.readlines() - - output_lines = [] - for line in lines: - if "=" in line: - variable, value = line.strip().split("=", 1) - generalized_value = get_generalized_value(variable) - if generalized_value is not None: - output_lines.append(f"{variable}={generalized_value}") - elif contains_any_substring(variable, SENSITIVE_VARS_TO_MASK): - output_lines.append(f'{variable}="use-some-very-secure-value-here"') - else: - output_lines.append(line.strip()) - else: - output_lines.append(line.strip()) +def mask_line(line: str) -> str: + """Mask sensitive values in a single environment file line. + + Returns the processed line with sensitive values replaced. + """ + if "=" not in line: + return line.strip() + + variable, _ = line.strip().split("=", 1) + variable = variable.strip() + normalized_variable = variable.upper() + generalized_value = get_generalized_value(normalized_variable) + + if generalized_value is not None: + return f"{variable}={generalized_value}" + + if contains_any_substring(normalized_variable, SENSITIVE_VARS_TO_MASK): + return f"{variable}={MASKED_VALUE}" + return line.strip() + + +def mask_sensitive_variables(input_file: Path) -> str: + """Read an environment file and mask all sensitive variables. + + Args: + input_file: Path to the input environment file. + + Returns: + The processed file content with sensitive values masked. + """ + lines = input_file.read_text().splitlines() + output_lines = [mask_line(line) for line in lines] return "\n".join(output_lines) + "\n" -def main() -> None: - if len(sys.argv) < 2: - print("Usage: ./update-example-env.py [output_env_file]") - return +def main() -> int: + """Parse arguments and process the environment file.""" + parser = argparse.ArgumentParser( + description="Mask sensitive variables in environment files for safe sharing.", + ) + parser.add_argument("input_file", type=Path, help="Input environment file to process") + parser.add_argument("output_file", type=Path, nargs="?", help="Output file (prints to stdout if not specified)") + + args = parser.parse_args() - input_file = sys.argv[1] - output = mask_sensitive_variables(input_file) + if not args.input_file.is_file(): + logger.error(f"Input file not found: {args.input_file}") + return 1 - if len(sys.argv) == 3: - output_file = sys.argv[2] - with open(output_file, "w") as f: - f.write(output) + output = mask_sensitive_variables(args.input_file) + + if args.output_file: + args.output_file.write_text(output) else: - print(output) + print(output, end="") + + return 0 if __name__ == "__main__": - main() + sys.exit(main())