diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8625a50..c90732d 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -10,6 +10,15 @@ on: pull_request: branches: [ master ] workflow_dispatch: # Allow manual trigger + inputs: + push: + description: 'Push image to registry' + required: false + default: 'false' + type: choice + options: + - 'false' + - 'true' env: # github.repository as / @@ -138,7 +147,7 @@ jobs: uses: docker/setup-buildx-action@v3 - name: Log into registry - if: github.event_name != 'pull_request' || github.event.inputs.push == 'true' + if: github.event_name == 'push' || github.event.inputs.push == 'true' uses: docker/login-action@v3 with: registry: ${{ secrets.DOCKER_REGISTRY }} @@ -205,7 +214,7 @@ jobs: build-args: | RELEASE_VERSION_ARG="${{ steps.tags.outputs.result }}" platforms: linux/amd64,linux/arm64 - push: ${{ github.event_name != 'pull_request' || github.event.inputs.push == 'true' }} + push: ${{ github.event_name == 'push' || github.event.inputs.push == 'true' }} tags: ${{ steps.normalized.outputs.result }} labels: ${{ steps.meta.outputs.labels }} @@ -214,7 +223,7 @@ jobs: # - name: Docker Hub Description - # if: github.event_name != 'pull_request' || github.event.inputs.push == 'true' + # if: github.event_name == 'push' || github.event.inputs.push == 'true' # run: | # wget -q https://github.com/christian-korneck/docker-pushrm/releases/download/v1.8.0/docker-pushrm_linux_amd64 -O $HOME/.docker/cli-plugins/docker-pushrm # chmod +x $HOME/.docker/cli-plugins/docker-pushrm diff --git a/src/easymapping/__init__.py b/src/easymapping/__init__.py index c6eb561..a636dbd 100644 --- a/src/easymapping/__init__.py +++ b/src/easymapping/__init__.py @@ -1,358 +1,4 @@ -import base64 -import json -import os -import re +from .config_generator import HaproxyConfigGenerator +from .label_handler import DockerLabelHandler -from jinja2 import Environment, FileSystemLoader - -from functions import Functions, logger_easyhaproxy - - -class DockerLabelHandler: - def __init__(self, label): - self.__data = None - self.__label_base = label - - def get_lookup_label(self): - return self.__label_base - - def create(self, key): - if isinstance(key, str): - return f"{self.__label_base}.{key}" - - return "{}.{}".format(self.__label_base, ".".join(key)) - - def get(self, label, default_value=""): - if self.has_label(label): - return self.__data[label] - return default_value - - def get_bool(self, label, default_value=False): - if self.has_label(label): - return self.__data[label].lower() in ["true", "1", "yes"] - return default_value - - def get_json(self, label, default_value=None): - if default_value is None: - default_value = {} - if self.has_label(label): - value = self.__data[label] - if not value: # Handle empty strings - return default_value - try: - return json.loads(value) - except json.JSONDecodeError as e: - logger_easyhaproxy.error( - f"Invalid JSON in label '{label}': {value}. Error: {e}. Using default value." - ) - return default_value - return default_value - - def set_data(self, data): - self.__data = data - - def has_label(self, label): - if label in self.__data: - return True - return False - - -class HaproxyConfigGenerator: - def __init__(self, mapping): - self.mapping = mapping - self.mapping.setdefault("ssl_mode", 'default') - self.mapping.setdefault("certbot", {"email": "", "server": False, "eab_kid": False, "eab_hmac_key": False}) - self.mapping["ssl_mode"] = self.mapping["ssl_mode"].lower() - self.label = DockerLabelHandler(mapping['lookup_label'] if 'lookup_label' in mapping else "easyhaproxy") - self.certbot_hosts = [] - self.serving_hosts = [] - self.certs = {} - self.defaults_plugin_configs = [] - - # Initialize plugin system - try: - from plugins import PluginManager - self.plugin_manager = PluginManager( - abort_on_error=self.mapping.get("plugins", {}).get("abort_on_error", False) - ) - self.plugin_manager.load_plugins() - self.plugin_manager.configure_plugins(self.mapping.get("plugins", {})) - self.plugin_manager.initialize_plugins() - self.global_plugin_configs = [] - except Exception as e: - # If plugin system fails to initialize, log but continue - logger_easyhaproxy.warning(f"Failed to initialize plugin system: {e}") - self.plugin_manager = None - self.global_plugin_configs = [] - - def generate(self, container_metadata={}): - self.mapping.setdefault("easymapping", []) - - if container_metadata != {}: - self.mapping["easymapping"] = self.parse(container_metadata) - - # Execute global plugins - if self.plugin_manager: - try: - from plugins import PluginContext - global_context = PluginContext( - parsed_object=container_metadata, - easymapping=self.mapping.get("easymapping", []), - container_env=self.mapping, - domain=None, - port=None, - host_config=None - ) - - # Get enabled plugins from config - enabled_list = self.mapping.get("plugins", {}).get("enabled", []) - # If enabled list contains only empty string, treat as no plugins enabled - if enabled_list and len(enabled_list) > 0 and enabled_list[0] == "": - enabled_list = [] - - global_results = self.plugin_manager.execute_global_plugins(global_context, enabled_list) - - # Extract all plugin configs in a single loop - for result in global_results: - # HAProxy config snippets - if result.haproxy_config: - self.global_plugin_configs.append(result.haproxy_config) - - # Global-level configs (e.g., fcgi-app definitions) - for config in result.global_configs: - if config and config not in self.global_plugin_configs: - self.global_plugin_configs.append(config) - - # Defaults-level configs (e.g., log-format) - for config in result.defaults_configs: - if config and config not in self.defaults_plugin_configs: - self.defaults_plugin_configs.append(config) - except Exception as e: - logger_easyhaproxy.warning(f"Failed to execute global plugins: {e}") - - templates_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'templates') - file_loader = FileSystemLoader(templates_dir) - env = Environment(loader=file_loader) - env.trim_blocks = True - env.lstrip_blocks = True - env.rstrip_blocks = True - template = env.get_template('haproxy.cfg.j2') - return template.render( - data=self.mapping, - global_plugin_configs=self.global_plugin_configs, - defaults_plugin_configs=self.defaults_plugin_configs - ) - - def parse(self, container_metadata): - easymapping = dict() - - for container in container_metadata: - d = container_metadata[container] - - # Extract the definitions dynamically - definitions = {} - r = re.compile(self.label.get_lookup_label() + r"\.(.*)\..*") - for key in d.keys(): - if r.match(key): - definitions[r.search(key).group(1)] = 1 - - if len(definitions.keys()) == 0: - continue - - self.label.set_data(d) - - # Parse each definition found. - for definition in sorted(definitions.keys()): - mode = self.label.get( - self.label.create([definition, "mode"]), - "http" - ) - - # TODO: we can ignore "host" in TCP, but it would break the template - host_label = self.label.create([definition, "host"]) - if not self.label.has_label(host_label): - continue - - port = self.label.get( - self.label.create([definition, "port"]), - "80" - ) - - certbot = self.label.get_bool( - self.label.create([definition, "certbot"]), - False - ) and self.mapping["certbot"]["email"] != "" - clone_to_ssl = self.label.get_bool( - self.label.create([definition, "clone_to_ssl"]) - ) - - # Check if this is a redirect-only entry (no backend) - redirect_only = self.label.get_bool( - self.label.create([definition, "redirect_only"]), - False - ) - - if port not in easymapping: - easymapping[port] = { - "mode": mode, - "ssl-check": "", - "port": port, - "hosts": dict(), - "redirect": dict(), - } - - if redirect_only: - easymapping[port]["redirect"].update(self.label.get_json( - self.label.create([definition, "redirect"]) - )) - continue - - # TODO: this could use `EXPOSE` from `Dockerfile`? - ct_port = self.label.get( - self.label.create([definition, "localport"]), - "80" - ) - - easymapping[port]["ssl-check"] = self.label.get( - self.label.create([definition, "ssl-check"]), - "" - ) - - # Protocol for backend server communication (e.g., fcgi, h2) - proto = self.label.get( - self.label.create([definition, "proto"]), - "" - ) - - # Unix socket path (alternative to host:port) - socket_path = self.label.get( - self.label.create([definition, "socket"]), - "" - ) - - for hostname in sorted(d[host_label].split(",")): - hostname = hostname.strip() - self.serving_hosts.append(f"{hostname}:{port}") - easymapping[port]["hosts"].setdefault(hostname, {}) - easymapping[port]["hosts"][hostname].setdefault("containers", []) - easymapping[port]["hosts"][hostname].setdefault("certbot", False) - easymapping[port]["hosts"][hostname].setdefault("proto", proto) - - # Determine server address: Unix socket or TCP host:port - if socket_path: - server_address = socket_path - else: - server_address = f"{container}:{ct_port}" - - easymapping[port]["hosts"][hostname]["containers"] += [server_address] - easymapping[port]["hosts"][hostname]["certbot"] = certbot - easymapping[port]["hosts"][hostname]["redirect_ssl"] = self.label.get_bool( - self.label.create([definition, "redirect_ssl"]) - ) - easymapping[port]["hosts"][hostname]["balance"] = self.label.get( - self.label.create([definition, "balance"]), - "roundrobin" - ) - - easymapping[port]["redirect"] = self.label.get_json( - self.label.create([definition, "redirect"]) - ) - - # Execute domain plugins for this host - if self.plugin_manager: - try: - from plugins import PluginContext - - domain_context = PluginContext( - parsed_object=container_metadata, - easymapping=easymapping, - container_env=self.mapping, - domain=hostname, - port=port, - host_config=easymapping[port]["hosts"][hostname] - ) - - # Check if plugins are enabled for this domain (from labels) - enabled_plugins = [] - if self.label.has_label(self.label.create([definition, "plugins"])): - enabled_plugins = self.label.get( - self.label.create([definition, "plugins"]), - "" - ).split(",") - enabled_plugins = [p.strip() for p in enabled_plugins if p.strip()] - - # Extract plugin configurations from labels - # Format: easyhaproxy.http.plugin.PLUGIN_NAME.CONFIG_KEY - plugin_configs = {} - for plugin_name in enabled_plugins: - plugin_configs[plugin_name] = {} - # Look for all labels matching easyhaproxy.{definition}.plugin.{plugin_name}.* - plugin_label_prefix = self.label.create([definition, "plugin", plugin_name]) - for label_key in d.keys(): - if label_key.startswith(plugin_label_prefix + "."): - # Extract config key (everything after plugin_label_prefix + ".") - config_key = label_key[len(plugin_label_prefix) + 1:] - plugin_configs[plugin_name][config_key] = d[label_key] - - # Configure plugins with label-specific configs before execution - for plugin_name, config in plugin_configs.items(): - if plugin_name in self.plugin_manager.plugins: - self.plugin_manager.plugins[plugin_name].configure(config) - - domain_results = self.plugin_manager.execute_domain_plugins( - domain_context, - enabled_list=enabled_plugins - ) - - # Extract all plugin configs in a single loop - plugin_configs_for_host = [] - for result in domain_results: - # HAProxy config snippets for this domain - if result.haproxy_config: - plugin_configs_for_host.append(result.haproxy_config) - - # Global-level configs (e.g., fcgi-app definitions) - for config in result.global_configs: - if config and config not in self.global_plugin_configs: - self.global_plugin_configs.append(config) - - # Defaults-level configs (e.g., log-format) - for config in result.defaults_configs: - if config and config not in self.defaults_plugin_configs: - self.defaults_plugin_configs.append(config) - - # Store domain plugin configs for this host - easymapping[port]["hosts"][hostname]["plugin_configs"] = plugin_configs_for_host - except Exception as e: - logger_easyhaproxy.warning(f"Failed to execute domain plugins for {hostname}: {e}") - easymapping[port]["hosts"][hostname]["plugin_configs"] = [] - else: - easymapping[port]["hosts"][hostname]["plugin_configs"] = [] - - if certbot or clone_to_ssl: - if "443" not in easymapping: - easymapping["443"] = { - "mode": "http", - "ssl-check": "ssl", - "port": "443", - "hosts": dict(), - "redirect": dict(), - } - easymapping["443"]["hosts"][hostname] = dict(easymapping[port]["hosts"][hostname]) - easymapping["443"]["hosts"][hostname]["certbot"] = False - easymapping["443"]["hosts"][hostname]["redirect_ssl"] = False - easymapping["443"]["ssl"] = True - self.certbot_hosts.append( - hostname) if certbot and hostname not in self.certbot_hosts else self.certbot_hosts - - # handle SSL - ssl_label = self.label.create([definition, "sslcert"]) - if self.label.has_label(ssl_label): - filename = f"{d[host_label]}.pem" - easymapping[port]["ssl"] = True if not clone_to_ssl else False - self.certs[filename] = base64.b64decode(d[ssl_label]).decode('ascii') - - if self.label.get_bool(self.label.create([definition, "ssl"])): - easymapping[port]["ssl"] = True if not clone_to_ssl else False - - return easymapping.values() +__all__ = ["DockerLabelHandler", "HaproxyConfigGenerator"] \ No newline at end of file diff --git a/src/easymapping/config_generator.py b/src/easymapping/config_generator.py new file mode 100644 index 0000000..01c0524 --- /dev/null +++ b/src/easymapping/config_generator.py @@ -0,0 +1,310 @@ +import base64 +import os +import re + +from jinja2 import Environment, FileSystemLoader + +from functions import Functions, logger_easyhaproxy + +from .label_handler import DockerLabelHandler + + +class HaproxyConfigGenerator: + def __init__(self, mapping): + self.mapping = mapping + self.mapping.setdefault("ssl_mode", 'default') + self.mapping.setdefault("certbot", {"email": "", "server": False, "eab_kid": False, "eab_hmac_key": False}) + self.mapping["ssl_mode"] = self.mapping["ssl_mode"].lower() + self.label = DockerLabelHandler(mapping['lookup_label'] if 'lookup_label' in mapping else "easyhaproxy") + self.certbot_hosts = [] + self.serving_hosts = [] + self.certs = {} + self.defaults_plugin_configs = [] + + # Initialize plugin system + try: + from plugins import PluginManager + self.plugin_manager = PluginManager( + abort_on_error=self.mapping.get("plugins", {}).get("abort_on_error", False) + ) + self.plugin_manager.load_plugins() + self.plugin_manager.configure_plugins(self.mapping.get("plugins", {})) + self.plugin_manager.initialize_plugins() + self.global_plugin_configs = [] + except Exception as e: + # If plugin system fails to initialize, log but continue + logger_easyhaproxy.warning(f"Failed to initialize plugin system: {e}") + self.plugin_manager = None + self.global_plugin_configs = [] + + def generate(self, container_metadata={}): + self.mapping.setdefault("easymapping", []) + + if container_metadata != {}: + self.mapping["easymapping"] = self.parse(container_metadata) + + # Execute global plugins + if self.plugin_manager: + try: + from plugins import PluginContext + global_context = PluginContext( + parsed_object=container_metadata, + easymapping=self.mapping.get("easymapping", []), + container_env=self.mapping, + domain=None, + port=None, + host_config=None + ) + + # Get enabled plugins from config + enabled_list = self.mapping.get("plugins", {}).get("enabled", []) + # If enabled list contains only empty string, treat as no plugins enabled + if enabled_list and len(enabled_list) > 0 and enabled_list[0] == "": + enabled_list = [] + + global_results = self.plugin_manager.execute_global_plugins(global_context, enabled_list) + + # Extract all plugin configs in a single loop + for result in global_results: + # HAProxy config snippets + if result.haproxy_config: + self.global_plugin_configs.append(result.haproxy_config) + + # Global-level configs (e.g., fcgi-app definitions) + for config in result.global_configs: + if config and config not in self.global_plugin_configs: + self.global_plugin_configs.append(config) + + # Defaults-level configs (e.g., log-format) + for config in result.defaults_configs: + if config and config not in self.defaults_plugin_configs: + self.defaults_plugin_configs.append(config) + except Exception as e: + logger_easyhaproxy.warning(f"Failed to execute global plugins: {e}") + + templates_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'templates') + file_loader = FileSystemLoader(templates_dir) + env = Environment(loader=file_loader) + env.trim_blocks = True + env.lstrip_blocks = True + env.rstrip_blocks = True + template = env.get_template('haproxy.cfg.j2') + return template.render( + data=self.mapping, + global_plugin_configs=self.global_plugin_configs, + defaults_plugin_configs=self.defaults_plugin_configs + ) + + def parse(self, container_metadata): + easymapping = dict() + + for container in container_metadata: + d = container_metadata[container] + + # Extract the definitions dynamically + definitions = {} + r = re.compile(self.label.get_lookup_label() + r"\.(.*)\..*") + for key in d.keys(): + if r.match(key): + definitions[r.search(key).group(1)] = 1 + + if len(definitions.keys()) == 0: + continue + + self.label.set_data(d) + + # Parse each definition found. + for definition in sorted(definitions.keys()): + mode = self.label.get( + self.label.create([definition, "mode"]), + "http" + ) + + # TODO: we can ignore "host" in TCP, but it would break the template + host_label = self.label.create([definition, "host"]) + if not self.label.has_label(host_label): + continue + + port = self.label.get( + self.label.create([definition, "port"]), + "80" + ) + + certbot = self.label.get_bool( + self.label.create([definition, "certbot"]), + False + ) and self.mapping["certbot"]["email"] != "" + clone_to_ssl = self.label.get_bool( + self.label.create([definition, "clone_to_ssl"]) + ) + + # Check if this is a redirect-only entry (no backend) + redirect_only = self.label.get_bool( + self.label.create([definition, "redirect_only"]), + False + ) + + if port not in easymapping: + easymapping[port] = { + "mode": mode, + "ssl-check": "", + "port": port, + "hosts": dict(), + "redirect": dict(), + } + + if redirect_only: + easymapping[port]["redirect"].update(self.label.get_json( + self.label.create([definition, "redirect"]) + )) + continue + + # TODO: this could use `EXPOSE` from `Dockerfile`? + ct_port = self.label.get( + self.label.create([definition, "localport"]), + "80" + ) + + easymapping[port]["ssl-check"] = self.label.get( + self.label.create([definition, "ssl-check"]), + "" + ) + + # Protocol for backend server communication (e.g., fcgi, h2) + proto = self.label.get( + self.label.create([definition, "proto"]), + "" + ) + + # Unix socket path (alternative to host:port) + socket_path = self.label.get( + self.label.create([definition, "socket"]), + "" + ) + + for hostname in sorted(d[host_label].split(",")): + hostname = hostname.strip() + self.serving_hosts.append(f"{hostname}:{port}") + easymapping[port]["hosts"].setdefault(hostname, {}) + easymapping[port]["hosts"][hostname].setdefault("containers", []) + easymapping[port]["hosts"][hostname].setdefault("certbot", False) + easymapping[port]["hosts"][hostname].setdefault("proto", proto) + + # Determine server address: Unix socket or TCP host:port + if socket_path: + server_address = socket_path + else: + server_address = f"{container}:{ct_port}" + + easymapping[port]["hosts"][hostname]["containers"] += [server_address] + easymapping[port]["hosts"][hostname]["certbot"] = certbot + easymapping[port]["hosts"][hostname]["redirect_ssl"] = self.label.get_bool( + self.label.create([definition, "redirect_ssl"]) + ) + easymapping[port]["hosts"][hostname]["balance"] = self.label.get( + self.label.create([definition, "balance"]), + "roundrobin" + ) + + easymapping[port]["redirect"] = self.label.get_json( + self.label.create([definition, "redirect"]) + ) + + # Execute domain plugins for this host + if self.plugin_manager: + try: + from plugins import PluginContext + + domain_context = PluginContext( + parsed_object=container_metadata, + easymapping=easymapping, + container_env=self.mapping, + domain=hostname, + port=port, + host_config=easymapping[port]["hosts"][hostname] + ) + + # Check if plugins are enabled for this domain (from labels) + enabled_plugins = [] + if self.label.has_label(self.label.create([definition, "plugins"])): + enabled_plugins = self.label.get( + self.label.create([definition, "plugins"]), + "" + ).split(",") + enabled_plugins = [p.strip() for p in enabled_plugins if p.strip()] + + # Extract plugin configurations from labels + # Format: easyhaproxy.http.plugin.PLUGIN_NAME.CONFIG_KEY + plugin_configs = {} + for plugin_name in enabled_plugins: + plugin_configs[plugin_name] = {} + # Look for all labels matching easyhaproxy.{definition}.plugin.{plugin_name}.* + plugin_label_prefix = self.label.create([definition, "plugin", plugin_name]) + for label_key in d.keys(): + if label_key.startswith(plugin_label_prefix + "."): + # Extract config key (everything after plugin_label_prefix + ".") + config_key = label_key[len(plugin_label_prefix) + 1:] + plugin_configs[plugin_name][config_key] = d[label_key] + + # Configure plugins with label-specific configs before execution + for plugin_name, config in plugin_configs.items(): + if plugin_name in self.plugin_manager.plugins: + self.plugin_manager.plugins[plugin_name].configure(config) + + domain_results = self.plugin_manager.execute_domain_plugins( + domain_context, + enabled_list=enabled_plugins + ) + + # Extract all plugin configs in a single loop + plugin_configs_for_host = [] + for result in domain_results: + # HAProxy config snippets for this domain + if result.haproxy_config: + plugin_configs_for_host.append(result.haproxy_config) + + # Global-level configs (e.g., fcgi-app definitions) + for config in result.global_configs: + if config and config not in self.global_plugin_configs: + self.global_plugin_configs.append(config) + + # Defaults-level configs (e.g., log-format) + for config in result.defaults_configs: + if config and config not in self.defaults_plugin_configs: + self.defaults_plugin_configs.append(config) + + # Store domain plugin configs for this host + easymapping[port]["hosts"][hostname]["plugin_configs"] = plugin_configs_for_host + except Exception as e: + logger_easyhaproxy.warning(f"Failed to execute domain plugins for {hostname}: {e}") + easymapping[port]["hosts"][hostname]["plugin_configs"] = [] + else: + easymapping[port]["hosts"][hostname]["plugin_configs"] = [] + + if certbot or clone_to_ssl: + if "443" not in easymapping: + easymapping["443"] = { + "mode": "http", + "ssl-check": "ssl", + "port": "443", + "hosts": dict(), + "redirect": dict(), + } + easymapping["443"]["hosts"][hostname] = dict(easymapping[port]["hosts"][hostname]) + easymapping["443"]["hosts"][hostname]["certbot"] = False + easymapping["443"]["hosts"][hostname]["redirect_ssl"] = False + easymapping["443"]["ssl"] = True + self.certbot_hosts.append( + hostname) if certbot and hostname not in self.certbot_hosts else self.certbot_hosts + + # handle SSL + ssl_label = self.label.create([definition, "sslcert"]) + if self.label.has_label(ssl_label): + filename = f"{d[host_label]}.pem" + easymapping[port]["ssl"] = True if not clone_to_ssl else False + self.certs[filename] = base64.b64decode(d[ssl_label]).decode('ascii') + + if self.label.get_bool(self.label.create([definition, "ssl"])): + easymapping[port]["ssl"] = True if not clone_to_ssl else False + + return easymapping.values() \ No newline at end of file diff --git a/src/easymapping/label_handler.py b/src/easymapping/label_handler.py new file mode 100644 index 0000000..220943d --- /dev/null +++ b/src/easymapping/label_handler.py @@ -0,0 +1,52 @@ +import json + +from functions import logger_easyhaproxy + + +class DockerLabelHandler: + def __init__(self, label): + self.__data = None + self.__label_base = label + + def get_lookup_label(self): + return self.__label_base + + def create(self, key): + if isinstance(key, str): + return f"{self.__label_base}.{key}" + + return "{}.{}".format(self.__label_base, ".".join(key)) + + def get(self, label, default_value=""): + if self.has_label(label): + return self.__data[label] + return default_value + + def get_bool(self, label, default_value=False): + if self.has_label(label): + return self.__data[label].lower() in ["true", "1", "yes"] + return default_value + + def get_json(self, label, default_value=None): + if default_value is None: + default_value = {} + if self.has_label(label): + value = self.__data[label] + if not value: # Handle empty strings + return default_value + try: + return json.loads(value) + except json.JSONDecodeError as e: + logger_easyhaproxy.error( + f"Invalid JSON in label '{label}': {value}. Error: {e}. Using default value." + ) + return default_value + return default_value + + def set_data(self, data): + self.__data = data + + def has_label(self, label): + if label in self.__data: + return True + return False \ No newline at end of file diff --git a/src/functions/__init__.py b/src/functions/__init__.py index 52fdcad..bd26cf3 100644 --- a/src/functions/__init__.py +++ b/src/functions/__init__.py @@ -1,723 +1,21 @@ -import logging -import os -import shlex -import shutil -import subprocess -import sys -import time -from datetime import datetime -from multiprocessing import Process -from pathlib import Path -from typing import Final - -import psutil -import requests -from OpenSSL import crypto - - -class ContainerEnv: - @staticmethod - def read(yaml_config=None): - """ - Read configuration from environment variables, optionally merged with YAML config. - - Args: - yaml_config: Optional dict from YAML file (for static mode). YAML values take precedence. - - Returns: - Dict with configuration settings - """ - # Convert YAML config to environment variables first (if provided) - if yaml_config: - ContainerEnv._yaml_to_env(yaml_config) - - env_vars = { - "customerrors": True if os.getenv("HAPROXY_CUSTOMERRORS") == "true" else False, - "ssl_mode": os.getenv("EASYHAPROXY_SSL_MODE").lower() if os.getenv("EASYHAPROXY_SSL_MODE") else 'default' - } - - if os.getenv("HAPROXY_PASSWORD"): - env_vars["stats"] = { - "username": os.getenv("HAPROXY_USERNAME") if os.getenv("HAPROXY_USERNAME") else "admin", - "password": os.getenv("HAPROXY_PASSWORD"), - "port": os.getenv("HAPROXY_STATS_PORT") if os.getenv("HAPROXY_STATS_PORT") else "1936", - "cors_origin": os.getenv("HAPROXY_STATS_CORS_ORIGIN", ""), - } - - env_vars["lookup_label"] = os.getenv("EASYHAPROXY_LABEL_PREFIX") if os.getenv( - "EASYHAPROXY_LABEL_PREFIX") else "easyhaproxy" - - env_vars["logLevel"] = { - "easyhaproxy": os.getenv("EASYHAPROXY_LOG_LEVEL") if os.getenv( - "EASYHAPROXY_LOG_LEVEL") else Functions.DEBUG, - "haproxy": os.getenv("HAPROXY_LOG_LEVEL") if os.getenv("HAPROXY_LOG_LEVEL") else Functions.INFO, - "certbot": os.getenv("CERTBOT_LOG_LEVEL") if os.getenv("CERTBOT_LOG_LEVEL") else Functions.DEBUG, - } - - env_vars["certbot"] = { - "autoconfig": os.getenv("EASYHAPROXY_CERTBOT_AUTOCONFIG", ""), - "email": os.getenv("EASYHAPROXY_CERTBOT_EMAIL", ""), - "server": os.getenv("EASYHAPROXY_CERTBOT_SERVER", False), - "eab_kid": os.getenv("EASYHAPROXY_CERTBOT_EAB_KID", ""), - "eab_hmac_key": os.getenv("EASYHAPROXY_CERTBOT_EAB_HMAC_KEY", ""), - "retry_count": int(os.getenv("EASYHAPROXY_CERTBOT_RETRY_COUNT", 60)), - "preferred_challenges": os.getenv("EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES", "http"), - "manual_auth_hook": os.getenv("EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK", False), - } - - if env_vars["certbot"]["autoconfig"] != "" and not env_vars["certbot"]["server"] and env_vars["certbot"]["email"] != "": - if env_vars["certbot"]["autoconfig"] == "letsencrypt": - env_vars["certbot"]["server"] = "https://acme-v02.api.letsencrypt.org/directory" - - if env_vars["certbot"]["autoconfig"] == "letsencrypt_test": - env_vars["certbot"]["server"] = "https://acme-staging-v02.api.letsencrypt.org/directory" - - if env_vars["certbot"]["autoconfig"] == "buypass": - env_vars["certbot"]["server"] = "https://api.buypass.com/acme/directory" - - if env_vars["certbot"]["autoconfig"] == "buypass_test": - env_vars["certbot"]["server"] = "https://api.test4.buypass.no/acme/directory" - - if env_vars["certbot"]["autoconfig"] == "sslcom_rca": - env_vars["certbot"]["server"] = "https://acme.ssl.com/sslcom-dv-rsa" - - if env_vars["certbot"]["autoconfig"] == "sslcom_ecc": - env_vars["certbot"]["server"] = "https://acme.ssl.com/sslcom-dv-ecc" - - if env_vars["certbot"]["autoconfig"] == "google": - env_vars["certbot"]["server"] = "https://dv.acme-v02.api.pki.goog/directory" - - if env_vars["certbot"]["autoconfig"] == "google_test": - env_vars["certbot"]["server"] = "https://dv.acme-v02.test-api.pki.goog/directory" - - if env_vars["certbot"]["autoconfig"] == "zerossl": - url = "https://api.zerossl.com/acme/eab-credentials-email" - headers = {"Content-Type": "application/x-www-form-urlencoded"} - data = "email=" + env_vars["certbot"]["email"] - resp = requests.post(url, headers=headers, data=data).json() - - if resp["success"]: - env_vars["certbot"]["server"] = "https://acme.zerossl.com/v2/DV90" - env_vars["certbot"]["eab_kid"] = os.environ['EASYHAPROXY_CERTBOT_EAB_KID'] = resp["eab_kid"] - env_vars["certbot"]["eab_hmac_key"] = os.environ['EASYHAPROXY_CERTBOT_EAB_HMAC_KEY'] = resp["eab_hmac_key"] - else: - del os.environ["EASYHAPROXY_CERTBOT_EMAIL"] - logger_certbot.error("Could not obtain ZeroSSL credentials " + resp["error"]["type"]) - - os.environ['EASYHAPROXY_CERTBOT_SERVER'] = env_vars["certbot"]["server"] - - # Plugin configuration - env_vars["plugins"] = { - "abort_on_error": os.getenv("EASYHAPROXY_PLUGINS_ABORT_ON_ERROR", "false").lower() == "true", - "enabled": os.getenv("EASYHAPROXY_PLUGINS_ENABLED", "").split(",") if os.getenv("EASYHAPROXY_PLUGINS_ENABLED") else [], - "config": {} # Individual plugin configs from env vars - } - - # Parse individual plugin configs (e.g., EASYHAPROXY_PLUGIN_CLOUDFLARE_*) - for key, value in os.environ.items(): - if key.startswith("EASYHAPROXY_PLUGIN_"): - parts = key.split("_", 3) # ['EASYHAPROXY', 'PLUGIN', 'NAME', 'KEY'] - if len(parts) >= 4: - plugin_name = parts[2].lower() - config_key = "_".join(parts[3:]).lower() - env_vars["plugins"]["config"].setdefault(plugin_name, {}) - env_vars["plugins"]["config"][plugin_name][config_key] = value - - # Ingress status update configuration - env_vars["update_ingress_status"] = os.getenv("EASYHAPROXY_UPDATE_INGRESS_STATUS", "true").lower() == "true" - env_vars["deployment_mode"] = os.getenv("EASYHAPROXY_DEPLOYMENT_MODE", "auto") - env_vars["external_hostname"] = os.getenv("EASYHAPROXY_EXTERNAL_HOSTNAME", "") - env_vars["ingress_status_update_interval"] = int(os.getenv("EASYHAPROXY_STATUS_UPDATE_INTERVAL", "30")) - - return env_vars - - @staticmethod - def _yaml_to_env(yaml_config): - """Convert YAML configuration to environment variables""" - - # Convert customerrors - if 'customerrors' in yaml_config: - os.environ['HAPROXY_CUSTOMERRORS'] = 'true' if yaml_config['customerrors'] else 'false' - - # Convert ssl_mode - if 'ssl_mode' in yaml_config: - os.environ['EASYHAPROXY_SSL_MODE'] = str(yaml_config['ssl_mode']) - - # Convert stats - if 'stats' in yaml_config: - stats = yaml_config['stats'] - if 'username' in stats: - os.environ['HAPROXY_USERNAME'] = str(stats['username']) - if 'password' in stats: - os.environ['HAPROXY_PASSWORD'] = str(stats['password']) - if 'port' in stats: - os.environ['HAPROXY_STATS_PORT'] = str(stats['port']) - if 'cors_origin' in stats: - os.environ['HAPROXY_STATS_CORS_ORIGIN'] = str(stats['cors_origin']) - - # Convert logLevel - if 'logLevel' in yaml_config: - log_level = yaml_config['logLevel'] - for source, level in log_level.items(): - os.environ[source.upper() + '_LOG_LEVEL'] = str(level) - - # Convert certbot - if 'certbot' in yaml_config: - certbot = yaml_config['certbot'] - for config, value in certbot.items(): - os.environ['EASYHAPROXY_CERTBOT_' + config.upper()] = str(value) - - # Convert plugins - if 'plugins' in yaml_config: - plugins = yaml_config['plugins'] - - # Convert enabled list - if 'enabled' in plugins: - enabled_list = plugins['enabled'] if isinstance(plugins['enabled'], list) else [plugins['enabled']] - os.environ['EASYHAPROXY_PLUGINS_ENABLED'] = ','.join(enabled_list) - - # Convert abort_on_error - if 'abort_on_error' in plugins: - os.environ['EASYHAPROXY_PLUGINS_ABORT_ON_ERROR'] = 'true' if plugins['abort_on_error'] else 'false' - - # Convert plugin configs - if 'config' in plugins: - for plugin_name, plugin_config in plugins['config'].items(): - for config_key, config_value in plugin_config.items(): - # Convert to env var format: EASYHAPROXY_PLUGIN__ - env_key = f"EASYHAPROXY_PLUGIN_{plugin_name.upper()}_{config_key.upper()}" - - # Convert list values to comma-separated strings - if isinstance(config_value, list): - env_value = ','.join(str(v) for v in config_value) - else: - env_value = str(config_value) - - os.environ[env_key] = env_value - - -class Functions: - HAPROXY_LOG: Final[str] = "HAPROXY" - EASYHAPROXY_LOG: Final[str] = "EASYHAPROXY" - CERTBOT_LOG: Final[str] = "CERTBOT" - INIT_LOG: Final[str] = "INIT" - - TRACE: Final[str] = "TRACE" - DEBUG: Final[str] = "DEBUG" - INFO: Final[str] = "INFO" - WARN: Final[str] = "WARN" - ERROR: Final[str] = "ERROR" - FATAL: Final[str] = "FATAL" - - @staticmethod - def setup_log(source): - level = os.getenv(f"{source.name.upper()}_LOG_LEVEL", "").upper() - level_importance = { - Functions.TRACE: logging.DEBUG, - Functions.DEBUG: logging.DEBUG, - Functions.INFO: logging.INFO, - Functions.WARN: logging.WARNING, - Functions.ERROR: logging.ERROR, - Functions.FATAL: logging.FATAL - } - selected_level = level_importance[level] if level in level_importance else logging.INFO - - log_source_handler = logging.StreamHandler(sys.stdout) - log_source_formatter = logging.Formatter('%(name)s [%(asctime)s] %(levelname)s - %(message)s') - log_source_handler.setFormatter(log_source_formatter) - log_source_handler.addFilter(SingleLineNonEmptyFilter()) - source.setLevel(selected_level) - source.addHandler(log_source_handler) - return selected_level - - @staticmethod - def load(filename): - with open(filename) as content_file: - return content_file.read() - - @staticmethod - def save(filename, contents): - with open(filename, 'w') as file: - file.write(contents) - - @staticmethod - def run_bash(log_source, command, log_output=True, return_result=True): - if not isinstance(command, (list, tuple)): - command = shlex.split(command) - - try: - process = subprocess.Popen(command, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - universal_newlines=True) - - output = [] - - while True: - line = process.stdout.readline().rstrip() - error_line = process.stderr.readline().rstrip() - output.append(line) if return_result else None - log_source.info(line) if log_output and len(line) > 0 else None - log_source.warning(error_line) if len(error_line) > 0 else None - return_code = process.poll() - if return_code is not None: - lines = [] - error_line = process.stderr.readline().rstrip() - for line in process.stdout.readlines(): - output.append(line.rstrip()) if return_result else None - lines.append(line.rstrip()) - log_source.info(lines) if log_output and len(lines) > 0 else None - log_source.warning(error_line) if len(error_line) > 0 else None - break - - return [return_code, output] - except Exception as e: - log_source.error(f"{e}") - return [-99, e] - - -class classproperty: - """Decorator for class-level properties.""" - def __init__(self, func): - self.func = func - - def __get__(self, obj, owner): - return self.func(owner) - - -class Consts: - """Configuration constants with dynamic path resolution based on EASYHAPROXY_BASE_PATH.""" - _base_path = None - - @classproperty - def base_path(cls): - """Base directory for all EasyHAProxy files.""" - if cls._base_path is None: - if os.getenv("EASYHAPROXY_BASE_PATH"): - default = os.getenv("EASYHAPROXY_BASE_PATH") - elif os.getuid() == 0: - default = "/etc/easyhaproxy" - else: - default = str(Path.home() / "easyhaproxy") - cls._base_path = default - return cls._base_path - - @classmethod - def reset(cls): - """Reset cached base path to pick up environment variable changes.""" - cls._base_path = None - - @classproperty - def easyhaproxy_config(cls): - """Path to static configuration file.""" - return f"{cls.base_path}/static/config.yml" - - @classproperty - def haproxy_config(cls): - """Path to generated HAProxy configuration file.""" - return f"{cls.base_path}/haproxy/haproxy.cfg" - - @classproperty - def custom_config_folder(cls): - """Path to custom HAProxy config snippets directory.""" - return f"{cls.base_path}/haproxy/conf.d" - - @classproperty - def certs_certbot(cls): - """Path to Certbot/ACME certificates directory.""" - return f"{cls.base_path}/certs/certbot" - - @classproperty - def certs_haproxy(cls): - """Path to user-provided certificates directory.""" - return f"{cls.base_path}/certs/haproxy" - - -class DaemonizeHAProxy: - HAPROXY_START: Final[str] = "start" - HAPROXY_RELOAD: Final[str] = "reload" - - def __init__(self, custom_config_folder = None): - self.process = None - self.thread = None - self.sleep_secs = None - self.custom_config_folder = custom_config_folder if custom_config_folder is not None else Consts.custom_config_folder - - def haproxy(self, action): - error = self.__prepare(self.get_haproxy_command(action), action) - - if error or self.process is None: - logger_haproxy.fatal(f"Failed to start HAProxy ({action}). Exiting.") - import sys - sys.exit(1) - - self.thread = Process(target=self.__start, args=()) - self.thread.start() - - @staticmethod - def get_haproxy_bin() -> str: - return shutil.which('haproxy') or '/usr/sbin/haproxy' - - def get_haproxy_command(self, action, pid_file="/run/haproxy.pid"): - haproxy_bin = DaemonizeHAProxy.get_haproxy_bin() - custom_config_files = "" - if len(list(self.get_custom_config_files().keys())) != 0: - custom_config_files = f"-f {self.custom_config_folder}" - - if action == DaemonizeHAProxy.HAPROXY_START or not os.path.exists(pid_file): - return f"{haproxy_bin} -W -f {Consts.haproxy_config} {custom_config_files} -p {pid_file} -S /var/run/haproxy.sock" - else: - return_code, output = Functions().run_bash(logger_haproxy, f"cat {pid_file}", log_output=False) - pid = "".join(output).rstrip() - if psutil.pid_exists(int(pid)): - return f"{haproxy_bin} -W -f {Consts.haproxy_config} {custom_config_files} -p {pid_file} -x /var/run/haproxy.sock -sf {pid}" - else: - os.unlink(pid_file) - logger_haproxy.warning( - f"PID file {pid_file} does not exist. Restarting haproxy instead of reload." - ) - return self.get_haproxy_command(DaemonizeHAProxy.HAPROXY_START, pid_file) - - def __validate_config(self): - """Validate HAProxy configuration before starting.""" - validation_cmd = ["haproxy", "-c", "-f", Consts.haproxy_config] - - # Add custom config files if they exist - for config_file in self.get_custom_config_files().keys(): - validation_cmd.extend(["-f", config_file]) - - try: - result = subprocess.run( - validation_cmd, - capture_output=True, - text=True, - timeout=10 - ) - - if result.returncode != 0: - return result.stderr if result.stderr else result.stdout - return None - - except subprocess.TimeoutExpired: - return "HAProxy configuration validation timed out" - except Exception as e: - return f"Error validating configuration: {e}" - - def __prepare(self, command, action=None): - if not isinstance(command, (list, tuple)): - command = shlex.split(command) - - # Validate HAProxy config before starting (but not on reload - HAProxy validates itself during reload) - if action == DaemonizeHAProxy.HAPROXY_START: - validation_error = self.__validate_config() - if validation_error: - logger_haproxy.fatal(f"HAProxy configuration validation failed:\n{validation_error}") - return validation_error - - try: - logger_haproxy.debug(f"HAPROXY command: {command}") - self.process = subprocess.Popen(command, - shell=False, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - bufsize=-1, - universal_newlines=True) - return None - - except Exception as e: - error_msg = f"Failed to start HAProxy process: {e}" - logger_haproxy.error(error_msg) - return error_msg - - def __start(self): - try: - with self.process.stdout: - for line in iter(self.process.stdout.readline, b''): - logger_haproxy.info(line.rstrip()) - - return_code = self.process.wait() - logger_haproxy.debug(f"Return code {return_code}") - - except Exception as e: - logger_haproxy.error(f"{e}") - - def is_alive(self): - return self.thread.is_alive() - - def kill(self): - self.process.kill() - self.thread.kill() - - def terminate(self): - self.process.terminate() - self.thread.terminate() - - def sleep(self): - if self.sleep_secs is None: - try: - self.sleep_secs = int(os.getenv("EASYHAPROXY_REFRESH_CONF", "10")) - except ValueError: - self.sleep_secs = 10 - - time.sleep(self.sleep_secs) - - def get_custom_config_files(self): - if not os.path.exists(self.custom_config_folder): - return {} - - files = {} - for file in os.listdir(self.custom_config_folder): - if file.endswith(".cfg"): - files[os.path.join(self.custom_config_folder, file)] = os.path.getmtime(os.path.join(self.custom_config_folder, file)) - return dict(sorted(files.items(), key=lambda t: t[0])) - - -class Certbot: - def __init__(self, certs): - env = ContainerEnv.read() - - self.certs = certs - self.email = env["certbot"]["email"] - self.acme_server = self.set_acme_server(env["certbot"]["server"]) - self.eab_kid = self.set_eab_kid(env["certbot"]["eab_kid"]) - self.eab_hmac_key = self.set_eab_hmac_key(env["certbot"]["eab_hmac_key"]) - self.freeze_issue = {} - self.retry_count = env["certbot"]["retry_count"] - self.certbot_preferred_challenges = env["certbot"]["preferred_challenges"] - self.certbot_manual_auth_hook = env["certbot"]["manual_auth_hook"] - - @staticmethod - def set_acme_server(acme_server): - if not acme_server: - return "" - if acme_server.lower() == "staging": - return "--staging" - elif acme_server.lower().startswith("http"): - return "--server " + acme_server - else: - return "" - - @staticmethod - def set_eab_kid(eab_kid): - if eab_kid != "": - return f'--eab-kid "{eab_kid}"' - else: - return "" - - @staticmethod - def set_eab_hmac_key(eab_hmac_key): - if eab_hmac_key != "": - return f'--eab-hmac-key "{eab_hmac_key}"' - else: - return "" - - @staticmethod - def check_acme_environment_ready(email, acme_server): - """ - Check if ACME environment is ready for certificate operations. - - Args: - email: EASYHAPROXY_CERTBOT_EMAIL value - acme_server: Processed ACME server string from set_acme_server() - - Returns: - tuple: (is_ready: bool, error_message: str) - """ - # Check 1: Email configured - if not email or email == "": - return False, "ACME email not configured (EASYHAPROXY_CERTBOT_EMAIL)" - - # Check 2: ACME server configured - if not acme_server or acme_server == "": - return False, "ACME server not configured (EASYHAPROXY_CERTBOT_SERVER)" - - # Check 3: ACME server reachability (if URL provided) - if "--server " in acme_server: - server_url = acme_server.replace("--server ", "") - try: - # Use 10s timeout, respect REQUESTS_CA_BUNDLE for Pebble CA - response = requests.get(server_url, timeout=10, verify=os.getenv("REQUESTS_CA_BUNDLE", True)) - if response.status_code != 200: - return False, f"ACME server {server_url} returned HTTP {response.status_code}" - - # Validate ACME directory structure (RFC 8555) - data = response.json() - if "newAccount" not in data: - return False, f"ACME server {server_url} returned invalid ACME directory" - except requests.exceptions.RequestException as e: - return False, f"ACME server {server_url} not reachable: {str(e)}" - except Exception as e: - return False, f"ACME server validation failed: {str(e)}" - - return True, "" - - def check_certificates(self, hosts): - if self.email == "" or len(hosts) == 0: - return False - - try: - request_certs = [] - renew_certs = [] - for host in hosts: - cert_status = self.get_certificate_status(host) - host_arg = f'-d {host}' - if cert_status == "ok" or cert_status == "error": - continue - elif host in self.freeze_issue: - freeze_count = self.freeze_issue.pop(host, 0) - if freeze_count > 0: - logger_certbot.debug(f"Waiting freezing period ({freeze_count}) for {host} due previous errors") - self.freeze_issue[host] = freeze_count-1 - elif cert_status == "not_found" or cert_status == "expired": - logger_certbot.debug(f"[{cert_status}] Request new certificate for {host}") - request_certs.append(host_arg) - elif cert_status == "expiring": - logger_certbot.debug(f"[{cert_status}] Renew certificate for {host}") - renew_certs.append(host_arg) - - certbot_certonly = ('/usr/bin/certbot certonly {acme_server}' - ' --config-dir {base_path}/certs' - ' --work-dir {base_path}/certs/work' - ' --logs-dir {base_path}/certs/logs' - ' --preferred-challenges {challenge}' - ' --agree-tos' - ' --issuance-timeout 90' - ' --no-eff-email' - ' --non-interactive' - ' --max-log-backups=0' - ' {eab_kid} {eab_hmac_key}' - ' {certs} --email {email}'.format(eab_kid=self.eab_kid, - eab_hmac_key=self.eab_hmac_key, - certs=' '.join(request_certs), - email=self.email, - challenge=self.certbot_preferred_challenges, - acme_server=self.acme_server, - base_path=Consts.base_path) - ) - - if 'http' in self.certbot_preferred_challenges: - certbot_certonly += (' --http-01-port 2080' - ' --standalone' - ) - - if self.certbot_manual_auth_hook: - certbot_certonly += f' --manual --manual-auth-hook \'{self.certbot_manual_auth_hook}\'' - - if logger_certbot.level == logging.DEBUG: - certbot_certonly += ' -v' - - logger_certbot.debug(f"certbot_certonly: {certbot_certonly}") - - ret_reload = False - return_code_issue = 0 - return_code_renew = 0 - if len(request_certs) > 0: - return_code_issue, output = Functions.run_bash(logger_certbot, certbot_certonly, return_result=False) - ret_reload = True - - if len(renew_certs) > 0: - certbot_renew = f"/usr/bin/certbot renew --config-dir {Consts.base_path}/certs --work-dir {Consts.base_path}/certs/work --logs-dir {Consts.base_path}/certs/logs" - return_code_renew, output = Functions.run_bash(logger_certbot, certbot_renew, return_result=False) - ret_reload = True - - if ret_reload: - self.find_live_certificates() - - if return_code_issue != 0: - self.find_missing_certificates(request_certs) - if return_code_renew != 0: - self.find_missing_certificates(renew_certs) - - return ret_reload - except Exception as e: - logger_certbot.error(f"{e}") - return False - - @staticmethod - def merge_certificate(cert, key, filename): - Functions.save(filename, cert + key) - - def find_live_certificates(self): - certbot_certs = f"{Consts.base_path}/certs/live/" - if not os.path.exists(certbot_certs): - return - for item in os.listdir(certbot_certs): - path = os.path.join(certbot_certs, item) - if os.path.isdir(path): - cert = Functions.load(os.path.join(path, "cert.pem")) - key = Functions.load(os.path.join(path, "privkey.pem")) - filename = f"{self.certs}/{item}.pem" - self.merge_certificate(cert, key, filename) - - def get_certificate_status(self, host): - current_time = time.time() - filename = f"{self.certs}/{host}.pem" - if not os.path.exists(filename): - return "not_found" - - try: - with open(filename, 'rb') as file: - certificate_str = file.read() - certificate = crypto.load_certificate(crypto.FILETYPE_PEM, certificate_str) - expiration_after = datetime.strptime(certificate.get_notAfter().decode()[:-1], '%Y%m%d%H%M%S').timestamp() - if current_time >= expiration_after: - return "expired" - elif (expiration_after - current_time) // (24 * 3600) <= 15: - return "expiring" - except Exception as e: - logger_certbot.error(f"Certificate {host} error {e}") - return "error" - - return "ok" - - def find_missing_certificates(self, hosts): - for host in hosts: - if host.startswith("-d "): - host = host[3:] - cert_status = self.get_certificate_status(host) - if cert_status != "ok": - self.freeze_issue[host] = self.retry_count - logger_certbot.debug(f"Freeze issuing ssl for {host} due failure. The certificate is {cert_status}") - - - -class SingleLineNonEmptyFilter(logging.Filter): - """ - Logging filter that ensures messages are single-line and non-empty. - - Collapses newlines into spaces and strips surrounding whitespace. - - Drops the record if the resulting message is empty. - """ - def filter(self, record: logging.LogRecord) -> int: - try: - msg = record.getMessage() - except Exception: - # If formatting fails, drop the record - return 0 - - # Convert any non-string to string representation - if not isinstance(msg, str): - msg = str(msg) - - # Collapse multi-line to single line and trim - sanitized = " ".join(msg.splitlines()).strip() - - if sanitized == "": - return 0 - - # If we changed the message, update the record and clear args - if sanitized != record.getMessage(): - record.msg = sanitized - record.args = () - return 1 - - -# #################################################################################################################### -# Setup Global Log -logger_init = logging.getLogger(Functions.INIT_LOG) -logger_haproxy = logging.getLogger(Functions.HAPROXY_LOG) -logger_easyhaproxy = logging.getLogger(Functions.EASYHAPROXY_LOG) -logger_certbot = logging.getLogger(Functions.CERTBOT_LOG) -Functions.setup_log(logger_init) -Functions.setup_log(logger_haproxy) -Functions.setup_log(logger_easyhaproxy) -Functions.setup_log(logger_certbot) +from .certbot import Certbot +from .consts import Consts, classproperty +from .container_env import ContainerEnv +from .filter import SingleLineNonEmptyFilter +from .functions import Functions +from .haproxy import DaemonizeHAProxy +from .loggers import logger_certbot, logger_easyhaproxy, logger_haproxy, logger_init + +__all__ = [ + "Certbot", + "classproperty", + "Consts", + "ContainerEnv", + "DaemonizeHAProxy", + "Functions", + "SingleLineNonEmptyFilter", + "logger_certbot", + "logger_easyhaproxy", + "logger_haproxy", + "logger_init", +] \ No newline at end of file diff --git a/src/functions/certbot.py b/src/functions/certbot.py new file mode 100644 index 0000000..b66f520 --- /dev/null +++ b/src/functions/certbot.py @@ -0,0 +1,220 @@ +import logging +import os +import time +from datetime import datetime + +import requests +from OpenSSL import crypto + +from .consts import Consts +from .container_env import ContainerEnv +from .functions import Functions +from .loggers import logger_certbot + + +class Certbot: + def __init__(self, certs): + env = ContainerEnv.read() + + self.certs = certs + self.email = env["certbot"]["email"] + self.acme_server = self.set_acme_server(env["certbot"]["server"]) + self.eab_kid = self.set_eab_kid(env["certbot"]["eab_kid"]) + self.eab_hmac_key = self.set_eab_hmac_key(env["certbot"]["eab_hmac_key"]) + self.freeze_issue = {} + self.retry_count = env["certbot"]["retry_count"] + self.certbot_preferred_challenges = env["certbot"]["preferred_challenges"] + self.certbot_manual_auth_hook = env["certbot"]["manual_auth_hook"] + + @staticmethod + def set_acme_server(acme_server): + if not acme_server: + return "" + if acme_server.lower() == "staging": + return "--staging" + elif acme_server.lower().startswith("http"): + return "--server " + acme_server + else: + return "" + + @staticmethod + def set_eab_kid(eab_kid): + if eab_kid != "": + return f'--eab-kid "{eab_kid}"' + else: + return "" + + @staticmethod + def set_eab_hmac_key(eab_hmac_key): + if eab_hmac_key != "": + return f'--eab-hmac-key "{eab_hmac_key}"' + else: + return "" + + @staticmethod + def check_acme_environment_ready(email, acme_server): + """ + Check if ACME environment is ready for certificate operations. + + Args: + email: EASYHAPROXY_CERTBOT_EMAIL value + acme_server: Processed ACME server string from set_acme_server() + + Returns: + tuple: (is_ready: bool, error_message: str) + """ + # Check 1: Email configured + if not email or email == "": + return False, "ACME email not configured (EASYHAPROXY_CERTBOT_EMAIL)" + + # Check 2: ACME server configured + if not acme_server or acme_server == "": + return False, "ACME server not configured (EASYHAPROXY_CERTBOT_SERVER)" + + # Check 3: ACME server reachability (if URL provided) + if "--server " in acme_server: + server_url = acme_server.replace("--server ", "") + try: + # Use 10s timeout, respect REQUESTS_CA_BUNDLE for Pebble CA + response = requests.get(server_url, timeout=10, verify=os.getenv("REQUESTS_CA_BUNDLE", True)) + if response.status_code != 200: + return False, f"ACME server {server_url} returned HTTP {response.status_code}" + + # Validate ACME directory structure (RFC 8555) + data = response.json() + if "newAccount" not in data: + return False, f"ACME server {server_url} returned invalid ACME directory" + except requests.exceptions.RequestException as e: + return False, f"ACME server {server_url} not reachable: {str(e)}" + except Exception as e: + return False, f"ACME server validation failed: {str(e)}" + + return True, "" + + def check_certificates(self, hosts): + if self.email == "" or len(hosts) == 0: + return False + + try: + request_certs = [] + renew_certs = [] + for host in hosts: + cert_status = self.get_certificate_status(host) + host_arg = f'-d {host}' + if cert_status == "ok" or cert_status == "error": + continue + elif host in self.freeze_issue: + freeze_count = self.freeze_issue.pop(host, 0) + if freeze_count > 0: + logger_certbot.debug(f"Waiting freezing period ({freeze_count}) for {host} due previous errors") + self.freeze_issue[host] = freeze_count-1 + elif cert_status == "not_found" or cert_status == "expired": + logger_certbot.debug(f"[{cert_status}] Request new certificate for {host}") + request_certs.append(host_arg) + elif cert_status == "expiring": + logger_certbot.debug(f"[{cert_status}] Renew certificate for {host}") + renew_certs.append(host_arg) + + certbot_certonly = ('/usr/bin/certbot certonly {acme_server}' + ' --config-dir {base_path}/certs' + ' --work-dir {base_path}/certs/work' + ' --logs-dir {base_path}/certs/logs' + ' --preferred-challenges {challenge}' + ' --agree-tos' + ' --issuance-timeout 90' + ' --no-eff-email' + ' --non-interactive' + ' --max-log-backups=0' + ' {eab_kid} {eab_hmac_key}' + ' {certs} --email {email}'.format(eab_kid=self.eab_kid, + eab_hmac_key=self.eab_hmac_key, + certs=' '.join(request_certs), + email=self.email, + challenge=self.certbot_preferred_challenges, + acme_server=self.acme_server, + base_path=Consts.base_path) + ) + + if 'http' in self.certbot_preferred_challenges: + certbot_certonly += (' --http-01-port 2080' + ' --standalone' + ) + + if self.certbot_manual_auth_hook: + certbot_certonly += f' --manual --manual-auth-hook \'{self.certbot_manual_auth_hook}\'' + + if logger_certbot.level == logging.DEBUG: + certbot_certonly += ' -v' + + logger_certbot.debug(f"certbot_certonly: {certbot_certonly}") + + ret_reload = False + return_code_issue = 0 + return_code_renew = 0 + if len(request_certs) > 0: + return_code_issue, output = Functions.run_bash(logger_certbot, certbot_certonly, return_result=False) + ret_reload = True + + if len(renew_certs) > 0: + certbot_renew = f"/usr/bin/certbot renew --config-dir {Consts.base_path}/certs --work-dir {Consts.base_path}/certs/work --logs-dir {Consts.base_path}/certs/logs" + return_code_renew, output = Functions.run_bash(logger_certbot, certbot_renew, return_result=False) + ret_reload = True + + if ret_reload: + self.find_live_certificates() + + if return_code_issue != 0: + self.find_missing_certificates(request_certs) + if return_code_renew != 0: + self.find_missing_certificates(renew_certs) + + return ret_reload + except Exception as e: + logger_certbot.error(f"{e}") + return False + + @staticmethod + def merge_certificate(cert, key, filename): + Functions.save(filename, cert + key) + + def find_live_certificates(self): + certbot_certs = f"{Consts.base_path}/certs/live/" + if not os.path.exists(certbot_certs): + return + for item in os.listdir(certbot_certs): + path = os.path.join(certbot_certs, item) + if os.path.isdir(path): + cert = Functions.load(os.path.join(path, "cert.pem")) + key = Functions.load(os.path.join(path, "privkey.pem")) + filename = f"{self.certs}/{item}.pem" + self.merge_certificate(cert, key, filename) + + def get_certificate_status(self, host): + current_time = time.time() + filename = f"{self.certs}/{host}.pem" + if not os.path.exists(filename): + return "not_found" + + try: + with open(filename, 'rb') as file: + certificate_str = file.read() + certificate = crypto.load_certificate(crypto.FILETYPE_PEM, certificate_str) + expiration_after = datetime.strptime(certificate.get_notAfter().decode()[:-1], '%Y%m%d%H%M%S').timestamp() + if current_time >= expiration_after: + return "expired" + elif (expiration_after - current_time) // (24 * 3600) <= 15: + return "expiring" + except Exception as e: + logger_certbot.error(f"Certificate {host} error {e}") + return "error" + + return "ok" + + def find_missing_certificates(self, hosts): + for host in hosts: + if host.startswith("-d "): + host = host[3:] + cert_status = self.get_certificate_status(host) + if cert_status != "ok": + self.freeze_issue[host] = self.retry_count + logger_certbot.debug(f"Freeze issuing ssl for {host} due failure. The certificate is {cert_status}") \ No newline at end of file diff --git a/src/functions/consts.py b/src/functions/consts.py new file mode 100644 index 0000000..0475933 --- /dev/null +++ b/src/functions/consts.py @@ -0,0 +1,59 @@ +import os +from pathlib import Path + + +class classproperty: + """Decorator for class-level properties.""" + def __init__(self, func): + self.func = func + + def __get__(self, obj, owner): + return self.func(owner) + + +class Consts: + """Configuration constants with dynamic path resolution based on EASYHAPROXY_BASE_PATH.""" + _base_path = None + + @classproperty + def base_path(cls): + """Base directory for all EasyHAProxy files.""" + if cls._base_path is None: + if os.getenv("EASYHAPROXY_BASE_PATH"): + default = os.getenv("EASYHAPROXY_BASE_PATH") + elif os.getuid() == 0: + default = "/etc/easyhaproxy" + else: + default = str(Path.home() / "easyhaproxy") + cls._base_path = default + return cls._base_path + + @classmethod + def reset(cls): + """Reset cached base path to pick up environment variable changes.""" + cls._base_path = None + + @classproperty + def easyhaproxy_config(cls): + """Path to static configuration file.""" + return f"{cls.base_path}/static/config.yml" + + @classproperty + def haproxy_config(cls): + """Path to generated HAProxy configuration file.""" + return f"{cls.base_path}/haproxy/haproxy.cfg" + + @classproperty + def custom_config_folder(cls): + """Path to custom HAProxy config snippets directory.""" + return f"{cls.base_path}/haproxy/conf.d" + + @classproperty + def certs_certbot(cls): + """Path to Certbot/ACME certificates directory.""" + return f"{cls.base_path}/certs/certbot" + + @classproperty + def certs_haproxy(cls): + """Path to user-provided certificates directory.""" + return f"{cls.base_path}/certs/haproxy" \ No newline at end of file diff --git a/src/functions/container_env.py b/src/functions/container_env.py new file mode 100644 index 0000000..5628b1e --- /dev/null +++ b/src/functions/container_env.py @@ -0,0 +1,187 @@ +import os + +import requests + +from .functions import Functions +from .loggers import logger_certbot + + +class ContainerEnv: + @staticmethod + def read(yaml_config=None): + """ + Read configuration from environment variables, optionally merged with YAML config. + + Args: + yaml_config: Optional dict from YAML file (for static mode). YAML values take precedence. + + Returns: + Dict with configuration settings + """ + # Convert YAML config to environment variables first (if provided) + if yaml_config: + ContainerEnv._yaml_to_env(yaml_config) + + env_vars = { + "customerrors": True if os.getenv("HAPROXY_CUSTOMERRORS") == "true" else False, + "ssl_mode": os.getenv("EASYHAPROXY_SSL_MODE").lower() if os.getenv("EASYHAPROXY_SSL_MODE") else 'default' + } + + if os.getenv("HAPROXY_PASSWORD"): + env_vars["stats"] = { + "username": os.getenv("HAPROXY_USERNAME") if os.getenv("HAPROXY_USERNAME") else "admin", + "password": os.getenv("HAPROXY_PASSWORD"), + "port": os.getenv("HAPROXY_STATS_PORT") if os.getenv("HAPROXY_STATS_PORT") else "1936", + "cors_origin": os.getenv("HAPROXY_STATS_CORS_ORIGIN", ""), + } + + env_vars["lookup_label"] = os.getenv("EASYHAPROXY_LABEL_PREFIX") if os.getenv( + "EASYHAPROXY_LABEL_PREFIX") else "easyhaproxy" + + env_vars["logLevel"] = { + "easyhaproxy": os.getenv("EASYHAPROXY_LOG_LEVEL") if os.getenv( + "EASYHAPROXY_LOG_LEVEL") else Functions.DEBUG, + "haproxy": os.getenv("HAPROXY_LOG_LEVEL") if os.getenv("HAPROXY_LOG_LEVEL") else Functions.INFO, + "certbot": os.getenv("CERTBOT_LOG_LEVEL") if os.getenv("CERTBOT_LOG_LEVEL") else Functions.DEBUG, + } + + env_vars["certbot"] = { + "autoconfig": os.getenv("EASYHAPROXY_CERTBOT_AUTOCONFIG", ""), + "email": os.getenv("EASYHAPROXY_CERTBOT_EMAIL", ""), + "server": os.getenv("EASYHAPROXY_CERTBOT_SERVER", False), + "eab_kid": os.getenv("EASYHAPROXY_CERTBOT_EAB_KID", ""), + "eab_hmac_key": os.getenv("EASYHAPROXY_CERTBOT_EAB_HMAC_KEY", ""), + "retry_count": int(os.getenv("EASYHAPROXY_CERTBOT_RETRY_COUNT", 60)), + "preferred_challenges": os.getenv("EASYHAPROXY_CERTBOT_PREFERRED_CHALLENGES", "http"), + "manual_auth_hook": os.getenv("EASYHAPROXY_CERTBOT_MANUAL_AUTH_HOOK", False), + } + + if env_vars["certbot"]["autoconfig"] != "" and not env_vars["certbot"]["server"] and env_vars["certbot"]["email"] != "": + if env_vars["certbot"]["autoconfig"] == "letsencrypt": + env_vars["certbot"]["server"] = "https://acme-v02.api.letsencrypt.org/directory" + + if env_vars["certbot"]["autoconfig"] == "letsencrypt_test": + env_vars["certbot"]["server"] = "https://acme-staging-v02.api.letsencrypt.org/directory" + + if env_vars["certbot"]["autoconfig"] == "buypass": + env_vars["certbot"]["server"] = "https://api.buypass.com/acme/directory" + + if env_vars["certbot"]["autoconfig"] == "buypass_test": + env_vars["certbot"]["server"] = "https://api.test4.buypass.no/acme/directory" + + if env_vars["certbot"]["autoconfig"] == "sslcom_rca": + env_vars["certbot"]["server"] = "https://acme.ssl.com/sslcom-dv-rsa" + + if env_vars["certbot"]["autoconfig"] == "sslcom_ecc": + env_vars["certbot"]["server"] = "https://acme.ssl.com/sslcom-dv-ecc" + + if env_vars["certbot"]["autoconfig"] == "google": + env_vars["certbot"]["server"] = "https://dv.acme-v02.api.pki.goog/directory" + + if env_vars["certbot"]["autoconfig"] == "google_test": + env_vars["certbot"]["server"] = "https://dv.acme-v02.test-api.pki.goog/directory" + + if env_vars["certbot"]["autoconfig"] == "zerossl": + url = "https://api.zerossl.com/acme/eab-credentials-email" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + data = "email=" + env_vars["certbot"]["email"] + resp = requests.post(url, headers=headers, data=data).json() + + if resp["success"]: + env_vars["certbot"]["server"] = "https://acme.zerossl.com/v2/DV90" + env_vars["certbot"]["eab_kid"] = os.environ['EASYHAPROXY_CERTBOT_EAB_KID'] = resp["eab_kid"] + env_vars["certbot"]["eab_hmac_key"] = os.environ['EASYHAPROXY_CERTBOT_EAB_HMAC_KEY'] = resp["eab_hmac_key"] + else: + del os.environ["EASYHAPROXY_CERTBOT_EMAIL"] + logger_certbot.error("Could not obtain ZeroSSL credentials " + resp["error"]["type"]) + + os.environ['EASYHAPROXY_CERTBOT_SERVER'] = env_vars["certbot"]["server"] + + # Plugin configuration + env_vars["plugins"] = { + "abort_on_error": os.getenv("EASYHAPROXY_PLUGINS_ABORT_ON_ERROR", "false").lower() == "true", + "enabled": os.getenv("EASYHAPROXY_PLUGINS_ENABLED", "").split(",") if os.getenv("EASYHAPROXY_PLUGINS_ENABLED") else [], + "config": {} # Individual plugin configs from env vars + } + + # Parse individual plugin configs (e.g., EASYHAPROXY_PLUGIN_CLOUDFLARE_*) + for key, value in os.environ.items(): + if key.startswith("EASYHAPROXY_PLUGIN_"): + parts = key.split("_", 3) # ['EASYHAPROXY', 'PLUGIN', 'NAME', 'KEY'] + if len(parts) >= 4: + plugin_name = parts[2].lower() + config_key = "_".join(parts[3:]).lower() + env_vars["plugins"]["config"].setdefault(plugin_name, {}) + env_vars["plugins"]["config"][plugin_name][config_key] = value + + # Ingress status update configuration + env_vars["update_ingress_status"] = os.getenv("EASYHAPROXY_UPDATE_INGRESS_STATUS", "true").lower() == "true" + env_vars["deployment_mode"] = os.getenv("EASYHAPROXY_DEPLOYMENT_MODE", "auto") + env_vars["external_hostname"] = os.getenv("EASYHAPROXY_EXTERNAL_HOSTNAME", "") + env_vars["ingress_status_update_interval"] = int(os.getenv("EASYHAPROXY_STATUS_UPDATE_INTERVAL", "30")) + + return env_vars + + @staticmethod + def _yaml_to_env(yaml_config): + """Convert YAML configuration to environment variables""" + + # Convert customerrors + if 'customerrors' in yaml_config: + os.environ['HAPROXY_CUSTOMERRORS'] = 'true' if yaml_config['customerrors'] else 'false' + + # Convert ssl_mode + if 'ssl_mode' in yaml_config: + os.environ['EASYHAPROXY_SSL_MODE'] = str(yaml_config['ssl_mode']) + + # Convert stats + if 'stats' in yaml_config: + stats = yaml_config['stats'] + if 'username' in stats: + os.environ['HAPROXY_USERNAME'] = str(stats['username']) + if 'password' in stats: + os.environ['HAPROXY_PASSWORD'] = str(stats['password']) + if 'port' in stats: + os.environ['HAPROXY_STATS_PORT'] = str(stats['port']) + if 'cors_origin' in stats: + os.environ['HAPROXY_STATS_CORS_ORIGIN'] = str(stats['cors_origin']) + + # Convert logLevel + if 'logLevel' in yaml_config: + log_level = yaml_config['logLevel'] + for source, level in log_level.items(): + os.environ[source.upper() + '_LOG_LEVEL'] = str(level) + + # Convert certbot + if 'certbot' in yaml_config: + certbot = yaml_config['certbot'] + for config, value in certbot.items(): + os.environ['EASYHAPROXY_CERTBOT_' + config.upper()] = str(value) + + # Convert plugins + if 'plugins' in yaml_config: + plugins = yaml_config['plugins'] + + # Convert enabled list + if 'enabled' in plugins: + enabled_list = plugins['enabled'] if isinstance(plugins['enabled'], list) else [plugins['enabled']] + os.environ['EASYHAPROXY_PLUGINS_ENABLED'] = ','.join(enabled_list) + + # Convert abort_on_error + if 'abort_on_error' in plugins: + os.environ['EASYHAPROXY_PLUGINS_ABORT_ON_ERROR'] = 'true' if plugins['abort_on_error'] else 'false' + + # Convert plugin configs + if 'config' in plugins: + for plugin_name, plugin_config in plugins['config'].items(): + for config_key, config_value in plugin_config.items(): + # Convert to env var format: EASYHAPROXY_PLUGIN__ + env_key = f"EASYHAPROXY_PLUGIN_{plugin_name.upper()}_{config_key.upper()}" + + # Convert list values to comma-separated strings + if isinstance(config_value, list): + env_value = ','.join(str(v) for v in config_value) + else: + env_value = str(config_value) + + os.environ[env_key] = env_value \ No newline at end of file diff --git a/src/functions/filter.py b/src/functions/filter.py new file mode 100644 index 0000000..e2035ad --- /dev/null +++ b/src/functions/filter.py @@ -0,0 +1,31 @@ +import logging + + +class SingleLineNonEmptyFilter(logging.Filter): + """ + Logging filter that ensures messages are single-line and non-empty. + - Collapses newlines into spaces and strips surrounding whitespace. + - Drops the record if the resulting message is empty. + """ + def filter(self, record: logging.LogRecord) -> int: + try: + msg = record.getMessage() + except Exception: + # If formatting fails, drop the record + return 0 + + # Convert any non-string to string representation + if not isinstance(msg, str): + msg = str(msg) + + # Collapse multi-line to single line and trim + sanitized = " ".join(msg.splitlines()).strip() + + if sanitized == "": + return 0 + + # If we changed the message, update the record and clear args + if sanitized != record.getMessage(): + record.msg = sanitized + record.args = () + return 1 \ No newline at end of file diff --git a/src/functions/functions.py b/src/functions/functions.py new file mode 100644 index 0000000..5f23277 --- /dev/null +++ b/src/functions/functions.py @@ -0,0 +1,88 @@ +import logging +import os +import shlex +import subprocess +import sys +from typing import Final + +from .filter import SingleLineNonEmptyFilter + + +class Functions: + HAPROXY_LOG: Final[str] = "HAPROXY" + EASYHAPROXY_LOG: Final[str] = "EASYHAPROXY" + CERTBOT_LOG: Final[str] = "CERTBOT" + INIT_LOG: Final[str] = "INIT" + + TRACE: Final[str] = "TRACE" + DEBUG: Final[str] = "DEBUG" + INFO: Final[str] = "INFO" + WARN: Final[str] = "WARN" + ERROR: Final[str] = "ERROR" + FATAL: Final[str] = "FATAL" + + @staticmethod + def setup_log(source): + level = os.getenv(f"{source.name.upper()}_LOG_LEVEL", "").upper() + level_importance = { + Functions.TRACE: logging.DEBUG, + Functions.DEBUG: logging.DEBUG, + Functions.INFO: logging.INFO, + Functions.WARN: logging.WARNING, + Functions.ERROR: logging.ERROR, + Functions.FATAL: logging.FATAL + } + selected_level = level_importance[level] if level in level_importance else logging.INFO + + log_source_handler = logging.StreamHandler(sys.stdout) + log_source_formatter = logging.Formatter('%(name)s [%(asctime)s] %(levelname)s - %(message)s') + log_source_handler.setFormatter(log_source_formatter) + log_source_handler.addFilter(SingleLineNonEmptyFilter()) + source.setLevel(selected_level) + source.addHandler(log_source_handler) + return selected_level + + @staticmethod + def load(filename): + with open(filename) as content_file: + return content_file.read() + + @staticmethod + def save(filename, contents): + with open(filename, 'w') as file: + file.write(contents) + + @staticmethod + def run_bash(log_source, command, log_output=True, return_result=True): + if not isinstance(command, (list, tuple)): + command = shlex.split(command) + + try: + process = subprocess.Popen(command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + universal_newlines=True) + + output = [] + + while True: + line = process.stdout.readline().rstrip() + error_line = process.stderr.readline().rstrip() + output.append(line) if return_result else None + log_source.info(line) if log_output and len(line) > 0 else None + log_source.warning(error_line) if len(error_line) > 0 else None + return_code = process.poll() + if return_code is not None: + lines = [] + error_line = process.stderr.readline().rstrip() + for line in process.stdout.readlines(): + output.append(line.rstrip()) if return_result else None + lines.append(line.rstrip()) + log_source.info(lines) if log_output and len(lines) > 0 else None + log_source.warning(error_line) if len(error_line) > 0 else None + break + + return [return_code, output] + except Exception as e: + log_source.error(f"{e}") + return [-99, e] \ No newline at end of file diff --git a/src/functions/haproxy.py b/src/functions/haproxy.py new file mode 100644 index 0000000..09423fc --- /dev/null +++ b/src/functions/haproxy.py @@ -0,0 +1,152 @@ +import os +import shlex +import shutil +import subprocess +import sys +import time +from multiprocessing import Process +from typing import Final + +import psutil + +from .consts import Consts +from .functions import Functions +from .loggers import logger_haproxy + + +class DaemonizeHAProxy: + HAPROXY_START: Final[str] = "start" + HAPROXY_RELOAD: Final[str] = "reload" + + def __init__(self, custom_config_folder=None): + self.process = None + self.thread = None + self.sleep_secs = None + self.custom_config_folder = custom_config_folder if custom_config_folder is not None else Consts.custom_config_folder + + def haproxy(self, action): + error = self.__prepare(self.get_haproxy_command(action), action) + + if error or self.process is None: + logger_haproxy.fatal(f"Failed to start HAProxy ({action}). Exiting.") + sys.exit(1) + + self.thread = Process(target=self.__start, args=()) + self.thread.start() + + @staticmethod + def get_haproxy_bin() -> str: + return shutil.which('haproxy') or '/usr/sbin/haproxy' + + def get_haproxy_command(self, action, pid_file="/run/haproxy.pid"): + haproxy_bin = DaemonizeHAProxy.get_haproxy_bin() + custom_config_files = "" + if len(list(self.get_custom_config_files().keys())) != 0: + custom_config_files = f"-f {self.custom_config_folder}" + + if action == DaemonizeHAProxy.HAPROXY_START or not os.path.exists(pid_file): + return f"{haproxy_bin} -W -f {Consts.haproxy_config} {custom_config_files} -p {pid_file} -S /var/run/haproxy.sock" + else: + return_code, output = Functions().run_bash(logger_haproxy, f"cat {pid_file}", log_output=False) + pid = "".join(output).rstrip() + if psutil.pid_exists(int(pid)): + return f"{haproxy_bin} -W -f {Consts.haproxy_config} {custom_config_files} -p {pid_file} -x /var/run/haproxy.sock -sf {pid}" + else: + os.unlink(pid_file) + logger_haproxy.warning( + f"PID file {pid_file} does not exist. Restarting haproxy instead of reload." + ) + return self.get_haproxy_command(DaemonizeHAProxy.HAPROXY_START, pid_file) + + def __validate_config(self): + """Validate HAProxy configuration before starting.""" + validation_cmd = ["haproxy", "-c", "-f", Consts.haproxy_config] + + # Add custom config files if they exist + for config_file in self.get_custom_config_files().keys(): + validation_cmd.extend(["-f", config_file]) + + try: + result = subprocess.run( + validation_cmd, + capture_output=True, + text=True, + timeout=10 + ) + + if result.returncode != 0: + return result.stderr if result.stderr else result.stdout + return None + + except subprocess.TimeoutExpired: + return "HAProxy configuration validation timed out" + except Exception as e: + return f"Error validating configuration: {e}" + + def __prepare(self, command, action=None): + if not isinstance(command, (list, tuple)): + command = shlex.split(command) + + # Validate HAProxy config before starting (but not on reload - HAProxy validates itself during reload) + if action == DaemonizeHAProxy.HAPROXY_START: + validation_error = self.__validate_config() + if validation_error: + logger_haproxy.fatal(f"HAProxy configuration validation failed:\n{validation_error}") + return validation_error + + try: + logger_haproxy.debug(f"HAPROXY command: {command}") + self.process = subprocess.Popen(command, + shell=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=-1, + universal_newlines=True) + return None + + except Exception as e: + error_msg = f"Failed to start HAProxy process: {e}" + logger_haproxy.error(error_msg) + return error_msg + + def __start(self): + try: + with self.process.stdout: + for line in iter(self.process.stdout.readline, b''): + logger_haproxy.info(line.rstrip()) + + return_code = self.process.wait() + logger_haproxy.debug(f"Return code {return_code}") + + except Exception as e: + logger_haproxy.error(f"{e}") + + def is_alive(self): + return self.thread.is_alive() + + def kill(self): + self.process.kill() + self.thread.kill() + + def terminate(self): + self.process.terminate() + self.thread.terminate() + + def sleep(self): + if self.sleep_secs is None: + try: + self.sleep_secs = int(os.getenv("EASYHAPROXY_REFRESH_CONF", "10")) + except ValueError: + self.sleep_secs = 10 + + time.sleep(self.sleep_secs) + + def get_custom_config_files(self): + if not os.path.exists(self.custom_config_folder): + return {} + + files = {} + for file in os.listdir(self.custom_config_folder): + if file.endswith(".cfg"): + files[os.path.join(self.custom_config_folder, file)] = os.path.getmtime(os.path.join(self.custom_config_folder, file)) + return dict(sorted(files.items(), key=lambda t: t[0])) \ No newline at end of file diff --git a/src/functions/loggers.py b/src/functions/loggers.py new file mode 100644 index 0000000..ba309fb --- /dev/null +++ b/src/functions/loggers.py @@ -0,0 +1,13 @@ +import logging + +from .functions import Functions + +logger_init = logging.getLogger(Functions.INIT_LOG) +logger_haproxy = logging.getLogger(Functions.HAPROXY_LOG) +logger_easyhaproxy = logging.getLogger(Functions.EASYHAPROXY_LOG) +logger_certbot = logging.getLogger(Functions.CERTBOT_LOG) + +Functions.setup_log(logger_init) +Functions.setup_log(logger_haproxy) +Functions.setup_log(logger_easyhaproxy) +Functions.setup_log(logger_certbot) \ No newline at end of file diff --git a/src/main.py b/src/main.py deleted file mode 100644 index d8638a1..0000000 --- a/src/main.py +++ /dev/null @@ -1,94 +0,0 @@ -import os - -from deepdiff import DeepDiff - -from functions import ( - Certbot, - Consts, - DaemonizeHAProxy, - Functions, - logger_easyhaproxy, - logger_init, -) -from processor import ProcessorInterface - - -def start(): - processor_obj = ProcessorInterface.factory(os.getenv("EASYHAPROXY_DISCOVER")) - if processor_obj is None: - exit(1) - - os.makedirs(Consts.certs_certbot, exist_ok=True) - os.makedirs(Consts.certs_haproxy, exist_ok=True) - - processor_obj.save_config(Consts.haproxy_config) - processor_obj.save_certs(Consts.certs_haproxy) - certbot_certs_found = processor_obj.get_certbot_hosts() - logger_easyhaproxy.info(f'Found hosts: {", ".join(processor_obj.get_hosts())}') # Needs to run after save_config - logger_easyhaproxy.debug(f'Object Found: {processor_obj.get_parsed_object()}') - - old_haproxy = None - haproxy = DaemonizeHAProxy() - current_custom_config_files = haproxy.get_custom_config_files() - haproxy.haproxy(DaemonizeHAProxy.HAPROXY_START) - haproxy.sleep() - - certbot = Certbot(Consts.certs_certbot) - - # Check ACME environment readiness if Certbot is configured - if certbot.email != "": - is_ready, error_msg = Certbot.check_acme_environment_ready(certbot.email, certbot.acme_server) - if not is_ready: - logger_easyhaproxy.warning(f"ACME environment not ready: {error_msg}") - logger_easyhaproxy.warning("Certificate auto-renewal may fail. Verify ACME server configuration.") - else: - logger_easyhaproxy.info("ACME environment validated and ready") - - while True: - if old_haproxy is not None: - old_haproxy.kill() - old_haproxy = None - try: - old_parsed = processor_obj.get_parsed_object() - processor_obj.refresh() - if certbot.check_certificates(certbot_certs_found) or DeepDiff(old_parsed, processor_obj.get_parsed_object()) != {} or not haproxy.is_alive() or DeepDiff(current_custom_config_files, haproxy.get_custom_config_files()) != {}: - logger_easyhaproxy.info('New configuration found. Reloading...') - logger_easyhaproxy.debug(f'Object Found: {processor_obj.get_parsed_object()}') - processor_obj.save_config(Consts.haproxy_config) - processor_obj.save_certs(Consts.certs_haproxy) - certbot_certs_found = processor_obj.get_certbot_hosts() - logger_easyhaproxy.info(f'Found hosts: {", ".join(processor_obj.get_hosts())}') # Needs to after save_config - old_haproxy = haproxy - haproxy = DaemonizeHAProxy() - current_custom_config_files = haproxy.get_custom_config_files() - haproxy.haproxy(DaemonizeHAProxy.HAPROXY_RELOAD) - old_haproxy.terminate() - - except Exception as e: - logger_easyhaproxy.fatal(f"Err: {e}") - - logger_easyhaproxy.info('Heartbeat') - haproxy.sleep() - - -def main(): - Functions.run_bash(logger_init, '/usr/sbin/haproxy -v') - - logger_init.info(r".........................__.....................................") - logger_init.info(r"..___ ____ ________ __/ /_ ____ _____ _________ _ ____ __") - logger_init.info(r"./ _ \/ __ `/ ___/ / / / __ \/ __ `/ __ \/ ___/ __ \| |/_/ / / /") - logger_init.info(r"/ __/ /_/ (__ ) /_/ / / / / /_/ / /_/ / / / /_/ /> str: - """Return the unique plugin name""" - pass - - @property - @abstractmethod - def plugin_type(self) -> PluginType: - """Return the plugin type (GLOBAL or DOMAIN)""" - pass - - @abstractmethod - def configure(self, config: dict) -> None: - """ - Configure the plugin with settings from YAML/env/labels - - Args: - config: Dictionary with plugin-specific configuration - """ - pass - - @abstractmethod - def process(self, context: PluginContext) -> PluginResult: - """ - Process the plugin logic and return result - - Args: - context: PluginContext with all necessary data - - Returns: - PluginResult with HAProxy config snippets and/or modified data - """ - pass - - def initialize(self) -> InitializationResult: - """ - Initialize plugin resources. Default: no-op for backward compatibility - - Returns: - InitializationResult with resource requests - """ - return InitializationResult() - - -class PluginManager: - """Manages plugin loading, configuration, and execution""" - - def __init__(self, plugins_dir: str | None = None, abort_on_error: bool = False): - """ - Initialize the plugin manager - - Args: - plugins_dir: Directory containing plugin files (defaults to EASYHAPROXY_PLUGINS_DIR env var or /etc/easyhaproxy/plugins) - abort_on_error: If True, abort on plugin errors; if False, log and continue - """ - self.plugins_dir = plugins_dir or os.getenv( - "EASYHAPROXY_PLUGINS_DIR", - Consts.base_path + "/plugins" - ) - self.abort_on_error = abort_on_error - self.plugins: dict[str, PluginInterface] = {} - self.global_plugins: list[PluginInterface] = [] - self.domain_plugins: list[PluginInterface] = [] - self.logger = logger_easyhaproxy - - def load_plugins(self) -> None: - """ - Discover and load plugins from the plugins directory - Loads both builtin plugins and external plugins - """ - # Load builtin plugins first - builtin_dir = os.path.join(os.path.dirname(__file__), "builtin") - self._load_plugins_from_directory(builtin_dir, "builtin") - - # Load external plugins from /etc/haproxy/plugins - if os.path.exists(self.plugins_dir): - self._load_plugins_from_directory(self.plugins_dir, "external") - else: - self.logger.debug(f"Plugin directory {self.plugins_dir} does not exist, skipping external plugins") - - def _load_plugins_from_directory(self, directory: str, source: str) -> None: - """ - Load plugins from a specific directory - - Args: - directory: Path to directory containing plugins - source: Source identifier ("builtin" or "external") - """ - if not os.path.exists(directory): - return - - for filename in os.listdir(directory): - if filename.endswith(".py") and not filename.startswith("__"): - filepath = os.path.join(directory, filename) - module_name = f"plugins.{source}.{filename[:-3]}" - - try: - # Load module from file - spec = importlib.util.spec_from_file_location(module_name, filepath) - if spec and spec.loader: - module = importlib.util.module_from_spec(spec) - sys.modules[module_name] = module - spec.loader.exec_module(module) - - # Find plugin classes in module - for item_name in dir(module): - item = getattr(module, item_name) - if (isinstance(item, type) and - issubclass(item, PluginInterface) and - item is not PluginInterface): - # Instantiate plugin - plugin = item() - self.plugins[plugin.name] = plugin - - # Categorize by type - if plugin.plugin_type == PluginType.GLOBAL: - self.global_plugins.append(plugin) - elif plugin.plugin_type == PluginType.DOMAIN: - self.domain_plugins.append(plugin) - - self.logger.debug(f"Loaded {source} plugin: {plugin.name} ({plugin.plugin_type.value})") - - except Exception as e: - self._handle_error(f"Failed to load plugin from {filepath}: {str(e)}") - - def configure_plugins(self, plugins_config: dict) -> None: - """ - Configure all loaded plugins with their settings - - Args: - plugins_config: Plugin configuration from YAML/env - Format: {"plugin_name": {"key": "value"}, ...} - """ - for plugin_name, plugin in self.plugins.items(): - try: - # Get plugin-specific config - plugin_cfg = plugins_config.get(plugin_name, {}) - - # Also check "config" sub-key for env var configs - if "config" in plugins_config and plugin_name in plugins_config["config"]: - plugin_cfg.update(plugins_config["config"][plugin_name]) - - # Configure plugin - plugin.configure(plugin_cfg) - self.logger.debug(f"Configured plugin: {plugin_name} with config: {plugin_cfg}") - - except Exception as e: - self._handle_error(f"Failed to configure plugin '{plugin_name}': {str(e)}") - - def initialize_plugins(self) -> None: - """Initialize all plugins and process their resource requests""" - for plugin_name, plugin in self.plugins.items(): - try: - result = plugin.initialize() - self._process_initialization_result(plugin_name, result) - except Exception as e: - self._handle_error(f"Plugin '{plugin_name}' initialization failed: {e}") - - def _process_initialization_result(self, plugin_name: str, result: InitializationResult) -> None: - """ - Process plugin initialization requests - - Args: - plugin_name: Name of the plugin - result: InitializationResult with resource requests - """ - for resource in result.resources: - if resource.resource_type == "directory": - os.makedirs(resource.path, exist_ok=True) - self.logger.debug(f"Plugin '{plugin_name}' created directory: {resource.path}") - elif resource.resource_type == "file": - if resource.overwrite or not os.path.exists(resource.path): - with open(resource.path, 'w') as f: - f.write(resource.content or "") - self.logger.debug(f"Plugin '{plugin_name}' created file: {resource.path}") - - def execute_global_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]: - """ - Execute all global plugins - - Args: - context: PluginContext with execution data - enabled_list: Optional list of plugin names to execute. If None, execute all. - - Returns: - List of PluginResult from each plugin - """ - results = [] - - for plugin in self.global_plugins: - # Check if plugin is in enabled list (if provided) - if enabled_list is not None and plugin.name not in enabled_list: - continue - - try: - self.logger.debug(f"Executing global plugin: {plugin.name}") - result = plugin.process(context) - results.append(result) - - if result.metadata: - self.logger.debug(f"Plugin {plugin.name} metadata: {result.metadata}") - - except Exception as e: - self._handle_error(f"Global plugin '{plugin.name}' failed: {str(e)}") - - return results - - def execute_domain_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]: - """ - Execute all domain plugins for a specific domain - - Args: - context: PluginContext with domain-specific data - enabled_list: Optional list of plugin names to execute. If None, execute all. - - Returns: - List of PluginResult from each plugin - """ - results = [] - - for plugin in self.domain_plugins: - # Check if plugin is in enabled list (if provided) - if enabled_list is not None and plugin.name not in enabled_list: - continue - - try: - self.logger.debug(f"Executing domain plugin: {plugin.name} for domain: {context.domain}") - result = plugin.process(context) - results.append(result) - - if result.metadata: - self.logger.debug(f"Plugin {plugin.name} metadata: {result.metadata}") - - except Exception as e: - self._handle_error(f"Domain plugin '{plugin.name}' failed for domain '{context.domain}': {str(e)}") - - return results - - def _handle_error(self, message: str) -> None: - """ - Handle plugin errors according to abort_on_error setting - - Args: - message: Error message to log - """ - if self.abort_on_error: - self.logger.error(message) - raise RuntimeError(message) - else: - self.logger.warning(message) +__all__ = [ + "InitializationResult", + "PluginContext", + "PluginInterface", + "PluginManager", + "PluginResult", + "PluginType", + "ResourceRequest", +] \ No newline at end of file diff --git a/src/plugins/interface.py b/src/plugins/interface.py new file mode 100644 index 0000000..bed14e0 --- /dev/null +++ b/src/plugins/interface.py @@ -0,0 +1,51 @@ +from abc import ABC, abstractmethod + +from .types import InitializationResult, PluginContext, PluginResult, PluginType + + +class PluginInterface(ABC): + """Base class all plugins must inherit""" + + @property + @abstractmethod + def name(self) -> str: + """Return the unique plugin name""" + pass + + @property + @abstractmethod + def plugin_type(self) -> PluginType: + """Return the plugin type (GLOBAL or DOMAIN)""" + pass + + @abstractmethod + def configure(self, config: dict) -> None: + """ + Configure the plugin with settings from YAML/env/labels + + Args: + config: Dictionary with plugin-specific configuration + """ + pass + + @abstractmethod + def process(self, context: PluginContext) -> PluginResult: + """ + Process the plugin logic and return result + + Args: + context: PluginContext with all necessary data + + Returns: + PluginResult with HAProxy config snippets and/or modified data + """ + pass + + def initialize(self) -> InitializationResult: + """ + Initialize plugin resources. Default: no-op for backward compatibility + + Returns: + InitializationResult with resource requests + """ + return InitializationResult() \ No newline at end of file diff --git a/src/plugins/manager.py b/src/plugins/manager.py new file mode 100644 index 0000000..09315f9 --- /dev/null +++ b/src/plugins/manager.py @@ -0,0 +1,216 @@ +import importlib.util +import os +import sys + +from functions import Consts, logger_easyhaproxy + +from .interface import PluginInterface +from .types import PluginContext, PluginResult, PluginType + + +class PluginManager: + """Manages plugin loading, configuration, and execution""" + + def __init__(self, plugins_dir: str | None = None, abort_on_error: bool = False): + """ + Initialize the plugin manager + + Args: + plugins_dir: Directory containing plugin files (defaults to EASYHAPROXY_PLUGINS_DIR env var or /etc/easyhaproxy/plugins) + abort_on_error: If True, abort on plugin errors; if False, log and continue + """ + self.plugins_dir = plugins_dir or os.getenv( + "EASYHAPROXY_PLUGINS_DIR", + Consts.base_path + "/plugins" + ) + self.abort_on_error = abort_on_error + self.plugins: dict[str, PluginInterface] = {} + self.global_plugins: list[PluginInterface] = [] + self.domain_plugins: list[PluginInterface] = [] + self.logger = logger_easyhaproxy + + def load_plugins(self) -> None: + """ + Discover and load plugins from the plugins directory + Loads both builtin plugins and external plugins + """ + # Load builtin plugins first + builtin_dir = os.path.join(os.path.dirname(__file__), "builtin") + self._load_plugins_from_directory(builtin_dir, "builtin") + + # Load external plugins from /etc/haproxy/plugins + if os.path.exists(self.plugins_dir): + self._load_plugins_from_directory(self.plugins_dir, "external") + else: + self.logger.debug(f"Plugin directory {self.plugins_dir} does not exist, skipping external plugins") + + def _load_plugins_from_directory(self, directory: str, source: str) -> None: + """ + Load plugins from a specific directory + + Args: + directory: Path to directory containing plugins + source: Source identifier ("builtin" or "external") + """ + if not os.path.exists(directory): + return + + for filename in os.listdir(directory): + if filename.endswith(".py") and not filename.startswith("__"): + filepath = os.path.join(directory, filename) + module_name = f"plugins.{source}.{filename[:-3]}" + + try: + # Load module from file + spec = importlib.util.spec_from_file_location(module_name, filepath) + if spec and spec.loader: + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + + # Find plugin classes in module + for item_name in dir(module): + item = getattr(module, item_name) + if (isinstance(item, type) and + issubclass(item, PluginInterface) and + item is not PluginInterface): + # Instantiate plugin + plugin = item() + self.plugins[plugin.name] = plugin + + # Categorize by type + if plugin.plugin_type == PluginType.GLOBAL: + self.global_plugins.append(plugin) + elif plugin.plugin_type == PluginType.DOMAIN: + self.domain_plugins.append(plugin) + + self.logger.debug(f"Loaded {source} plugin: {plugin.name} ({plugin.plugin_type.value})") + + except Exception as e: + self._handle_error(f"Failed to load plugin from {filepath}: {str(e)}") + + def configure_plugins(self, plugins_config: dict) -> None: + """ + Configure all loaded plugins with their settings + + Args: + plugins_config: Plugin configuration from YAML/env + Format: {"plugin_name": {"key": "value"}, ...} + """ + for plugin_name, plugin in self.plugins.items(): + try: + # Get plugin-specific config + plugin_cfg = plugins_config.get(plugin_name, {}) + + # Also check "config" sub-key for env var configs + if "config" in plugins_config and plugin_name in plugins_config["config"]: + plugin_cfg.update(plugins_config["config"][plugin_name]) + + # Configure plugin + plugin.configure(plugin_cfg) + self.logger.debug(f"Configured plugin: {plugin_name} with config: {plugin_cfg}") + + except Exception as e: + self._handle_error(f"Failed to configure plugin '{plugin_name}': {str(e)}") + + def initialize_plugins(self) -> None: + """Initialize all plugins and process their resource requests""" + for plugin_name, plugin in self.plugins.items(): + try: + result = plugin.initialize() + self._process_initialization_result(plugin_name, result) + except Exception as e: + self._handle_error(f"Plugin '{plugin_name}' initialization failed: {e}") + + def _process_initialization_result(self, plugin_name: str, result) -> None: + """ + Process plugin initialization requests + + Args: + plugin_name: Name of the plugin + result: InitializationResult with resource requests + """ + for resource in result.resources: + if resource.resource_type == "directory": + os.makedirs(resource.path, exist_ok=True) + self.logger.debug(f"Plugin '{plugin_name}' created directory: {resource.path}") + elif resource.resource_type == "file": + if resource.overwrite or not os.path.exists(resource.path): + with open(resource.path, 'w') as f: + f.write(resource.content or "") + self.logger.debug(f"Plugin '{plugin_name}' created file: {resource.path}") + + def execute_global_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]: + """ + Execute all global plugins + + Args: + context: PluginContext with execution data + enabled_list: Optional list of plugin names to execute. If None, execute all. + + Returns: + List of PluginResult from each plugin + """ + results = [] + + for plugin in self.global_plugins: + # Check if plugin is in enabled list (if provided) + if enabled_list is not None and plugin.name not in enabled_list: + continue + + try: + self.logger.debug(f"Executing global plugin: {plugin.name}") + result = plugin.process(context) + results.append(result) + + if result.metadata: + self.logger.debug(f"Plugin {plugin.name} metadata: {result.metadata}") + + except Exception as e: + self._handle_error(f"Global plugin '{plugin.name}' failed: {str(e)}") + + return results + + def execute_domain_plugins(self, context: PluginContext, enabled_list: list[str] | None = None) -> list[PluginResult]: + """ + Execute all domain plugins for a specific domain + + Args: + context: PluginContext with domain-specific data + enabled_list: Optional list of plugin names to execute. If None, execute all. + + Returns: + List of PluginResult from each plugin + """ + results = [] + + for plugin in self.domain_plugins: + # Check if plugin is in enabled list (if provided) + if enabled_list is not None and plugin.name not in enabled_list: + continue + + try: + self.logger.debug(f"Executing domain plugin: {plugin.name} for domain: {context.domain}") + result = plugin.process(context) + results.append(result) + + if result.metadata: + self.logger.debug(f"Plugin {plugin.name} metadata: {result.metadata}") + + except Exception as e: + self._handle_error(f"Domain plugin '{plugin.name}' failed for domain '{context.domain}': {str(e)}") + + return results + + def _handle_error(self, message: str) -> None: + """ + Handle plugin errors according to abort_on_error setting + + Args: + message: Error message to log + """ + if self.abort_on_error: + self.logger.error(message) + raise RuntimeError(message) + else: + self.logger.warning(message) \ No newline at end of file diff --git a/src/plugins/types.py b/src/plugins/types.py new file mode 100644 index 0000000..9526c60 --- /dev/null +++ b/src/plugins/types.py @@ -0,0 +1,46 @@ +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + + +class PluginType(Enum): + """Plugin execution types""" + GLOBAL = "global" # Execute once per discovery cycle + DOMAIN = "domain" # Execute per domain/host + + +@dataclass +class PluginContext: + """Container for all plugin execution data""" + parsed_object: dict # {IP: labels} from discovery + easymapping: list # Current HAProxy mapping structure + container_env: dict # Environment configuration + domain: str | None = None # Domain name (for DOMAIN plugins) + port: str | None = None # Port (for DOMAIN plugins) + host_config: dict | None = None # Domain-specific config + + +@dataclass +class ResourceRequest: + """Request for file system resources""" + resource_type: str # "directory" or "file" + path: str + content: str | None = None + overwrite: bool = False + + +@dataclass +class InitializationResult: + """Plugin initialization result with resource requests""" + resources: list[ResourceRequest] = field(default_factory=list) + metadata: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class PluginResult: + """Plugin execution result""" + haproxy_config: str = "" # HAProxy config snippet to inject + modified_easymapping: list | None = None # Modified easymapping structure + metadata: dict[str, Any] = field(default_factory=dict) # Plugin metadata for logging + global_configs: list[str] = field(default_factory=list) # HAProxy global-level configs + defaults_configs: list[str] = field(default_factory=list) # HAProxy defaults-level configs \ No newline at end of file diff --git a/src/processor/__init__.py b/src/processor/__init__.py index f0a00bf..aba7414 100644 --- a/src/processor/__init__.py +++ b/src/processor/__init__.py @@ -1,758 +1,7 @@ -import base64 -import socket -from typing import Final +from .docker import Docker +from .interface import ProcessorInterface +from .kubernetes import Kubernetes +from .static import Static +from .swarm import Swarm -import docker -import yaml -from kubernetes import client, config -from kubernetes.client.rest import ApiException - -from easymapping import HaproxyConfigGenerator -from functions import Consts, ContainerEnv, Functions, logger_easyhaproxy - - -class ProcessorInterface: - STATIC: Final[str] = "static" - DOCKER: Final[str] = "docker" - SWARM: Final[str] = "swarm" - KUBERNETES: Final[str] = "kubernetes" - - static_file = Consts.easyhaproxy_config - - def __init__(self, filename=None): - self.certbot_hosts = None - self.parsed_object = None - self.cfg = None - self.hosts = None - self.cfg = None - self.certbot_hosts = None - self.hosts = None - self.filename = filename - self.label = ContainerEnv.read()['lookup_label'] - self.refresh() - - @staticmethod - def factory(mode): - if mode == ProcessorInterface.STATIC: - return Static(ProcessorInterface.static_file) - elif mode == ProcessorInterface.DOCKER: - return Docker() - elif mode == ProcessorInterface.SWARM: - return Swarm() - elif mode == ProcessorInterface.KUBERNETES: - return Kubernetes() - else: - logger_easyhaproxy.fatal(f"Expected mode to be 'static', 'docker', 'swarm' or 'kubernetes'. I got '{mode}'") - return None - - def refresh(self): - self.certbot_hosts = None - self.parsed_object = None - self.cfg = None - self.hosts = None - self.inspect_network() - self.parse() - - def inspect_network(self): - # Abstract - pass - - def parse(self): - self.cfg = HaproxyConfigGenerator(ContainerEnv.read()) - - def get_certbot_hosts(self): - return self.certbot_hosts - - def get_hosts(self): - return self.hosts - - def get_parsed_object(self): - return self.parsed_object - - def get_certs(self, key=None): - if key is None: - return self.cfg.certs - else: - return None if key not in self.cfg.certs else self.cfg.certs[key] - - def get_haproxy_conf(self): - conf = self.cfg.generate(self.parsed_object) - self.certbot_hosts = self.cfg.certbot_hosts - self.hosts = self.cfg.serving_hosts - return conf - - def save_config(self, filename): - Functions.save(filename, self.get_haproxy_conf()) - - def save_certs(self, path): - for cert in self.get_certs(): - Functions.save(f"{path}/{cert}", self.get_certs(cert)) - - -class Static(ProcessorInterface): - def __init__(self, filename=None): - self.parsed_object = None - self.static_content = None - self.static_content = None - self.cfg = None - super().__init__(filename) - - def inspect_network(self): - """Load YAML and convert containers to Docker-style container metadata""" - # Load YAML - self.static_content = yaml.load(Functions.load(self.filename), Loader=yaml.FullLoader) - - # Convert containers to label format - self.parsed_object = self._convert_yaml_to_labels() - - def _convert_yaml_to_labels(self): - """ - Convert static YAML containers to Docker label format. - Returns: {IP: {labels}} structure that parse() can process - """ - import json - - container_metadata = {} - - # Get global plugin configuration - global_plugins = self.static_content.get("plugins", {}) - global_enabled = global_plugins.get("enabled", []) - global_plugin_config = global_plugins.get("config", {}) - - for host_port, config in self.static_content.get("containers", {}).items(): - # Parse hostname:port from key - if ":" in host_port: - hostname, port = host_port.rsplit(":", 1) - else: - hostname = host_port - port = "80" - - # Create definition: hostname_port (e.g., host1_com_br_80) - definition = hostname.replace(".", "_") + f"_{port}" - - # Handle redirect-only entries (no backend) - if "redirect" in config and "ip" not in config: - # Create metadata with redirect but mark as redirect-only to skip backend creation - fake_ip = f"redirect-{hostname}-{port}" - if fake_ip not in container_metadata: - container_metadata[fake_ip] = {} - - container_metadata[fake_ip].update({ - f"easyhaproxy.{definition}.host": hostname, - f"easyhaproxy.{definition}.port": port, - f"easyhaproxy.{definition}.redirect": json.dumps({hostname: config["redirect"]}), - f"easyhaproxy.{definition}.redirect_only": "true", # Marker to skip backend - }) - continue - - # Get IPs/containers - ip_list = config.get("ip", [hostname]) - - # Process each container/IP - for container_spec in ip_list: - # Parse container:localport - if ":" in container_spec: - container_addr, localport = container_spec.rsplit(":", 1) - else: - container_addr = container_spec - localport = "80" - - # Use container address as IP (could be IP, DNS, or container name) - ip = container_addr - - # Build labels dict - labels = { - f"easyhaproxy.{definition}.host": hostname, - f"easyhaproxy.{definition}.port": port, - f"easyhaproxy.{definition}.localport": localport, - } - - # Add optional settings - for key in ["mode", "certbot", "redirect_ssl", "ssl", "balance", "proto", "ssl-check", "clone_to_ssl"]: - if key in config: - value = config[key] - # Convert boolean to string - if isinstance(value, bool): - value = "true" if value else "false" - labels[f"easyhaproxy.{definition}.{key}"] = str(value) - - # Handle plugins - host_plugins = config.get("plugins", global_enabled) - if host_plugins: - # Convert list to comma-separated string if needed - if isinstance(host_plugins, list): - plugins_str = ",".join(host_plugins) - else: - plugins_str = host_plugins - labels[f"easyhaproxy.{definition}.plugins"] = plugins_str - - # Process plugin configurations - host_plugin_config = config.get("plugin", {}) - - # Parse plugins list - plugins_list = host_plugins if isinstance(host_plugins, list) else [p.strip() for p in host_plugins.split(",")] - - for plugin_name in plugins_list: - # Merge global and host-specific config - merged_config = {} - if plugin_name in global_plugin_config: - merged_config.update(global_plugin_config[plugin_name]) - if plugin_name in host_plugin_config: - merged_config.update(host_plugin_config[plugin_name]) - - # Convert plugin config to labels - for config_key, config_value in merged_config.items(): - label_key = f"easyhaproxy.{definition}.plugin.{plugin_name}.{config_key}" - - # Convert list values to comma-separated strings - if isinstance(config_value, list): - label_value = ",".join(str(v) for v in config_value) - else: - label_value = str(config_value) - - labels[label_key] = label_value - - # Initialize container entry if it doesn't exist - if ip not in container_metadata: - container_metadata[ip] = {} - - # Merge labels instead of overwriting - container_metadata[ip].update(labels) - - return container_metadata - - def parse(self): - """Create HaproxyConfigGenerator with YAML config merged into env vars""" - self.cfg = HaproxyConfigGenerator(ContainerEnv.read(self.static_content)) - - -class Docker(ProcessorInterface): - def __init__(self, filename=None): - self.parsed_object = None - self.client = docker.from_env() - super().__init__() - - def inspect_network(self): - try: - ha_proxy_network_name = next( - iter(self.client.containers.get(socket.gethostname()).attrs["NetworkSettings"]["Networks"])) - except Exception: - # HAProxy is not running in a container, get first container network - if len(self.client.containers.list()) == 0: - return - ha_proxy_network_name = next(iter( - self.client.containers.get(self.client.containers.list()[0].name).attrs["NetworkSettings"]["Networks"])) - - ha_proxy_network = self.client.networks.get(ha_proxy_network_name) - - self.parsed_object = {} - for container in self.client.containers.list(): - # Issue 32 - Docker container cannot connect to containers in different network. - if ha_proxy_network_name not in container.attrs["NetworkSettings"]["Networks"].keys(): - ha_proxy_network.connect(container.name) - container = self.client.containers.get(container.name) # refresh object - - ip_address = container.attrs["NetworkSettings"]["Networks"][ha_proxy_network_name]["IPAddress"] - self.parsed_object[ip_address] = container.labels - - -class Swarm(ProcessorInterface): - def __init__(self, filename=None): - self.parsed_object = None - self.client = docker.from_env() - super().__init__() - - def inspect_network(self): - ha_proxy_service_name = self.client.containers.get(socket.gethostname()).name.split('.')[0] - ha_proxy_network_id = None - swarm_ingress_id = None - - # Get the HAProxy network and the ingress network - for endpoint in self.client.services.get(ha_proxy_service_name).attrs['Endpoint']["VirtualIPs"]: - network_name = self.client.networks.get(endpoint["NetworkID"]).name - if swarm_ingress_id is None and network_name == 'ingress': - swarm_ingress_id = endpoint["NetworkID"] - if ha_proxy_network_id is None and network_name != 'ingress': - ha_proxy_network_id = endpoint["NetworkID"] - if ha_proxy_network_id is not None and swarm_ingress_id is not None: - break - - # Check if the service is attached to the HAProxy network - self.parsed_object = {} - for service in self.client.services.list(): - if not any(self.label in key for key in service.attrs["Spec"]["Labels"]): - continue - - ip_address = None - network_list = [] - for endpoint in service.attrs["Endpoint"]["VirtualIPs"]: - if ha_proxy_network_id == endpoint["NetworkID"]: - ip_address = endpoint["Addr"].split("/")[0] - break - elif swarm_ingress_id != endpoint["NetworkID"]: - network_list.append(endpoint["NetworkID"]) - - # Attach the service to the HAProxy network - if ip_address is None: - network_list.append(ha_proxy_network_id) - service.update(networks = network_list) - continue # skip to the next service to give time to update the network - - self.parsed_object[ip_address] = service.attrs["Spec"]["Labels"] - - -class Kubernetes(ProcessorInterface): - def __init__(self, filename=None, api_instance=None, v1=None): - self.parsed_object = None - - # Only load config if API clients are not provided (allows dependency injection for testing) - if api_instance is None or v1 is None: - config.load_incluster_config() - config.verify_ssl = False - - # Use injected clients or create new ones (dependency injection pattern) - self.api_instance = api_instance or client.CoreV1Api() - self.v1 = v1 or client.NetworkingV1Api() - self.cert_cache = {} - self.deployment_mode_cache = None - self.ingress_addresses_cache = None - self.addresses_cache_time = 0 - super().__init__() - - def _detect_deployment_mode(self): - """ - Detect the deployment mode (daemonset, nodeport, clusterip). - Returns: tuple (mode: str, service: V1Service or None) - """ - import os - - # Return cached if available - if self.deployment_mode_cache: - return self.deployment_mode_cache - - env_config = ContainerEnv.read() - - # Check for manual override - if env_config['deployment_mode'] != 'auto': - logger_easyhaproxy.info(f"Using manual deployment mode: {env_config['deployment_mode']}") - service = self._get_easyhaproxy_service() if env_config['deployment_mode'] in ['nodeport', 'clusterip'] else None - self.deployment_mode_cache = (env_config['deployment_mode'], service) - return self.deployment_mode_cache - - try: - # Get current pod name from hostname - pod_name = socket.gethostname() - namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') - - # Read current pod - pod = self.api_instance.read_namespaced_pod(pod_name, namespace) - - # Check owner references to determine if DaemonSet or Deployment - if pod.metadata.owner_references: - owner_kind = pod.metadata.owner_references[0].kind - - if owner_kind == 'DaemonSet': - logger_easyhaproxy.info("Detected deployment mode: daemonset") - self.deployment_mode_cache = ('daemonset', None) - return self.deployment_mode_cache - elif owner_kind in ['ReplicaSet', 'Deployment']: - # Check if Service exists - service = self._get_easyhaproxy_service() - if service: - if service.spec.type == 'NodePort': - logger_easyhaproxy.info("Detected deployment mode: nodeport") - self.deployment_mode_cache = ('nodeport', service) - return self.deployment_mode_cache - else: - logger_easyhaproxy.info("Detected deployment mode: clusterip") - self.deployment_mode_cache = ('clusterip', service) - return self.deployment_mode_cache - except Exception as e: - logger_easyhaproxy.warn(f"Failed to detect deployment mode: {e}, defaulting to daemonset") - - self.deployment_mode_cache = ('daemonset', None) - return self.deployment_mode_cache - - def _get_easyhaproxy_service(self): - """Get the EasyHAProxy service if it exists.""" - import os - - try: - namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') - # Try common service names - service_names = ['easyhaproxy', 'ingress-easyhaproxy'] - - for service_name in service_names: - try: - service = self.api_instance.read_namespaced_service(service_name, namespace) - return service - except Exception: - continue - return None - except Exception as e: - logger_easyhaproxy.warn(f"Failed to get EasyHAProxy service: {e}") - return None - - def _get_ingress_addresses(self, mode, service): - """ - Get IP addresses or hostnames to report in ingress status. - - Args: - mode: Deployment mode (daemonset, nodeport, clusterip) - service: V1Service object (for nodeport/clusterip modes) - - Returns: - List of dicts: [{"ip": "..."}, {"hostname": "..."}] - """ - import os - import time - - env_config = ContainerEnv.read() - cache_ttl = env_config.get('ingress_status_update_interval', 30) - - # Return cached if still valid - if self.ingress_addresses_cache and (time.time() - self.addresses_cache_time) < cache_ttl: - return self.ingress_addresses_cache - - addresses = [] - - try: - if mode == 'daemonset': - # Get nodes where DaemonSet pods are running - namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') - label_selector = "app.kubernetes.io/name=easyhaproxy" - - pods = self.api_instance.list_namespaced_pod(namespace, label_selector=label_selector) - node_names = set(pod.spec.node_name for pod in pods.items if pod.spec.node_name) - - # Get external IPs from these nodes - for node_name in node_names: - node = self.api_instance.read_node(node_name) - for addr in node.status.addresses: - if addr.type == 'ExternalIP': - addresses.append({"ip": addr.address}) - break - else: - # Fallback to InternalIP if no ExternalIP - for addr in node.status.addresses: - if addr.type == 'InternalIP': - addresses.append({"ip": addr.address}) - break - - elif mode == 'nodeport': - # Get all node IPs (traffic can reach any node via NodePort) - nodes = self.api_instance.list_node() - for node in nodes.items: - for addr in node.status.addresses: - if addr.type == 'ExternalIP': - addresses.append({"ip": addr.address}) - break - else: - # Fallback to InternalIP - for addr in node.status.addresses: - if addr.type == 'InternalIP': - addresses.append({"ip": addr.address}) - break - - elif mode == 'clusterip': - # Check if LoadBalancer status is available - if service and service.status and service.status.load_balancer: - lb_ingress = service.status.load_balancer.ingress or [] - for ing in lb_ingress: - if ing.ip: - addresses.append({"ip": ing.ip}) - if ing.hostname: - addresses.append({"hostname": ing.hostname}) - - # If no LoadBalancer, check for external hostname override - if not addresses and env_config['external_hostname']: - addresses.append({"hostname": env_config['external_hostname']}) - - # Fallback to ClusterIP - if not addresses and service: - addresses.append({"ip": service.spec.cluster_ip}) - - except Exception as e: - logger_easyhaproxy.warn(f"Failed to get ingress addresses: {e}") - - # Cache the result - self.ingress_addresses_cache = addresses - self.addresses_cache_time = time.time() - - return addresses - - def _update_ingress_status(self, ingress, addresses): - """ - Update the status of an ingress resource. - - Args: - ingress: V1Ingress object - addresses: List of address dicts [{"ip": "..."}, {"hostname": "..."}] - """ - if not addresses: - return - - try: - # Create status patch - status_body = { - "status": { - "loadBalancer": { - "ingress": addresses - } - } - } - - # Update status using patch (not replace) - self.v1.patch_namespaced_ingress_status( - name=ingress.metadata.name, - namespace=ingress.metadata.namespace, - body=status_body, - field_manager="easyhaproxy" - ) - - logger_easyhaproxy.debug( - f"Updated ingress {ingress.metadata.namespace}/{ingress.metadata.name} " - f"status with {len(addresses)} address(es)" - ) - - except Exception as e: - logger_easyhaproxy.warn( - f"Failed to update status for ingress " - f"{ingress.metadata.namespace}/{ingress.metadata.name}: {e}" - ) - - def _check_annotation(self, annotations, key, default=None): - if key not in annotations: - return default - return annotations[key] - - def inspect_network(self): - - ret = self.v1.list_ingress_for_all_namespaces(watch=False) - - # Detect deployment mode once per cycle for ingress status updates - env_config = ContainerEnv.read() - if env_config['update_ingress_status']: - deployment_mode, service = self._detect_deployment_mode() - ingress_addresses = self._get_ingress_addresses(deployment_mode, service) - else: - ingress_addresses = [] - - self.parsed_object = {} - for ingress in ret.items: - # Support both new spec.ingressClassName and deprecated annotation for backward compatibility - ingress_class = None - is_match = False - - # Check new spec.ingressClassName first (preferred) - if hasattr(ingress.spec, 'ingress_class_name') and ingress.spec.ingress_class_name is not None: - ingress_class = ingress.spec.ingress_class_name - # Modern spec uses 'easyhaproxy' - is_match = (ingress_class == "easyhaproxy") - # Fall back to deprecated annotation - elif ingress.metadata.annotations and 'kubernetes.io/ingress.class' in ingress.metadata.annotations: - ingress_class = ingress.metadata.annotations['kubernetes.io/ingress.class'] - # Deprecated annotation uses 'easyhaproxy-ingress' for backward compatibility - is_match = (ingress_class == "easyhaproxy-ingress") - - # Skip if no ingress class is defined or it doesn't match - if not is_match: - continue - - ssl_hosts = [] - - certbot = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.certbot") - redirect_ssl = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.redirect_ssl") - redirect = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.redirect") - mode = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.mode") - listen_port = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.listen_port", 80) - plugins = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.plugins") - - # Extract plugin-specific configurations - plugin_annotations = {} - for annotation_key, annotation_value in ingress.metadata.annotations.items(): - if annotation_key.startswith("easyhaproxy.plugin."): - plugin_annotations[annotation_key] = annotation_value - - # Get ingress name for logging - ingress_name = f"{ingress.metadata.namespace}/{ingress.metadata.name}" - - # Generic k8s_secret annotation processing - # Pattern: easyhaproxy.plugin.X.k8s_secret.KEY: "secret_name" or "secret_name/key_name" - # Result: easyhaproxy.plugin.X.KEY: "" - k8s_secret_annotations = {} - for annotation_key, secret_value in list(plugin_annotations.items()): - # Check if this annotation contains k8s_secret pattern - if ".k8s_secret." in annotation_key: - try: - # Parse the annotation key - # Example: "easyhaproxy.plugin.jwt_validator.k8s_secret.pubkey" -> "pubkey" - parts = annotation_key.split(".k8s_secret.") - if len(parts) != 2: - logger_easyhaproxy.warn( - f"Ingress {ingress_name} - Malformed k8s_secret annotation: {annotation_key}" - ) - continue - - prefix = parts[0] # "easyhaproxy.plugin.jwt_validator" - config_key = parts[1] # "pubkey" - target_annotation = f"{prefix}.{config_key}" # "easyhaproxy.plugin.jwt_validator.pubkey" - - # Parse secret_value: can be "secret_name" or "secret_name/key_name" - if "/" in secret_value: - secret_name, explicit_key_name = secret_value.split("/", 1) - use_explicit_key = True - else: - secret_name = secret_value - explicit_key_name = None - use_explicit_key = False - - # Read the secret - secret = self.api_instance.read_namespaced_secret( - secret_name, - ingress.metadata.namespace - ) - - # Try to find the key in the secret data - secret_data = None - tried_keys = [] - - if use_explicit_key: - # User specified exact key name - only try that one - tried_keys = [explicit_key_name] - if explicit_key_name in secret.data: - secret_data = secret.data[explicit_key_name] - logger_easyhaproxy.debug( - f"Ingress {ingress_name} - Found explicit secret key '{explicit_key_name}' " - f"in secret '{secret_name}'" - ) - else: - # No explicit key - try config_key and common variations - tried_keys = [config_key] - if config_key in secret.data: - secret_data = secret.data[config_key] - else: - # Try common variations for the requested key - variations = [] - if config_key == "pubkey": - variations = ["public-key", "jwt.pub", "tls.crt"] - elif config_key == "password": - variations = ["pass", "pwd"] - elif config_key == "api_key": - variations = ["apikey", "api-key", "key"] - - for variation in variations: - tried_keys.append(variation) - if variation in secret.data: - secret_data = secret.data[variation] - logger_easyhaproxy.debug( - f"Ingress {ingress_name} - Found secret key '{variation}' " - f"for requested key '{config_key}'" - ) - break - - if secret_data: - # Decode from base64 (Kubernetes secrets are base64-encoded) - # Then re-encode to base64 for plugin (plugin expects base64-encoded) - decoded = base64.b64decode(secret_data).decode('ascii') - reencoded = base64.b64encode(decoded.encode('ascii')).decode('ascii') - - # Store the processed annotation - k8s_secret_annotations[target_annotation] = reencoded - - logger_easyhaproxy.info( - f"Ingress {ingress_name} - Loaded '{config_key}' from secret " - f"'{secret_name}' for annotation '{target_annotation}'" - ) - else: - logger_easyhaproxy.warn( - f"Ingress {ingress_name} - Secret '{secret_name}' found but " - f"no matching key (tried: {', '.join(tried_keys)})" - ) - - except Exception as e: - logger_easyhaproxy.warn( - f"Ingress {ingress_name} - Failed to process k8s_secret annotation " - f"'{annotation_key}' with value '{secret_value}': {e}" - ) - - # Merge k8s_secret annotations into plugin_annotations - # k8s_secret annotations will NOT override existing explicit annotations (lower priority) - for key, value in k8s_secret_annotations.items(): - if key not in plugin_annotations: - plugin_annotations[key] = value - else: - logger_easyhaproxy.debug( - f"Ingress {ingress_name} - Skipping k8s_secret annotation '{key}' " - f"because explicit annotation already exists" - ) - - data = {"creation_timestamp": ingress.metadata.creation_timestamp.strftime("%x %X"), - "resource_version": ingress.metadata.resource_version, "namespace": ingress.metadata.namespace} - - if ingress.spec.tls is not None: - for tls in ingress.spec.tls: - try: - secret = self.api_instance.read_namespaced_secret(tls.secret_name, ingress.metadata.namespace) - if "tls.crt" not in secret.data or "tls.key" not in secret.data: - continue - - if tls.secret_name not in self.cert_cache or self.cert_cache[tls.secret_name] != secret.data: - self.cert_cache[tls.secret_name] = secret.data - Functions.save( - f"{Consts.certs_haproxy}/{tls.secret_name}.pem", - base64.b64decode(secret.data["tls.crt"]).decode('ascii') + "\n" + base64.b64decode( - secret.data["tls.key"]).decode('ascii') - ) - - ssl_hosts.extend(tls.hosts) - except Exception as e: - logger_easyhaproxy.warn(f"Ingress {ingress_name} - Get secret failed: '{e}'") - - logger_easyhaproxy.debug(f"Ingress {ingress_name} - SSL Hosts found '{ssl_hosts}'") - - for rule in ingress.spec.rules: - rule_data = {} - port_number = rule.http.paths[0].backend.service.port.number - definition = f"easyhaproxy.{rule.host.replace('.', '-')}_{port_number}" - rule_data[f"{definition}.host"] = rule.host - rule_data[f"{definition}.port"] = listen_port - rule_data[f"{definition}.localport"] = port_number - if rule.host in ssl_hosts: - rule_data[f"{definition}.clone_to_ssl"] = 'true' - if redirect_ssl is not None: - rule_data[f"{definition}.redirect_ssl"] = redirect_ssl - if certbot is not None: - rule_data[f"{definition}.certbot"] = certbot - if redirect is not None: - rule_data[f"{definition}.redirect"] = redirect - if mode is not None: - rule_data[f"{definition}.mode"] = mode - rule_data[f"{definition}.balance"] = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.balance", "roundrobin") - - # Add plugin configuration - if plugins is not None: - rule_data[f"{definition}.plugins"] = plugins - - # Add plugin-specific configurations - for plugin_key, plugin_value in plugin_annotations.items(): - # Convert easyhaproxy.plugin.X.Y to easyhaproxy.{definition}.plugin.X.Y - plugin_config_key = plugin_key.replace("easyhaproxy.plugin.", f"{definition}.plugin.") - rule_data[plugin_config_key] = plugin_value - - service_name = rule.http.paths[0].backend.service.name - try: - api_response = self.api_instance.read_namespaced_service(service_name, ingress.metadata.namespace) - cluster_ip = api_response.spec.cluster_ip - except ApiException as e: - cluster_ip = None - logger_easyhaproxy.warn(f"Ingress {ingress_name} - Service {service_name} - Failed: '{e}'") - - if cluster_ip is not None: - if cluster_ip not in self.parsed_object.keys(): - self.parsed_object[cluster_ip] = data - self.parsed_object[cluster_ip].update(rule_data) - - # Update ingress status if enabled - if env_config['update_ingress_status'] and ingress_addresses: - self._update_ingress_status(ingress, ingress_addresses) +__all__ = ["ProcessorInterface", "Static", "Docker", "Swarm", "Kubernetes"] \ No newline at end of file diff --git a/src/processor/docker.py b/src/processor/docker.py new file mode 100644 index 0000000..3911a4d --- /dev/null +++ b/src/processor/docker.py @@ -0,0 +1,35 @@ +import socket + +import docker + +from .interface import ProcessorInterface + + +class Docker(ProcessorInterface): + def __init__(self, filename=None): + self.parsed_object = None + self.client = docker.from_env() + super().__init__() + + def inspect_network(self): + try: + ha_proxy_network_name = next( + iter(self.client.containers.get(socket.gethostname()).attrs["NetworkSettings"]["Networks"])) + except Exception: + # HAProxy is not running in a container, get first container network + if len(self.client.containers.list()) == 0: + return + ha_proxy_network_name = next(iter( + self.client.containers.get(self.client.containers.list()[0].name).attrs["NetworkSettings"]["Networks"])) + + ha_proxy_network = self.client.networks.get(ha_proxy_network_name) + + self.parsed_object = {} + for container in self.client.containers.list(): + # Issue 32 - Docker container cannot connect to containers in different network. + if ha_proxy_network_name not in container.attrs["NetworkSettings"]["Networks"].keys(): + ha_proxy_network.connect(container.name) + container = self.client.containers.get(container.name) # refresh object + + ip_address = container.attrs["NetworkSettings"]["Networks"][ha_proxy_network_name]["IPAddress"] + self.parsed_object[ip_address] = container.labels \ No newline at end of file diff --git a/src/processor/interface.py b/src/processor/interface.py new file mode 100644 index 0000000..438de16 --- /dev/null +++ b/src/processor/interface.py @@ -0,0 +1,87 @@ +from typing import Final + +from easymapping import HaproxyConfigGenerator +from functions import Consts, ContainerEnv, Functions, logger_easyhaproxy + + +class ProcessorInterface: + STATIC: Final[str] = "static" + DOCKER: Final[str] = "docker" + SWARM: Final[str] = "swarm" + KUBERNETES: Final[str] = "kubernetes" + + static_file = Consts.easyhaproxy_config + + def __init__(self, filename=None): + self.certbot_hosts = None + self.parsed_object = None + self.cfg = None + self.hosts = None + self.cfg = None + self.certbot_hosts = None + self.hosts = None + self.filename = filename + self.label = ContainerEnv.read()['lookup_label'] + self.refresh() + + @staticmethod + def factory(mode): + from .static import Static + from .docker import Docker + from .swarm import Swarm + from .kubernetes import Kubernetes + + if mode == ProcessorInterface.STATIC: + return Static(ProcessorInterface.static_file) + elif mode == ProcessorInterface.DOCKER: + return Docker() + elif mode == ProcessorInterface.SWARM: + return Swarm() + elif mode == ProcessorInterface.KUBERNETES: + return Kubernetes() + else: + logger_easyhaproxy.fatal(f"Expected mode to be 'static', 'docker', 'swarm' or 'kubernetes'. I got '{mode}'") + return None + + def refresh(self): + self.certbot_hosts = None + self.parsed_object = None + self.cfg = None + self.hosts = None + self.inspect_network() + self.parse() + + def inspect_network(self): + # Abstract + pass + + def parse(self): + self.cfg = HaproxyConfigGenerator(ContainerEnv.read()) + + def get_certbot_hosts(self): + return self.certbot_hosts + + def get_hosts(self): + return self.hosts + + def get_parsed_object(self): + return self.parsed_object + + def get_certs(self, key=None): + if key is None: + return self.cfg.certs + else: + return None if key not in self.cfg.certs else self.cfg.certs[key] + + def get_haproxy_conf(self): + conf = self.cfg.generate(self.parsed_object) + self.certbot_hosts = self.cfg.certbot_hosts + self.hosts = self.cfg.serving_hosts + return conf + + def save_config(self, filename): + Functions.save(filename, self.get_haproxy_conf()) + + def save_certs(self, path): + for cert in self.get_certs(): + Functions.save(f"{path}/{cert}", self.get_certs(cert)) \ No newline at end of file diff --git a/src/processor/kubernetes.py b/src/processor/kubernetes.py new file mode 100644 index 0000000..86d7ad6 --- /dev/null +++ b/src/processor/kubernetes.py @@ -0,0 +1,461 @@ +import base64 +import os +import socket +import time + +from kubernetes import client, config +from kubernetes.client.rest import ApiException + +from functions import Consts, ContainerEnv, Functions, logger_easyhaproxy + +from .interface import ProcessorInterface + + +class Kubernetes(ProcessorInterface): + def __init__(self, filename=None, api_instance=None, v1=None): + self.parsed_object = None + + # Only load config if API clients are not provided (allows dependency injection for testing) + if api_instance is None or v1 is None: + config.load_incluster_config() + config.verify_ssl = False + + # Use injected clients or create new ones (dependency injection pattern) + self.api_instance = api_instance or client.CoreV1Api() + self.v1 = v1 or client.NetworkingV1Api() + self.cert_cache = {} + self.deployment_mode_cache = None + self.ingress_addresses_cache = None + self.addresses_cache_time = 0 + super().__init__() + + def _detect_deployment_mode(self): + """ + Detect the deployment mode (daemonset, nodeport, clusterip). + Returns: tuple (mode: str, service: V1Service or None) + """ + # Return cached if available + if self.deployment_mode_cache: + return self.deployment_mode_cache + + env_config = ContainerEnv.read() + + # Check for manual override + if env_config['deployment_mode'] != 'auto': + logger_easyhaproxy.info(f"Using manual deployment mode: {env_config['deployment_mode']}") + service = self._get_easyhaproxy_service() if env_config['deployment_mode'] in ['nodeport', 'clusterip'] else None + self.deployment_mode_cache = (env_config['deployment_mode'], service) + return self.deployment_mode_cache + + try: + # Get current pod name from hostname + pod_name = socket.gethostname() + namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') + + # Read current pod + pod = self.api_instance.read_namespaced_pod(pod_name, namespace) + + # Check owner references to determine if DaemonSet or Deployment + if pod.metadata.owner_references: + owner_kind = pod.metadata.owner_references[0].kind + + if owner_kind == 'DaemonSet': + logger_easyhaproxy.info("Detected deployment mode: daemonset") + self.deployment_mode_cache = ('daemonset', None) + return self.deployment_mode_cache + elif owner_kind in ['ReplicaSet', 'Deployment']: + # Check if Service exists + service = self._get_easyhaproxy_service() + if service: + if service.spec.type == 'NodePort': + logger_easyhaproxy.info("Detected deployment mode: nodeport") + self.deployment_mode_cache = ('nodeport', service) + return self.deployment_mode_cache + else: + logger_easyhaproxy.info("Detected deployment mode: clusterip") + self.deployment_mode_cache = ('clusterip', service) + return self.deployment_mode_cache + except Exception as e: + logger_easyhaproxy.warn(f"Failed to detect deployment mode: {e}, defaulting to daemonset") + + self.deployment_mode_cache = ('daemonset', None) + return self.deployment_mode_cache + + def _get_easyhaproxy_service(self): + """Get the EasyHAProxy service if it exists.""" + try: + namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') + # Try common service names + service_names = ['easyhaproxy', 'ingress-easyhaproxy'] + + for service_name in service_names: + try: + service = self.api_instance.read_namespaced_service(service_name, namespace) + return service + except Exception: + continue + return None + except Exception as e: + logger_easyhaproxy.warn(f"Failed to get EasyHAProxy service: {e}") + return None + + def _get_ingress_addresses(self, mode, service): + """ + Get IP addresses or hostnames to report in ingress status. + + Args: + mode: Deployment mode (daemonset, nodeport, clusterip) + service: V1Service object (for nodeport/clusterip modes) + + Returns: + List of dicts: [{"ip": "..."}, {"hostname": "..."}] + """ + env_config = ContainerEnv.read() + cache_ttl = env_config.get('ingress_status_update_interval', 30) + + # Return cached if still valid + if self.ingress_addresses_cache and (time.time() - self.addresses_cache_time) < cache_ttl: + return self.ingress_addresses_cache + + addresses = [] + + try: + if mode == 'daemonset': + # Get nodes where DaemonSet pods are running + namespace = os.getenv('POD_NAMESPACE', 'easyhaproxy') + label_selector = "app.kubernetes.io/name=easyhaproxy" + + pods = self.api_instance.list_namespaced_pod(namespace, label_selector=label_selector) + node_names = set(pod.spec.node_name for pod in pods.items if pod.spec.node_name) + + # Get external IPs from these nodes + for node_name in node_names: + node = self.api_instance.read_node(node_name) + for addr in node.status.addresses: + if addr.type == 'ExternalIP': + addresses.append({"ip": addr.address}) + break + else: + # Fallback to InternalIP if no ExternalIP + for addr in node.status.addresses: + if addr.type == 'InternalIP': + addresses.append({"ip": addr.address}) + break + + elif mode == 'nodeport': + # Get all node IPs (traffic can reach any node via NodePort) + nodes = self.api_instance.list_node() + for node in nodes.items: + for addr in node.status.addresses: + if addr.type == 'ExternalIP': + addresses.append({"ip": addr.address}) + break + else: + # Fallback to InternalIP + for addr in node.status.addresses: + if addr.type == 'InternalIP': + addresses.append({"ip": addr.address}) + break + + elif mode == 'clusterip': + # Check if LoadBalancer status is available + if service and service.status and service.status.load_balancer: + lb_ingress = service.status.load_balancer.ingress or [] + for ing in lb_ingress: + if ing.ip: + addresses.append({"ip": ing.ip}) + if ing.hostname: + addresses.append({"hostname": ing.hostname}) + + # If no LoadBalancer, check for external hostname override + if not addresses and env_config['external_hostname']: + addresses.append({"hostname": env_config['external_hostname']}) + + # Fallback to ClusterIP + if not addresses and service: + addresses.append({"ip": service.spec.cluster_ip}) + + except Exception as e: + logger_easyhaproxy.warn(f"Failed to get ingress addresses: {e}") + + # Cache the result + self.ingress_addresses_cache = addresses + self.addresses_cache_time = time.time() + + return addresses + + def _update_ingress_status(self, ingress, addresses): + """ + Update the status of an ingress resource. + + Args: + ingress: V1Ingress object + addresses: List of address dicts [{"ip": "..."}, {"hostname": "..."}] + """ + if not addresses: + return + + try: + # Create status patch + status_body = { + "status": { + "loadBalancer": { + "ingress": addresses + } + } + } + + # Update status using patch (not replace) + self.v1.patch_namespaced_ingress_status( + name=ingress.metadata.name, + namespace=ingress.metadata.namespace, + body=status_body, + field_manager="easyhaproxy" + ) + + logger_easyhaproxy.debug( + f"Updated ingress {ingress.metadata.namespace}/{ingress.metadata.name} " + f"status with {len(addresses)} address(es)" + ) + + except Exception as e: + logger_easyhaproxy.warn( + f"Failed to update status for ingress " + f"{ingress.metadata.namespace}/{ingress.metadata.name}: {e}" + ) + + def _check_annotation(self, annotations, key, default=None): + if key not in annotations: + return default + return annotations[key] + + def inspect_network(self): + + ret = self.v1.list_ingress_for_all_namespaces(watch=False) + + # Detect deployment mode once per cycle for ingress status updates + env_config = ContainerEnv.read() + if env_config['update_ingress_status']: + deployment_mode, service = self._detect_deployment_mode() + ingress_addresses = self._get_ingress_addresses(deployment_mode, service) + else: + ingress_addresses = [] + + self.parsed_object = {} + for ingress in ret.items: + # Support both new spec.ingressClassName and deprecated annotation for backward compatibility + ingress_class = None + is_match = False + + # Check new spec.ingressClassName first (preferred) + if hasattr(ingress.spec, 'ingress_class_name') and ingress.spec.ingress_class_name is not None: + ingress_class = ingress.spec.ingress_class_name + # Modern spec uses 'easyhaproxy' + is_match = (ingress_class == "easyhaproxy") + # Fall back to deprecated annotation + elif ingress.metadata.annotations and 'kubernetes.io/ingress.class' in ingress.metadata.annotations: + ingress_class = ingress.metadata.annotations['kubernetes.io/ingress.class'] + # Deprecated annotation uses 'easyhaproxy-ingress' for backward compatibility + is_match = (ingress_class == "easyhaproxy-ingress") + + # Skip if no ingress class is defined or it doesn't match + if not is_match: + continue + + ssl_hosts = [] + + certbot = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.certbot") + redirect_ssl = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.redirect_ssl") + redirect = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.redirect") + mode = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.mode") + listen_port = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.listen_port", 80) + plugins = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.plugins") + + # Extract plugin-specific configurations + plugin_annotations = {} + for annotation_key, annotation_value in ingress.metadata.annotations.items(): + if annotation_key.startswith("easyhaproxy.plugin."): + plugin_annotations[annotation_key] = annotation_value + + # Get ingress name for logging + ingress_name = f"{ingress.metadata.namespace}/{ingress.metadata.name}" + + # Generic k8s_secret annotation processing + # Pattern: easyhaproxy.plugin.X.k8s_secret.KEY: "secret_name" or "secret_name/key_name" + # Result: easyhaproxy.plugin.X.KEY: "" + k8s_secret_annotations = {} + for annotation_key, secret_value in list(plugin_annotations.items()): + # Check if this annotation contains k8s_secret pattern + if ".k8s_secret." in annotation_key: + try: + # Parse the annotation key + # Example: "easyhaproxy.plugin.jwt_validator.k8s_secret.pubkey" -> "pubkey" + parts = annotation_key.split(".k8s_secret.") + if len(parts) != 2: + logger_easyhaproxy.warn( + f"Ingress {ingress_name} - Malformed k8s_secret annotation: {annotation_key}" + ) + continue + + prefix = parts[0] # "easyhaproxy.plugin.jwt_validator" + config_key = parts[1] # "pubkey" + target_annotation = f"{prefix}.{config_key}" # "easyhaproxy.plugin.jwt_validator.pubkey" + + # Parse secret_value: can be "secret_name" or "secret_name/key_name" + if "/" in secret_value: + secret_name, explicit_key_name = secret_value.split("/", 1) + use_explicit_key = True + else: + secret_name = secret_value + explicit_key_name = None + use_explicit_key = False + + # Read the secret + secret = self.api_instance.read_namespaced_secret( + secret_name, + ingress.metadata.namespace + ) + + # Try to find the key in the secret data + secret_data = None + tried_keys = [] + + if use_explicit_key: + # User specified exact key name - only try that one + tried_keys = [explicit_key_name] + if explicit_key_name in secret.data: + secret_data = secret.data[explicit_key_name] + logger_easyhaproxy.debug( + f"Ingress {ingress_name} - Found explicit secret key '{explicit_key_name}' " + f"in secret '{secret_name}'" + ) + else: + # No explicit key - try config_key and common variations + tried_keys = [config_key] + if config_key in secret.data: + secret_data = secret.data[config_key] + else: + # Try common variations for the requested key + variations = [] + if config_key == "pubkey": + variations = ["public-key", "jwt.pub", "tls.crt"] + elif config_key == "password": + variations = ["pass", "pwd"] + elif config_key == "api_key": + variations = ["apikey", "api-key", "key"] + + for variation in variations: + tried_keys.append(variation) + if variation in secret.data: + secret_data = secret.data[variation] + logger_easyhaproxy.debug( + f"Ingress {ingress_name} - Found secret key '{variation}' " + f"for requested key '{config_key}'" + ) + break + + if secret_data: + # Decode from base64 (Kubernetes secrets are base64-encoded) + # Then re-encode to base64 for plugin (plugin expects base64-encoded) + decoded = base64.b64decode(secret_data).decode('ascii') + reencoded = base64.b64encode(decoded.encode('ascii')).decode('ascii') + + # Store the processed annotation + k8s_secret_annotations[target_annotation] = reencoded + + logger_easyhaproxy.info( + f"Ingress {ingress_name} - Loaded '{config_key}' from secret " + f"'{secret_name}' for annotation '{target_annotation}'" + ) + else: + logger_easyhaproxy.warn( + f"Ingress {ingress_name} - Secret '{secret_name}' found but " + f"no matching key (tried: {', '.join(tried_keys)})" + ) + + except Exception as e: + logger_easyhaproxy.warn( + f"Ingress {ingress_name} - Failed to process k8s_secret annotation " + f"'{annotation_key}' with value '{secret_value}': {e}" + ) + + # Merge k8s_secret annotations into plugin_annotations + # k8s_secret annotations will NOT override existing explicit annotations (lower priority) + for key, value in k8s_secret_annotations.items(): + if key not in plugin_annotations: + plugin_annotations[key] = value + else: + logger_easyhaproxy.debug( + f"Ingress {ingress_name} - Skipping k8s_secret annotation '{key}' " + f"because explicit annotation already exists" + ) + + data = {"creation_timestamp": ingress.metadata.creation_timestamp.strftime("%x %X"), + "resource_version": ingress.metadata.resource_version, "namespace": ingress.metadata.namespace} + + if ingress.spec.tls is not None: + for tls in ingress.spec.tls: + try: + secret = self.api_instance.read_namespaced_secret(tls.secret_name, ingress.metadata.namespace) + if "tls.crt" not in secret.data or "tls.key" not in secret.data: + continue + + if tls.secret_name not in self.cert_cache or self.cert_cache[tls.secret_name] != secret.data: + self.cert_cache[tls.secret_name] = secret.data + Functions.save( + f"{Consts.certs_haproxy}/{tls.secret_name}.pem", + base64.b64decode(secret.data["tls.crt"]).decode('ascii') + "\n" + base64.b64decode( + secret.data["tls.key"]).decode('ascii') + ) + + ssl_hosts.extend(tls.hosts) + except Exception as e: + logger_easyhaproxy.warn(f"Ingress {ingress_name} - Get secret failed: '{e}'") + + logger_easyhaproxy.debug(f"Ingress {ingress_name} - SSL Hosts found '{ssl_hosts}'") + + for rule in ingress.spec.rules: + rule_data = {} + port_number = rule.http.paths[0].backend.service.port.number + definition = f"easyhaproxy.{rule.host.replace('.', '-')}_{port_number}" + rule_data[f"{definition}.host"] = rule.host + rule_data[f"{definition}.port"] = listen_port + rule_data[f"{definition}.localport"] = port_number + if rule.host in ssl_hosts: + rule_data[f"{definition}.clone_to_ssl"] = 'true' + if redirect_ssl is not None: + rule_data[f"{definition}.redirect_ssl"] = redirect_ssl + if certbot is not None: + rule_data[f"{definition}.certbot"] = certbot + if redirect is not None: + rule_data[f"{definition}.redirect"] = redirect + if mode is not None: + rule_data[f"{definition}.mode"] = mode + rule_data[f"{definition}.balance"] = self._check_annotation(ingress.metadata.annotations, "easyhaproxy.balance", "roundrobin") + + # Add plugin configuration + if plugins is not None: + rule_data[f"{definition}.plugins"] = plugins + + # Add plugin-specific configurations + for plugin_key, plugin_value in plugin_annotations.items(): + # Convert easyhaproxy.plugin.X.Y to easyhaproxy.{definition}.plugin.X.Y + plugin_config_key = plugin_key.replace("easyhaproxy.plugin.", f"{definition}.plugin.") + rule_data[plugin_config_key] = plugin_value + + service_name = rule.http.paths[0].backend.service.name + try: + api_response = self.api_instance.read_namespaced_service(service_name, ingress.metadata.namespace) + cluster_ip = api_response.spec.cluster_ip + except ApiException as e: + cluster_ip = None + logger_easyhaproxy.warn(f"Ingress {ingress_name} - Service {service_name} - Failed: '{e}'") + + if cluster_ip is not None: + if cluster_ip not in self.parsed_object.keys(): + self.parsed_object[cluster_ip] = data + self.parsed_object[cluster_ip].update(rule_data) + + # Update ingress status if enabled + if env_config['update_ingress_status'] and ingress_addresses: + self._update_ingress_status(ingress, ingress_addresses) \ No newline at end of file diff --git a/src/processor/static.py b/src/processor/static.py new file mode 100644 index 0000000..c27b7c0 --- /dev/null +++ b/src/processor/static.py @@ -0,0 +1,143 @@ +import json + +import yaml + +from easymapping import HaproxyConfigGenerator +from functions import ContainerEnv, Functions + +from .interface import ProcessorInterface + + +class Static(ProcessorInterface): + def __init__(self, filename=None): + self.parsed_object = None + self.static_content = None + self.static_content = None + self.cfg = None + super().__init__(filename) + + def inspect_network(self): + """Load YAML and convert containers to Docker-style container metadata""" + # Load YAML + self.static_content = yaml.load(Functions.load(self.filename), Loader=yaml.FullLoader) + + # Convert containers to label format + self.parsed_object = self._convert_yaml_to_labels() + + def _convert_yaml_to_labels(self): + """ + Convert static YAML containers to Docker label format. + Returns: {IP: {labels}} structure that parse() can process + """ + container_metadata = {} + + # Get global plugin configuration + global_plugins = self.static_content.get("plugins", {}) + global_enabled = global_plugins.get("enabled", []) + global_plugin_config = global_plugins.get("config", {}) + + for host_port, config in self.static_content.get("containers", {}).items(): + # Parse hostname:port from key + if ":" in host_port: + hostname, port = host_port.rsplit(":", 1) + else: + hostname = host_port + port = "80" + + # Create definition: hostname_port (e.g., host1_com_br_80) + definition = hostname.replace(".", "_") + f"_{port}" + + # Handle redirect-only entries (no backend) + if "redirect" in config and "ip" not in config: + # Create metadata with redirect but mark as redirect-only to skip backend creation + fake_ip = f"redirect-{hostname}-{port}" + if fake_ip not in container_metadata: + container_metadata[fake_ip] = {} + + container_metadata[fake_ip].update({ + f"easyhaproxy.{definition}.host": hostname, + f"easyhaproxy.{definition}.port": port, + f"easyhaproxy.{definition}.redirect": json.dumps({hostname: config["redirect"]}), + f"easyhaproxy.{definition}.redirect_only": "true", # Marker to skip backend + }) + continue + + # Get IPs/containers + ip_list = config.get("ip", [hostname]) + + # Process each container/IP + for container_spec in ip_list: + # Parse container:localport + if ":" in container_spec: + container_addr, localport = container_spec.rsplit(":", 1) + else: + container_addr = container_spec + localport = "80" + + # Use container address as IP (could be IP, DNS, or container name) + ip = container_addr + + # Build labels dict + labels = { + f"easyhaproxy.{definition}.host": hostname, + f"easyhaproxy.{definition}.port": port, + f"easyhaproxy.{definition}.localport": localport, + } + + # Add optional settings + for key in ["mode", "certbot", "redirect_ssl", "ssl", "balance", "proto", "ssl-check", "clone_to_ssl"]: + if key in config: + value = config[key] + # Convert boolean to string + if isinstance(value, bool): + value = "true" if value else "false" + labels[f"easyhaproxy.{definition}.{key}"] = str(value) + + # Handle plugins + host_plugins = config.get("plugins", global_enabled) + if host_plugins: + # Convert list to comma-separated string if needed + if isinstance(host_plugins, list): + plugins_str = ",".join(host_plugins) + else: + plugins_str = host_plugins + labels[f"easyhaproxy.{definition}.plugins"] = plugins_str + + # Process plugin configurations + host_plugin_config = config.get("plugin", {}) + + # Parse plugins list + plugins_list = host_plugins if isinstance(host_plugins, list) else [p.strip() for p in host_plugins.split(",")] + + for plugin_name in plugins_list: + # Merge global and host-specific config + merged_config = {} + if plugin_name in global_plugin_config: + merged_config.update(global_plugin_config[plugin_name]) + if plugin_name in host_plugin_config: + merged_config.update(host_plugin_config[plugin_name]) + + # Convert plugin config to labels + for config_key, config_value in merged_config.items(): + label_key = f"easyhaproxy.{definition}.plugin.{plugin_name}.{config_key}" + + # Convert list values to comma-separated strings + if isinstance(config_value, list): + label_value = ",".join(str(v) for v in config_value) + else: + label_value = str(config_value) + + labels[label_key] = label_value + + # Initialize container entry if it doesn't exist + if ip not in container_metadata: + container_metadata[ip] = {} + + # Merge labels instead of overwriting + container_metadata[ip].update(labels) + + return container_metadata + + def parse(self): + """Create HaproxyConfigGenerator with YAML config merged into env vars""" + self.cfg = HaproxyConfigGenerator(ContainerEnv.read(self.static_content)) \ No newline at end of file diff --git a/src/processor/swarm.py b/src/processor/swarm.py new file mode 100644 index 0000000..b2be86f --- /dev/null +++ b/src/processor/swarm.py @@ -0,0 +1,50 @@ +import socket + +import docker + +from .interface import ProcessorInterface + + +class Swarm(ProcessorInterface): + def __init__(self, filename=None): + self.parsed_object = None + self.client = docker.from_env() + super().__init__() + + def inspect_network(self): + ha_proxy_service_name = self.client.containers.get(socket.gethostname()).name.split('.')[0] + ha_proxy_network_id = None + swarm_ingress_id = None + + # Get the HAProxy network and the ingress network + for endpoint in self.client.services.get(ha_proxy_service_name).attrs['Endpoint']["VirtualIPs"]: + network_name = self.client.networks.get(endpoint["NetworkID"]).name + if swarm_ingress_id is None and network_name == 'ingress': + swarm_ingress_id = endpoint["NetworkID"] + if ha_proxy_network_id is None and network_name != 'ingress': + ha_proxy_network_id = endpoint["NetworkID"] + if ha_proxy_network_id is not None and swarm_ingress_id is not None: + break + + # Check if the service is attached to the HAProxy network + self.parsed_object = {} + for service in self.client.services.list(): + if not any(self.label in key for key in service.attrs["Spec"]["Labels"]): + continue + + ip_address = None + network_list = [] + for endpoint in service.attrs["Endpoint"]["VirtualIPs"]: + if ha_proxy_network_id == endpoint["NetworkID"]: + ip_address = endpoint["Addr"].split("/")[0] + break + elif swarm_ingress_id != endpoint["NetworkID"]: + network_list.append(endpoint["NetworkID"]) + + # Attach the service to the HAProxy network + if ip_address is None: + network_list.append(ha_proxy_network_id) + service.update(networks=network_list) + continue # skip to the next service to give time to update the network + + self.parsed_object[ip_address] = service.attrs["Spec"]["Labels"] \ No newline at end of file