Fluentbit Validator

*.db
*.db-journal

#!/usr/bin/env python3
"""
Fluent Bit Configuration Validator

Comprehensive validation tool for Fluent Bit configurations.
Checks syntax, semantics, security, performance, and best practices.
"""

import argparse
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
from typing import Dict, List, Optional, Set


class FluentBitValidator:
    """Validates Fluent Bit configuration files."""

    VALID_INPUT_PLUGINS = {
        "tail",
        "systemd",
        "tcp",
        "udp",
        "forward",
        "http",
        "syslog",
        "docker",
        "kubernetes",
        "exec",
        "dummy",
    }

    VALID_FILTER_PLUGINS = {
        "grep",
        "kubernetes",
        "parser",
        "modify",
        "nest",
        "rewrite_tag",
        "throttle",
        "multiline",
        "record_modifier",
        "lua",
        "geoip2",
        "expect",
        "type_converter",
        "stdout",
        "log_to_metrics",
        "wasm",
    }

    VALID_OUTPUT_PLUGINS = {
        "es",
        "elasticsearch",
        "kafka",
        "loki",
        "s3",
        "cloudwatch",
        "cloudwatch_logs",
        "http",
        "forward",
        "stdout",
        "file",
        "opentelemetry",
        "null",
        "influxdb",
        "datadog",
        "splunk",
        "stackdriver",
        "azure",
        "tcp",
        "udp",
        "nats",
        "counter",
        "flowcounter",
        "firehose",
        "kinesis_firehose",
        "kinesis_streams",
        "gelf",
        "pgsql",
    }

    def __init__(self, config_file: str, require_dry_run: bool = False):
        self.config_file = config_file
        self.require_dry_run = require_dry_run
        self.errors = []
        self.warnings = []
        self.recommendations = []
        self.sections = []

    def validate_all(self) -> bool:
        """Run all validation checks."""
        checks = [
            self.validate_structure,
            self.validate_syntax,
            self.validate_sections,
            self.validate_tags,
            self.validate_security,
            self.validate_performance,
            self.validate_best_practices,
            self.validate_dry_run,
        ]

        for check in checks:
            check()

        return len(self.errors) == 0

    def validate_structure(self) -> None:
        """Validate basic file structure."""
        if not os.path.exists(self.config_file):
            self.errors.append(f"Configuration file not found: {self.config_file}")
            return

        if not os.access(self.config_file, os.R_OK):
            self.errors.append(f"Configuration file not readable: {self.config_file}")
            return

        # Check if file is empty
        if os.path.getsize(self.config_file) == 0:
            self.errors.append("Configuration file is empty")
            return

        # Parse file and store line numbers
        self._parse_config()

    def _parse_config(self) -> None:
        """Parse configuration file and build section list."""
        try:
            self.sections = []

            with open(self.config_file, "r", encoding="utf-8") as f:
                lines = f.readlines()

            current_section = None

            for i, line in enumerate(lines, start=1):
                stripped = line.strip()

                # Skip empty lines and comments
                if not stripped or stripped.startswith("#"):
                    continue

                # Handle @INCLUDE and @SET preprocessor directives
                if stripped.startswith("@"):
                    directive_lower = stripped.lower()
                    if directive_lower.startswith("@include"):
                        # Included file is not followed here; note it as a recommendation
                        included = stripped[len("@include"):].strip()
                        self.recommendations.append(
                            f"Line {i}: @INCLUDE '{included}' is not validated "
                            f"(run the validator on the included file separately)"
                        )
                    elif directive_lower.startswith("@set"):
                        pass  # @SET variable definitions are silently accepted
                    else:
                        self.warnings.append(
                            f"Line {i}: Unknown preprocessor directive '{stripped.split()[0]}'"
                        )
                    continue

                # Detect mixed indentation (tabs + spaces)
                indent_match = re.match(r"^[ \t]+", line)
                if indent_match:
                    indent = indent_match.group(0)
                    if " " in indent and "\t" in indent:
                        self.warnings.append(
                            f"Line {i}: Mixed tabs and spaces in indentation"
                        )

                # Section header
                if stripped.startswith("["):
                    if not stripped.endswith("]"):
                        self.errors.append(
                            f"Line {i}: Malformed section header '{stripped}'"
                        )
                        current_section = None
                        continue

                    section_name = stripped[1:-1].strip().upper()
                    if not section_name:
                        self.errors.append(f"Line {i}: Empty section header []")
                        current_section = None
                        continue

                    current_section = {
                        "type": section_name,
                        "line": i,
                        "params": {},
                        "param_entries": [],
                    }
                    self.sections.append(current_section)
                    continue

                if current_section is None:
                    self.errors.append(
                        f"Line {i}: Parameter outside of a section '{stripped}'"
                    )
                    continue

                key, value = self._parse_key_value(stripped)
                if key is None:
                    self.errors.append(f"Line {i}: Malformed key-value pair '{stripped}'")
                    continue

                normalized_key = key.lower()
                current_section["param_entries"].append(
                    {
                        "key": normalized_key,
                        "value": value,
                        "line": i,
                    }
                )
                current_section["params"][normalized_key] = {
                    "value": value,
                    "line": i,
                }

        except Exception as e:
            self.errors.append(f"Failed to parse configuration: {str(e)}")

    def _parse_key_value(self, line: str):
        """Parse key-value pair supporting both whitespace and '=' delimiters."""
        equals_match = re.match(r"^([^\s=]+)\s*=\s*(.*)$", line)
        if equals_match:
            return equals_match.group(1), equals_match.group(2).strip()

        space_match = re.match(r"^([^\s=]+)\s+(.*)$", line)
        if space_match:
            return space_match.group(1), space_match.group(2).strip()

        return None, None

    def validate_syntax(self) -> None:
        """Validate INI syntax."""
        valid_sections = ["SERVICE", "INPUT", "FILTER", "OUTPUT", "PARSER", "MULTILINE_PARSER"]

        for section in self.sections:
            if section["type"] not in valid_sections:
                self.warnings.append(
                    f"Line {section['line']}: Unknown section type [{section['type']}]"
                )

    def validate_sections(self) -> None:
        """Validate individual sections."""
        has_service = False
        has_input = False
        has_output = False

        for section in self.sections:
            section_type = section["type"]
            params = section["params"]

            if section_type == "SERVICE":
                has_service = True
                self._validate_service_section(section)
            elif section_type == "INPUT":
                has_input = True
                self._validate_input_section(section)
            elif section_type == "FILTER":
                self._validate_filter_section(section)
            elif section_type == "OUTPUT":
                has_output = True
                self._validate_output_section(section)
            elif section_type in ["PARSER", "MULTILINE_PARSER"]:
                self._validate_parser_section(section)

        # Check required sections
        if not has_service:
            self.warnings.append("Missing [SERVICE] section (recommended)")
        if not has_input:
            self.errors.append("Missing [INPUT] section (required)")
        if not has_output:
            self.errors.append("Missing [OUTPUT] section (required)")

    def _validate_service_section(self, section: Dict) -> None:
        """Validate SERVICE section."""
        params = section["params"]

        # Check Flush parameter
        if "flush" not in params:
            self.warnings.append(
                f"Line {section['line']}: [SERVICE] missing Flush parameter (recommended)"
            )
        else:
            flush_line = params["flush"]["line"]
            try:
                flush_val = float(params["flush"]["value"])
                if flush_val < 1:
                    self.warnings.append(
                        f"Line {flush_line}: Flush interval < 1 second (very low, high CPU usage)"
                    )
                elif flush_val > 10:
                    self.warnings.append(
                        f"Line {flush_line}: Flush interval > 10 seconds (high latency)"
                    )
            except ValueError:
                self.errors.append(
                    f"Line {flush_line}: Flush must be a number (got: {params['flush']['value']})"
                )

        # Check Log_Level
        if "log_level" in params:
            valid_levels = ["off", "error", "warn", "info", "debug", "trace"]
            log_level = params["log_level"]["value"].lower()
            if log_level not in valid_levels:
                self.errors.append(
                    f"Line {params['log_level']['line']}: Invalid Log_Level '{log_level}' "
                    f"(valid: {', '.join(valid_levels)})"
                )

        # Check Parsers_File existence
        if "parsers_file" in params:
            parser_file = params["parsers_file"]["value"]
            # Try to resolve relative to config file
            config_dir = os.path.dirname(self.config_file)
            parser_path = os.path.join(config_dir, parser_file)
            if not os.path.exists(parser_path) and not os.path.exists(parser_file):
                self.warnings.append(
                    f"Line {params['parsers_file']['line']}: Parsers_File '{parser_file}' not found"
                )

    def _validate_input_section(self, section: Dict) -> None:
        """Validate INPUT section."""
        params = section["params"]

        # Check required Name parameter
        if "name" not in params:
            self.errors.append(f"Line {section['line']}: [INPUT] missing required parameter 'Name'")
            return

        plugin_name = params["name"]["value"]
        plugin_name_normalized = plugin_name.lower()

        if plugin_name_normalized not in self.VALID_INPUT_PLUGINS:
            self.warnings.append(
                f"Line {params['name']['line']}: Unknown INPUT plugin '{plugin_name}'"
            )

        # Check Tag parameter (recommended)
        if "tag" not in params:
            if plugin_name_normalized != "forward":  # forward provides dynamic tags
                self.warnings.append(
                    f"Line {section['line']}: [INPUT] missing Tag parameter (recommended)"
                )

        # tail plugin specific checks
        if plugin_name_normalized == "tail":
            if "path" not in params:
                self.errors.append(
                    f"Line {section['line']}: [INPUT tail] missing required parameter 'Path'"
                )

            if "mem_buf_limit" not in params:
                self.warnings.append(
                    f"Line {section['line']}: [INPUT tail] missing Mem_Buf_Limit (OOM risk)"
                )

            if "db" not in params:
                self.warnings.append(
                    f"Line {section['line']}: [INPUT tail] missing DB parameter (no crash recovery)"
                )

            if "skip_long_lines" not in params:
                self.recommendations.append(
                    f"Line {section['line']}: [INPUT tail] consider adding Skip_Long_Lines On"
                )

    def _validate_filter_section(self, section: Dict) -> None:
        """Validate FILTER section."""
        params = section["params"]

        # Check required parameters
        if "name" not in params:
            self.errors.append(f"Line {section['line']}: [FILTER] missing required parameter 'Name'")
            return

        filter_name = params["name"]["value"]
        filter_name_normalized = filter_name.lower()

        if filter_name_normalized not in self.VALID_FILTER_PLUGINS:
            self.warnings.append(
                f"Line {params['name']['line']}: Unknown FILTER plugin '{filter_name}' "
                f"(plugin-specific validation skipped)"
            )

        if "match" not in params and "match_regex" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER] missing required parameter 'Match' or 'Match_Regex'"
            )

        # Filter-specific validation
        if filter_name_normalized == "kubernetes":
            self._validate_kubernetes_filter(section, params)
        elif filter_name_normalized == "parser":
            self._validate_parser_filter(section, params)
        elif filter_name_normalized == "grep":
            self._validate_grep_filter(section, params)
        elif filter_name_normalized == "modify":
            self._validate_modify_filter(section, params)
        elif filter_name_normalized == "nest":
            self._validate_nest_filter(section, params)
        elif filter_name_normalized == "rewrite_tag":
            self._validate_rewrite_tag_filter(section, params)
        elif filter_name_normalized == "throttle":
            self._validate_throttle_filter(section, params)
        elif filter_name_normalized == "multiline":
            self._validate_multiline_filter(section, params)

    def _validate_kubernetes_filter(self, section: Dict, params: Dict) -> None:
        """Validate kubernetes filter specific parameters."""
        # Check for common K8s filter parameters
        if "kube_url" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [FILTER kubernetes] consider setting Kube_URL "
                f"(default: https://kubernetes.default.svc:443)"
            )

        # Recommend best practices
        if "merge_log" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [FILTER kubernetes] consider setting Merge_Log On to parse JSON logs"
            )

        if "keep_log" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [FILTER kubernetes] consider setting Keep_Log Off to reduce payload size"
            )

        if "labels" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [FILTER kubernetes] consider enabling Labels On for pod labels"
            )

        # Buffer_Size recommendation
        if "buffer_size" in params:
            buffer_size = params["buffer_size"]["value"]
            if buffer_size != "0":
                self.recommendations.append(
                    f"Line {params['buffer_size']['line']}: [FILTER kubernetes] Buffer_Size 0 is recommended for performance"
                )

    def _validate_parser_filter(self, section: Dict, params: Dict) -> None:
        """Validate parser filter specific parameters."""
        if "key_name" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER parser] missing required parameter 'Key_Name'"
            )

        if "parser" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER parser] missing required parameter 'Parser'"
            )

        # Recommend Reserve_Data
        if "reserve_data" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [FILTER parser] consider setting Reserve_Data On to keep unparsed data"
            )

    def _validate_grep_filter(self, section: Dict, params: Dict) -> None:
        """Validate grep filter specific parameters."""
        has_regex = "regex" in params
        has_exclude = "exclude" in params

        if not has_regex and not has_exclude:
            self.warnings.append(
                f"Line {section['line']}: [FILTER grep] has neither Regex nor Exclude parameter (no filtering will occur)"
            )

        # Validate regex patterns if present
        if has_regex:
            regex_value = params["regex"]["value"]
            parts = regex_value.split(None, 1)
            if len(parts) != 2:
                self.warnings.append(
                    f"Line {params['regex']['line']}: [FILTER grep] Regex format should be 'key pattern'"
                )

    def _validate_modify_filter(self, section: Dict, params: Dict) -> None:
        """Validate modify filter specific parameters."""
        has_operation = any(key in params for key in ["add", "remove", "set", "rename", "copy", "hard_rename", "hard_copy"])

        if not has_operation:
            self.warnings.append(
                f"Line {section['line']}: [FILTER modify] no operation specified "
                f"(expected: Add, Remove, Set, Rename, Copy, etc.)"
            )

    def _validate_nest_filter(self, section: Dict, params: Dict) -> None:
        """Validate nest filter specific parameters."""
        if "operation" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER nest] missing required parameter 'Operation'"
            )
        else:
            operation = params["operation"]["value"].lower()
            valid_operations = ["nest", "lift"]
            if operation not in valid_operations:
                self.errors.append(
                    f"Line {params['operation']['line']}: [FILTER nest] invalid Operation '{operation}' "
                    f"(valid: {', '.join(valid_operations)})"
                )

        if "nested_under" not in params and "nest_under" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER nest] missing required parameter 'Nested_under'"
            )

    def _validate_rewrite_tag_filter(self, section: Dict, params: Dict) -> None:
        """Validate rewrite_tag filter specific parameters."""
        if "rule" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER rewrite_tag] missing required parameter 'Rule'"
            )

    def _validate_throttle_filter(self, section: Dict, params: Dict) -> None:
        """Validate throttle filter specific parameters."""
        if "rate" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER throttle] missing required parameter 'Rate'"
            )

    def _validate_multiline_filter(self, section: Dict, params: Dict) -> None:
        """Validate multiline filter specific parameters."""
        if "multiline.parser" not in params:
            self.errors.append(
                f"Line {section['line']}: [FILTER multiline] missing required parameter 'multiline.parser'"
            )

    def _validate_output_section(self, section: Dict) -> None:
        """Validate OUTPUT section."""
        params = section["params"]

        # Check required parameters
        if "name" not in params:
            self.errors.append(f"Line {section['line']}: [OUTPUT] missing required parameter 'Name'")
            return

        plugin_name = params["name"]["value"]
        plugin_name_normalized = plugin_name.lower()

        if plugin_name_normalized not in self.VALID_OUTPUT_PLUGINS:
            self.warnings.append(
                f"Line {params['name']['line']}: Unknown OUTPUT plugin '{plugin_name}' "
                f"(plugin-specific validation skipped)"
            )

        if "match" not in params and "match_regex" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT] missing required parameter 'Match' or 'Match_Regex'"
            )

        # Check Retry_Limit
        if "retry_limit" not in params:
            self.warnings.append(
                f"Line {section['line']}: [OUTPUT] missing Retry_Limit (infinite retries)"
            )

        # Plugin-specific checks
        if plugin_name_normalized in ["es", "elasticsearch"]:
            self._validate_elasticsearch_output(section, params)
        elif plugin_name_normalized == "kafka":
            self._validate_kafka_output(section, params)
        elif plugin_name_normalized == "loki":
            self._validate_loki_output(section, params)
        elif plugin_name_normalized == "s3":
            self._validate_s3_output(section, params)
        elif plugin_name_normalized in ["cloudwatch", "cloudwatch_logs"]:
            self._validate_cloudwatch_output(section, params)
        elif plugin_name_normalized == "http":
            self._validate_http_output(section, params)
        elif plugin_name_normalized == "forward":
            self._validate_forward_output(section, params)
        elif plugin_name_normalized == "stdout":
            self._validate_stdout_output(section, params)
        elif plugin_name_normalized == "file":
            self._validate_file_output(section, params)
        elif plugin_name_normalized == "opentelemetry":
            self._validate_opentelemetry_output(section, params)

    def _validate_elasticsearch_output(self, section: Dict, params: Dict) -> None:
        """Validate Elasticsearch output specific parameters."""
        if "host" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT es] missing required parameter 'Host'"
            )

        # Recommend Logstash format for better indexing
        if "logstash_format" not in params and "index" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT es] consider using Logstash_Format On or specify Index"
            )

        # Check for TLS in production
        if "tls" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT es] consider enabling TLS for production"
            )

    def _validate_kafka_output(self, section: Dict, params: Dict) -> None:
        """Validate Kafka output specific parameters."""
        if "brokers" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT kafka] missing required parameter 'Brokers'"
            )

        if "topics" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT kafka] missing required parameter 'Topics'"
            )

        # Recommend message format
        if "format" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT kafka] consider setting Format (json, msgpack, gelf)"
            )

    def _validate_loki_output(self, section: Dict, params: Dict) -> None:
        """Validate Loki output specific parameters."""
        if "host" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT loki] missing required parameter 'Host'"
            )

        # Recommend label configuration
        if "labels" not in params and "auto_kubernetes_labels" not in params:
            self.warnings.append(
                f"Line {section['line']}: [OUTPUT loki] missing labels configuration "
                f"(set 'labels' or 'auto_kubernetes_labels on')"
            )

        # Check line format
        if "line_format" in params:
            line_format = params["line_format"]["value"].lower()
            valid_formats = ["json", "key_value"]
            if line_format not in valid_formats:
                self.warnings.append(
                    f"Line {params['line_format']['line']}: [OUTPUT loki] invalid line_format '{line_format}' "
                    f"(valid: {', '.join(valid_formats)})"
                )

    def _validate_s3_output(self, section: Dict, params: Dict) -> None:
        """Validate S3 output specific parameters."""
        if "bucket" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT s3] missing required parameter 'bucket'"
            )

        if "region" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT s3] missing required parameter 'region'"
            )

        # Recommend compression
        if "compression" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT s3] consider enabling compression (gzip)"
            )

        # Recommend s3_key_format for organization
        if "s3_key_format" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT s3] consider setting s3_key_format for log organization"
            )

    def _validate_cloudwatch_output(self, section: Dict, params: Dict) -> None:
        """Validate CloudWatch Logs output specific parameters."""
        if "region" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT cloudwatch_logs] missing required parameter 'region'"
            )

        if "log_group_name" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT cloudwatch_logs] missing required parameter 'log_group_name'"
            )

        # Recommend auto_create_group
        if "auto_create_group" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT cloudwatch_logs] consider setting auto_create_group On"
            )

    def _validate_http_output(self, section: Dict, params: Dict) -> None:
        """Validate HTTP output specific parameters."""
        if "host" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT http] missing required parameter 'Host'"
            )

        if "uri" not in params:
            self.warnings.append(
                f"Line {section['line']}: [OUTPUT http] missing URI parameter (will use /)"
            )

        # Recommend format
        if "format" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT http] consider setting Format (json, msgpack)"
            )

        # Recommend compression
        if "compress" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT http] consider enabling Compress (gzip)"
            )

    def _validate_forward_output(self, section: Dict, params: Dict) -> None:
        """Validate Forward output specific parameters."""
        if "host" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT forward] missing required parameter 'Host'"
            )

        # Check for shared_key in secure mode
        if "require_ack_response" in params:
            require_ack = params["require_ack_response"]["value"].lower()
            if require_ack in ["on", "true", "yes"] and "shared_key" not in params:
                self.warnings.append(
                    f"Line {section['line']}: [OUTPUT forward] Require_ack_response On but missing Shared_Key"
                )

    def _validate_stdout_output(self, section: Dict, params: Dict) -> None:
        """Validate stdout output specific parameters."""
        # stdout is mainly for debugging, check format
        if "format" in params:
            format_val = params["format"]["value"].lower()
            valid_formats = ["json", "json_lines", "msgpack"]
            if format_val not in valid_formats:
                self.warnings.append(
                    f"Line {params['format']['line']}: [OUTPUT stdout] invalid Format '{format_val}' "
                    f"(valid: {', '.join(valid_formats)})"
                )

    def _validate_file_output(self, section: Dict, params: Dict) -> None:
        """Validate file output specific parameters."""
        if "path" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT file] missing required parameter 'Path'"
            )

        # Check if path is writable
        if "path" in params:
            path = params["path"]["value"]
            parent_dir = os.path.dirname(path) if os.path.dirname(path) else "."
            if os.path.exists(parent_dir) and not os.access(parent_dir, os.W_OK):
                self.warnings.append(
                    f"Line {params['path']['line']}: [OUTPUT file] Path '{path}' may not be writable"
                )

    def _validate_opentelemetry_output(self, section: Dict, params: Dict) -> None:
        """Validate OpenTelemetry output specific parameters (Fluent Bit 2.x+)."""
        # Check for Host parameter (required)
        if "host" not in params:
            self.errors.append(
                f"Line {section['line']}: [OUTPUT opentelemetry] missing required parameter 'Host'"
            )

        # Check Port (optional, defaults to 4317 for gRPC, 4318 for HTTP)
        if "port" in params:
            try:
                port = int(params["port"]["value"])
                if port < 1 or port > 65535:
                    self.errors.append(
                        f"Line {params['port']['line']}: [OUTPUT opentelemetry] Port must be between 1-65535"
                    )
            except ValueError:
                self.errors.append(
                    f"Line {params['port']['line']}: [OUTPUT opentelemetry] Port must be a number"
                )

        # Recommend specific URI endpoints
        has_uri = any(key in params for key in ["metrics_uri", "logs_uri", "traces_uri"])
        if not has_uri:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT opentelemetry] consider specifying metrics_uri, logs_uri, or traces_uri"
            )

        # Check for authentication header
        if "header" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT opentelemetry] consider adding Header for authentication "
                f"(e.g., Header Authorization Bearer ${{OTEL_TOKEN}})"
            )
        else:
            # Check if header contains hardcoded credentials
            header_value = params["header"]["value"]
            if "Bearer " in header_value and "${" not in header_value:
                self.warnings.append(
                    f"Line {params['header']['line']}: [OUTPUT opentelemetry] Header may contain hardcoded credentials "
                    f"(use environment variable: Header Authorization Bearer ${{OTEL_TOKEN}})"
                )

        # TLS checks are handled globally by validate_security(); no duplication here.

        # Recommend add_label for metadata
        if "add_label" not in params:
            self.recommendations.append(
                f"Line {section['line']}: [OUTPUT opentelemetry] consider using add_label to add resource attributes"
            )

    def _validate_parser_section(self, section: Dict) -> None:
        """Validate PARSER and MULTILINE_PARSER sections."""
        params = section["params"]
        section_type = section["type"]

        if "name" not in params:
            self.errors.append(f"Line {section['line']}: [{section_type}] missing required parameter 'Name'")

        # PARSER-specific validation
        if section_type == "PARSER":
            if "format" not in params:
                self.errors.append(f"Line {section['line']}: [PARSER] missing required parameter 'Format'")
            else:
                parser_format = params["format"]["value"].lower()
                valid_formats = ["json", "regex", "ltsv", "logfmt"]
                if parser_format not in valid_formats:
                    self.warnings.append(
                        f"Line {params['format']['line']}: [PARSER] unknown Format '{parser_format}' "
                        f"(expected: {', '.join(valid_formats)})"
                    )

                # Regex-specific checks
                if parser_format == "regex":
                    if "regex" not in params:
                        self.errors.append(
                            f"Line {section['line']}: [PARSER regex] missing required parameter 'Regex'"
                        )

                # Time parsing checks
                if "time_key" in params and "time_format" not in params:
                    self.warnings.append(
                        f"Line {section['line']}: [PARSER] has Time_Key but missing Time_Format"
                    )

        # MULTILINE_PARSER-specific validation
        elif section_type == "MULTILINE_PARSER":
            if "type" not in params:
                self.errors.append(
                    f"Line {section['line']}: [MULTILINE_PARSER] missing required parameter 'Type'"
                )
            else:
                multiline_type = params["type"]["value"].lower()
                valid_types = ["regex"]
                if multiline_type not in valid_types:
                    self.errors.append(
                        f"Line {params['type']['line']}: [MULTILINE_PARSER] invalid Type '{multiline_type}' "
                        f"(valid: {', '.join(valid_types)})"
                    )

            # Check for rule definitions (keys are already lowercased)
            has_rule = any(key.startswith("rule") for key in params.keys())
            if not has_rule:
                self.errors.append(
                    f"Line {section['line']}: [MULTILINE_PARSER] missing 'rule' definitions"
                )

            # Recommend flush_timeout
            if "flush_timeout" not in params:
                self.recommendations.append(
                    f"Line {section['line']}: [MULTILINE_PARSER] consider setting flush_timeout (e.g., 1000ms)"
                )

    def validate_tags(self) -> None:
        """Validate tag consistency across INPUT, FILTER, OUTPUT."""
        input_tags = []
        for section in self.sections:
            if section["type"] == "INPUT":
                if "tag" in section["params"]:
                    input_tags.append(section["params"]["tag"]["value"])
        if not input_tags:
            return

        produced_tags = set(input_tags)

        # Process filters in order to simulate tag flow and rewrite_tag emissions.
        for section in self.sections:
            if section["type"] != "FILTER":
                continue

            params = section["params"]
            descriptor = self._match_descriptor(params)

            if not self._section_matches_any_tags(params, produced_tags, section, "FILTER"):
                if descriptor:
                    self.warnings.append(
                        f"Line {section['line']}: [FILTER] {descriptor} doesn't match any INPUT/FILTER tags"
                    )
                continue

            filter_name = params.get("name", {}).get("value", "").lower()
            if filter_name == "rewrite_tag":
                generated_patterns = self._extract_rewrite_tag_patterns(section)
                produced_tags.update(generated_patterns)

        # Validate outputs against tags produced by inputs and filters.
        for section in self.sections:
            if section["type"] != "OUTPUT":
                continue

            params = section["params"]
            descriptor = self._match_descriptor(params)
            if not self._section_matches_any_tags(params, produced_tags, section, "OUTPUT"):
                if descriptor:
                    self.warnings.append(
                        f"Line {section['line']}: [OUTPUT] {descriptor} doesn't match any INPUT/FILTER tags"
                    )

    def _extract_rewrite_tag_patterns(self, section: Dict) -> Set[str]:
        """Extract tags emitted by rewrite_tag filters from Rule entries."""
        params = section["params"]
        generated = set()

        param_entries = section.get("param_entries")
        if not param_entries:
            # Backward compatibility for section objects created in tests/tools.
            param_entries = [
                {"key": key, "value": meta["value"], "line": meta["line"]}
                for key, meta in params.items()
            ]

        for entry in param_entries:
            key = entry["key"]
            if not key.startswith("rule"):
                continue

            rule = entry["value"].strip()
            try:
                parts = shlex.split(rule)
            except ValueError as exc:
                self.warnings.append(
                    f"Line {entry['line']}: [FILTER rewrite_tag] invalid Rule syntax: {exc}"
                )
                continue

            # Rule format: $KEY REGEX NEW_TAG KEEP [AND_COMBINE]
            if len(parts) < 4:
                self.warnings.append(
                    f"Line {entry['line']}: [FILTER rewrite_tag] Rule should be "
                    f"'$KEY REGEX NEW_TAG KEEP [AND_COMBINE]'"
                )
                continue

            new_tag = parts[2]
            generated.add("*" if "$" in new_tag else new_tag)

        return generated

    def _section_matches_any_tags(
        self, params: Dict, tags: Set[str], section: Dict, section_type: str
    ) -> bool:
        """Check if Match/Match_Regex for a section matches any known tags."""
        has_match = False
        has_regex = False

        if "match" in params:
            has_match = True
            match_pattern = params["match"]["value"]

            if match_pattern == "*":
                return True

            for tag in tags:
                if self._tag_patterns_overlap(tag, match_pattern):
                    return True

        if "match_regex" in params:
            has_regex = True
            regex_pattern = params["match_regex"]["value"]
            try:
                regex = re.compile(regex_pattern)
            except re.error as exc:
                self.errors.append(
                    f"Line {params['match_regex']['line']}: [{section_type}] "
                    f"invalid Match_Regex '{regex_pattern}': {exc}"
                )
                return False

            overlap_unknown = False
            for tag in tags:
                overlap = self._tag_pattern_vs_regex(tag, regex)
                if overlap is True:
                    return True
                if overlap is None:
                    overlap_unknown = True

            if overlap_unknown:
                descriptor = self._match_descriptor(params)
                if descriptor:
                    self.recommendations.append(
                        f"Line {section['line']}: [{section_type}] {descriptor} overlap with wildcard tags "
                        f"is ambiguous; verify with Fluent Bit dry-run if strict routing is required"
                    )
                return True

        return not has_match and not has_regex

    def _tag_pattern_vs_regex(
        self, tag_pattern: str, regex: re.Pattern
    ) -> Optional[bool]:
        """
        Compare a wildcard tag pattern against Match_Regex.

        Returns:
            True: overlap found
            False: provably disjoint
            None: overlap unknown
        """
        if "*" not in tag_pattern:
            return regex.match(tag_pattern) is not None

        for candidate in self._sample_tags_from_pattern(tag_pattern):
            if regex.match(candidate):
                return True

        return None

    def _sample_tags_from_pattern(self, pattern: str) -> List[str]:
        """Generate representative sample tags from wildcard patterns."""
        if "*" not in pattern:
            return [pattern]

        samples = []
        for token in ["sample", "x", "0", "123", "prod-1", "_", ""]:
            sample = pattern.replace("*", token)
            if sample not in samples:
                samples.append(sample)

        return samples

    def _tag_patterns_overlap(self, left: str, right: str) -> bool:
        """Approximate overlap check for two wildcard-style tag patterns."""
        if left == "*" or right == "*":
            return True

        for sample in self._sample_tags_from_pattern(left):
            if self._tag_matches(sample, right):
                return True

        for sample in self._sample_tags_from_pattern(right):
            if self._tag_matches(sample, left):
                return True

        return False

    def _match_descriptor(self, params: Dict):
        """Return human-friendly descriptor for Match or Match_Regex."""
        if "match" in params:
            return f"Match pattern '{params['match']['value']}'"
        if "match_regex" in params:
            return f"Match_Regex pattern '{params['match_regex']['value']}'"
        return None

    def _tag_matches(self, tag: str, pattern: str) -> bool:
        """Check if tag matches pattern (with wildcard support)."""
        if pattern == "*":
            return True

        # Convert wildcard pattern to regex
        regex_pattern = pattern.replace(".", r"\.").replace("*", ".*")
        return re.match(f"^{regex_pattern}$", tag) is not None

    def validate_security(self) -> None:
        """Security audit of configuration."""
        for section in self.sections:
            params = section["params"]

            # Check for hardcoded credentials (keys are stored lowercase)
            sensitive_keys = [
                "http_user",
                "http_passwd",
                "password",
                "aws_access_key",
                "aws_secret_key",
                "secret",
                "api_key",
                "token",
            ]
            # Display names for human-readable messages
            sensitive_key_display = {
                "http_user": "HTTP_User",
                "http_passwd": "HTTP_Passwd",
                "password": "Password",
                "aws_access_key": "AWS_Access_Key",
                "aws_secret_key": "AWS_Secret_Key",
                "secret": "Secret",
                "api_key": "API_Key",
                "token": "Token",
            }
            for key in sensitive_keys:
                if key in params:
                    value = params[key]["value"]
                    # Check if it's an environment variable reference
                    if not value.startswith("${"):
                        display = sensitive_key_display[key]
                        self.warnings.append(
                            f"Line {params[key]['line']}: Hardcoded credential '{display}' "
                            f"(use environment variable: ${{{display}}})"
                        )

            # Check TLS configuration
            if section["type"] == "OUTPUT":
                if "tls" in params:
                    tls_value = params["tls"]["value"].lower()
                    if tls_value in ["off", "false", "no"]:
                        self.warnings.append(
                            f"Line {params['tls']['line']}: TLS disabled (security risk in production)"
                        )

                if "tls.verify" in params:
                    verify_value = params["tls.verify"]["value"].lower()
                    if verify_value in ["off", "false", "no"]:
                        self.warnings.append(
                            f"Line {params['tls.verify']['line']}: TLS verification disabled (MITM risk)"
                        )

            # Check network exposure in SERVICE HTTP server
            if section["type"] == "SERVICE":
                http_server_on = params.get("http_server", {}).get("value", "").lower() in [
                    "on",
                    "true",
                    "yes",
                ]
                if http_server_on and params.get("http_listen", {}).get("value") == "0.0.0.0":
                    self.warnings.append(
                        f"Line {params['http_listen']['line']}: HTTP_Server exposed on 0.0.0.0 "
                        f"(limit to internal interface in production)"
                    )

            # Check network listener exposure for INPUT network plugins
            if section["type"] == "INPUT":
                plugin_name = params.get("name", {}).get("value", "").lower()
                network_inputs = {"http", "tcp", "udp", "forward", "syslog"}
                if plugin_name in network_inputs:
                    listener = params.get("listen", params.get("host"))
                    if listener and listener["value"] == "0.0.0.0":
                        self.warnings.append(
                            f"Line {listener['line']}: [INPUT {plugin_name}] listening on 0.0.0.0 "
                            f"(ensure network controls and authentication are in place)"
                        )

    def validate_performance(self) -> None:
        """Analyze performance configuration."""
        for section in self.sections:
            params = section["params"]

            # Check tail input buffer limits
            if section["type"] == "INPUT" and params.get("name", {}).get("value", "").lower() == "tail":
                if "mem_buf_limit" in params:
                    buf_limit = params["mem_buf_limit"]["value"]
                    # Parse size (e.g., "50MB", "1GB", "512" where unit defaults to bytes)
                    size_match = re.match(r"^(\d+(?:\.\d+)?)\s*(MB|GB|KB|M|G|K|B)?$", buf_limit, re.IGNORECASE)
                    if size_match:
                        size = float(size_match.group(1))
                        unit = (size_match.group(2) or "B").upper()

                        # Normalize unit names (M -> MB, G -> GB, K -> KB)
                        if unit == "M":
                            unit = "MB"
                        elif unit == "G":
                            unit = "GB"
                        elif unit == "K":
                            unit = "KB"

                        # Convert to MB
                        if unit == "B":
                            size_mb = size / (1024 * 1024)
                        elif unit == "KB":
                            size_mb = size / 1024
                        elif unit == "GB":
                            size_mb = size * 1024
                        else:
                            size_mb = size

                        if size_mb < 10:
                            self.warnings.append(
                                f"Line {params['mem_buf_limit']['line']}: Mem_Buf_Limit < 10MB (may cause backpressure)"
                            )
                        elif size_mb > 500:
                            self.warnings.append(
                                f"Line {params['mem_buf_limit']['line']}: Mem_Buf_Limit > 500MB (high memory usage)"
                            )
                    else:
                        self.errors.append(
                            f"Line {params['mem_buf_limit']['line']}: Invalid Mem_Buf_Limit format '{buf_limit}' "
                            f"(expected format: number with optional unit KB/MB/GB)"
                        )

            # Check OUTPUT storage limits
            if section["type"] == "OUTPUT":
                if "storage.total_limit_size" not in params:
                    self.recommendations.append(
                        f"Line {section['line']}: [OUTPUT] consider setting storage.total_limit_size"
                    )

    def validate_best_practices(self) -> None:
        """Check best practices."""
        has_http_server = False
        has_storage_metrics = False
        has_exclude_path_for_k8s = False
        is_kubernetes_setup = False
        tail_inputs = 0
        tail_inputs_with_db = 0
        tail_inputs_with_mem_buf_limit = 0
        total_outputs = 0
        outputs_with_retry_limit = 0

        for section in self.sections:
            section_type = section["type"]
            params = section["params"]

            # SERVICE section checks
            if section_type == "SERVICE":
                if "http_server" in params:
                    value = params["http_server"]["value"].lower()
                    if value in ["on", "true", "yes"]:
                        has_http_server = True

                if "storage.metrics" in params:
                    value = params["storage.metrics"]["value"].lower()
                    if value in ["on", "true", "yes"]:
                        has_storage_metrics = True

            # INPUT section checks
            elif section_type == "INPUT":
                if params.get("name", {}).get("value", "").lower() == "tail":
                    tail_inputs += 1

                    # Check for DB parameter
                    if "db" in params:
                        tail_inputs_with_db += 1

                    # Check Mem_Buf_Limit
                    if "mem_buf_limit" in params:
                        tail_inputs_with_mem_buf_limit += 1

                    # Check for Kubernetes setup
                    if "path" in params:
                        path = params["path"]["value"]
                        if "/var/log/containers" in path or "kube" in path.lower():
                            is_kubernetes_setup = True

                            # Check Exclude_Path for Kubernetes
                            if "exclude_path" in params:
                                exclude = params["exclude_path"]["value"]
                                if "fluent-bit" in exclude or "fluentbit" in exclude:
                                    has_exclude_path_for_k8s = True

            # OUTPUT section checks
            elif section_type == "OUTPUT":
                total_outputs += 1
                if "retry_limit" in params:
                    outputs_with_retry_limit += 1

        # Generate best practice recommendations
        if not has_http_server:
            self.recommendations.append(
                "Consider enabling HTTP_Server for health checks and metrics (SERVICE: HTTP_Server On)"
            )

        if not has_storage_metrics:
            self.recommendations.append(
                "Consider enabling storage metrics for monitoring (SERVICE: storage.metrics on)"
            )

        if tail_inputs > 0 and tail_inputs_with_db < tail_inputs:
            self.recommendations.append(
                "Consider adding DB parameter to all tail INPUTs for crash recovery and offset tracking"
            )

        if tail_inputs > 0 and tail_inputs_with_mem_buf_limit < tail_inputs:
            self.recommendations.append(
                "Consider adding Mem_Buf_Limit to all tail INPUTs to avoid unbounded memory usage"
            )

        if total_outputs > 0 and outputs_with_retry_limit < total_outputs:
            self.recommendations.append(
                "Consider setting Retry_Limit on all OUTPUTs to avoid infinite retry loops"
            )

        if is_kubernetes_setup:
            if not has_exclude_path_for_k8s:
                self.recommendations.append(
                    "Consider excluding Fluent Bit's own logs to prevent loops (INPUT tail: Exclude_Path *fluent-bit*.log)"
                )

            # Check for kubernetes filter
            has_k8s_filter = any(
                section["type"] == "FILTER" and
                section["params"].get("name", {}).get("value", "").lower() == "kubernetes"
                for section in self.sections
            )

            if not has_k8s_filter:
                self.recommendations.append(
                    "Consider adding kubernetes FILTER for metadata enrichment in Kubernetes environments"
                )

    def validate_dry_run(self) -> None:
        """Test configuration with fluent-bit --dry-run if binary is available."""
        # Check if fluent-bit binary is available
        fluent_bit_path = shutil.which("fluent-bit")

        if not fluent_bit_path:
            message = (
                "Dry-run skipped because fluent-bit binary is not available in PATH; "
                "run dry-run in CI or a Fluent Bit runtime image."
            )
            if self.require_dry_run:
                self.errors.append(message)
            else:
                self.recommendations.append(message)
            return

        # Get absolute path to config file
        config_abs_path = os.path.abspath(self.config_file)

        try:
            # Run fluent-bit with --dry-run flag
            # --dry-run: Test configuration and exit
            result = subprocess.run(
                [fluent_bit_path, "-c", config_abs_path, "--dry-run"],
                capture_output=True,
                text=True,
                timeout=10,  # 10 second timeout
                check=False,  # Don't raise exception on non-zero exit
            )

            # Parse output for errors
            if result.returncode != 0:
                # Dry-run failed - parse error messages
                error_lines = []

                # Check both stdout and stderr
                for line in (result.stdout + result.stderr).splitlines():
                    line_lower = line.lower()

                    # Look for error indicators
                    if any(indicator in line_lower for indicator in ["error", "fail", "invalid", "cannot"]):
                        error_lines.append(line.strip())

                if error_lines:
                    self.errors.append(
                        f"Dry-run test failed:\n  " + "\n  ".join(error_lines[:5])  # Limit to first 5 errors
                    )
                else:
                    self.errors.append(
                        f"Dry-run test failed with exit code {result.returncode}"
                    )
            else:
                # Dry-run succeeded
                self.recommendations.append("Dry-run test passed - configuration is valid")

                # Check for warnings in output
                warning_lines = []
                for line in (result.stdout + result.stderr).splitlines():
                    line_lower = line.lower()
                    if "warn" in line_lower:
                        warning_lines.append(line.strip())

                if warning_lines:
                    for warning in warning_lines[:3]:  # Limit to first 3 warnings
                        self.recommendations.append(f"Dry-run warning: {warning}")

        except subprocess.TimeoutExpired:
            self.warnings.append(
                "Dry-run test timed out after 10 seconds (configuration may have issues)"
            )
        except Exception as e:
            self.warnings.append(
                f"Dry-run test failed with exception: {str(e)}"
            )

    def print_report(self) -> None:
        """Print validation report."""
        print(f"\nValidation Report: {self.config_file}")

        if self.errors:
            print("\nError:")
            for error in self.errors:
                print(f"  - {error}")

        if self.warnings:
            print("\nWarning:")
            for warning in self.warnings:
                print(f"  - {warning}")

        if self.recommendations:
            print("\nRecommendation:")
            for recommendation in self.recommendations:
                print(f"  - {recommendation}")

        if not self.errors and not self.warnings and not self.recommendations:
            print("\nRecommendation:")
            print("  - No findings.")

        print()

    def get_summary(self, fail_on_warning: bool = False) -> Dict:
        """Get validation summary as dict."""
        return {
            "file": self.config_file,
            "valid": len(self.errors) == 0 and (not fail_on_warning or len(self.warnings) == 0),
            "errors": self.errors,
            "warnings": self.warnings,
            "recommendations": self.recommendations,
        }


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Validate Fluent Bit configuration files",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument(
        "--file",
        required=True,
        help="Path to Fluent Bit configuration file",
    )

    parser.add_argument(
        "--check",
        default="all",
        choices=[
            "all",
            "structure",
            "syntax",
            "sections",
            "tags",
            "security",
            "performance",
            "best-practices",
            "dry-run",
        ],
        help="Validation check to perform (default: all)",
    )

    parser.add_argument(
        "--json",
        action="store_true",
        help="Output results as JSON",
    )

    parser.add_argument(
        "--fail-on-warning",
        action="store_true",
        help="Return non-zero exit code when warnings are present",
    )
    parser.add_argument(
        "--require-dry-run",
        action="store_true",
        help="Treat missing fluent-bit binary as an error for dry-run checks",
    )

    args = parser.parse_args()

    # Validate configuration
    validator = FluentBitValidator(args.file, require_dry_run=args.require_dry_run)

    if args.check == "all":
        validator.validate_all()
    elif args.check == "structure":
        validator.validate_structure()
    elif args.check == "syntax":
        # Parse config first, then check syntax
        validator.validate_structure()
        validator.validate_syntax()
    elif args.check == "sections":
        # Parse config first, then validate sections
        validator.validate_structure()
        validator.validate_sections()
    elif args.check == "tags":
        # Parse config first, then check tag consistency
        validator.validate_structure()
        validator.validate_tags()
    elif args.check == "security":
        # Parse config first, then run security audit
        validator.validate_structure()
        validator.validate_security()
    elif args.check == "performance":
        # Parse config first, then analyze performance
        validator.validate_structure()
        validator.validate_performance()
    elif args.check == "best-practices":
        # Parse config first, then check best practices
        validator.validate_structure()
        validator.validate_best_practices()
    elif args.check == "dry-run":
        # Parse config first, then run dry-run test
        validator.validate_structure()
        validator.validate_dry_run()

    # Output results
    if args.json:
        print(json.dumps(validator.get_summary(args.fail_on_warning), indent=2))
    else:
        validator.print_report()

    # Exit with error code if validation failed
    has_failures = len(validator.errors) > 0 or (
        args.fail_on_warning and len(validator.warnings) > 0
    )
    sys.exit(0 if not has_failures else 1)


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
#
# Fluent Bit Config Validator - Convenience Wrapper Script
#
# This script provides a simpler interface to validate_config.py
# Usage: bash validate.sh [--precheck] <config-file> [--check <type>] [--json]
#
# Examples:
#   bash validate.sh --precheck
#   bash validate.sh fluent-bit.conf
#   bash validate.sh fluent-bit.conf --check security
#   bash validate.sh fluent-bit.conf --json
#

set -euo pipefail

# Get the directory where this script is located
SCRIPT_PATH="${BASH_SOURCE[0]}"
SCRIPT_DIR_REL="${SCRIPT_PATH%/*}"
if [ "${SCRIPT_DIR_REL}" = "${SCRIPT_PATH}" ]; then
    SCRIPT_DIR_REL="."
fi
SCRIPT_DIR="$(cd "${SCRIPT_DIR_REL}" && pwd)"

# Path to the Python validation script
VALIDATOR_SCRIPT="${SCRIPT_DIR}/validate_config.py"

print_help() {
    echo "Fluent Bit Config Validator"
    echo ""
    echo "Usage: $0 [--precheck] <config-file> [options]"
    echo "       $0 --precheck"
    echo ""
    echo "Options:"
    echo "  --precheck            Print binary availability and precheck status"
    echo "  --file <path>         Path to Fluent Bit config file (can be first argument)"
    echo "  --check <type>        Run specific check type:"
    echo "                        structure, syntax, sections, tags, security,"
    echo "                        performance, best-practices, dry-run, all"
    echo "  --json                Output results in JSON format"
    echo "  --fail-on-warning     Exit non-zero when warnings are present"
    echo "  --require-dry-run     Treat missing fluent-bit binary as an error"
    echo "  --help                Show this help message"
    echo ""
    echo "Examples:"
    echo "  $0 --precheck"
    echo "  $0 fluent-bit.conf"
    echo "  $0 fluent-bit.conf --check security"
    echo "  $0 fluent-bit.conf --check all --json"
    echo "  $0 --precheck --file fluent-bit.conf --check all"
    echo ""
}

run_precheck() {
    local precheck_failed=0

    if command -v python3 >/dev/null 2>&1; then
        echo "python3: available"
    else
        echo "python3: missing"
        echo "Error: python3 is required for script-based validation; use manual review until python3 is installed."
        precheck_failed=1
    fi

    if command -v fluent-bit >/dev/null 2>&1; then
        echo "fluent-bit: available"
    else
        echo "fluent-bit: missing (dry-run will be skipped)"
        echo "Recommendation: Dry-run skipped because fluent-bit binary is not available in PATH; run dry-run in CI or a Fluent Bit runtime image."
    fi

    return "${precheck_failed}"
}

# Show help if no arguments provided
if [ $# -eq 0 ]; then
    print_help
    exit 0
fi

RUN_PRECHECK=false
FORWARD_ARGS=()
for arg in "$@"; do
    case "${arg}" in
        --precheck)
            RUN_PRECHECK=true
            ;;
        --help|-h)
            print_help
            exit 0
            ;;
        *)
            FORWARD_ARGS+=("${arg}")
            ;;
    esac
done

if [ "${RUN_PRECHECK}" = true ]; then
    run_precheck || exit 1
    if [ "${#FORWARD_ARGS[@]}" -eq 0 ]; then
        exit 0
    fi
fi

# Check if Python 3 is available before invoking the validator
if ! command -v python3 >/dev/null 2>&1; then
    echo "Error: python3 is not installed or not in PATH"
    echo "Please install Python 3 to use this validator"
    exit 1
fi

# Check if validator script exists
if [ ! -f "${VALIDATOR_SCRIPT}" ]; then
    echo "Error: validator script not found at ${VALIDATOR_SCRIPT}"
    exit 1
fi

# If first argument doesn't start with --, treat it as the config file
if [[ "${FORWARD_ARGS[0]}" != --* ]]; then
    CONFIG_FILE="${FORWARD_ARGS[0]}"
    if [ "${#FORWARD_ARGS[@]}" -gt 1 ]; then
        python3 "${VALIDATOR_SCRIPT}" --file "${CONFIG_FILE}" "${FORWARD_ARGS[@]:1}"
    else
        python3 "${VALIDATOR_SCRIPT}" --file "${CONFIG_FILE}"
    fi
else
    # Otherwise pass all arguments to the Python script
    python3 "${VALIDATOR_SCRIPT}" "${FORWARD_ARGS[@]}"
fi

# Invalid configuration - missing required parameters

[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    # Missing Name parameter
    Tag    test.*

[FILTER]
    # Missing Name and Match parameters
    Kube_URL   https://kubernetes.default.svc:443

[OUTPUT]
    # Missing Name and Match parameters
    Host    elasticsearch.default.svc

# Invalid Fluent Bit configuration - OpenTelemetry issues

[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    Name    tail
    Path    /var/log/*.log
    Tag     app.*

[OUTPUT]
    Name            opentelemetry
    Match           *
    # Missing Host parameter (required)
    Port            99999
    Header          Authorization Bearer hardcoded-token-123
    tls             Off
    tls.verify      Off

# Invalid configuration - security issues

[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    Name    tail
    Path    /var/log/*.log
    Tag     app.*

[OUTPUT]
    Name            es
    Match           *
    Host            elasticsearch.default.svc
    Port            9200
    HTTP_User       admin
    HTTP_Passwd     hardcoded-password-123
    tls             Off
    Retry_Limit     5

# Invalid configuration - tag mismatches

[SERVICE]
    Flush        5
    Log_Level    info

[INPUT]
    Name    tail
    Path    /var/log/app.log
    Tag     app.logs

[FILTER]
    Name     grep
    Match    database.*
    Regex    level ERROR

[OUTPUT]
    Name         stdout
    Match        nonexistent.*
    Format       json
    Retry_Limit  3

#!/usr/bin/env python3
"""Automated tests for Fluent Bit validator regressions and core behavior."""

import json
import os
import subprocess
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path


SKILL_DIR = Path(__file__).resolve().parent.parent
VALIDATOR = SKILL_DIR / "scripts" / "validate_config.py"


class ValidatorTestCase(unittest.TestCase):
    """Behavioral tests for validate_config.py."""

    def run_validator(
        self,
        config_text,
        check="all",
        fail_on_warning=False,
        require_dry_run=False,
        env=None,
    ):
        """Run validator against temporary config and return (proc, summary)."""
        with tempfile.NamedTemporaryFile(mode="w", suffix=".conf", delete=False) as config_file:
            config_file.write(textwrap.dedent(config_text).strip() + "\n")
            config_path = config_file.name

        cmd = [
            sys.executable,
            str(VALIDATOR),
            "--file",
            config_path,
            "--check",
            check,
            "--json",
        ]
        if fail_on_warning:
            cmd.append("--fail-on-warning")
        if require_dry_run:
            cmd.append("--require-dry-run")

        run_env = None
        if env is not None:
            run_env = os.environ.copy()
            run_env.update(env)

        proc = subprocess.run(cmd, capture_output=True, text=True, check=False, env=run_env)
        summary = json.loads(proc.stdout)

        Path(config_path).unlink(missing_ok=True)
        return proc, summary

    def run_validator_file(self, config_path, check="all", env=None):
        """Run validator against a checked-in config file and return (proc, summary)."""
        cmd = [
            sys.executable,
            str(VALIDATOR),
            "--file",
            str(config_path),
            "--check",
            check,
            "--json",
        ]

        run_env = None
        if env is not None:
            run_env = os.environ.copy()
            run_env.update(env)

        proc = subprocess.run(cmd, capture_output=True, text=True, check=False, env=run_env)
        summary = json.loads(proc.stdout)
        return proc, summary

    def test_single_space_and_equals_delimiters_are_supported(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
            Flush=5
            Log_Level info

            [INPUT]
            Name tail
            Path /var/log/*.log
            Tag app.logs
            Mem_Buf_Limit 50MB
            DB /tmp/flb.db
            Skip_Long_Lines On

            [OUTPUT]
            Name stdout
            Match *
            Retry_Limit 3
            """,
            check="sections",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertTrue(summary["valid"])
        self.assertEqual(summary["errors"], [])

    def test_structure_reports_malformed_key_value_pairs(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
            Flush 5
            InvalidLineWithoutDelimiter
            """,
            check="structure",
        )
        self.assertNotEqual(proc.returncode, 0)
        self.assertFalse(summary["valid"])
        self.assertTrue(
            any("Malformed key-value pair" in error for error in summary["errors"])
        )

    def test_tag_check_supports_match_regex(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs

            [FILTER]
                Name grep
                Match_Regex ^app\\..*$
                Regex level ERROR

            [OUTPUT]
                Name stdout
                Match_Regex ^app\\..*$
                Retry_Limit 3
            """,
            check="tags",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertFalse(
            any("doesn't match any INPUT" in warning for warning in summary["warnings"])
        )

    def test_parser_handles_equals_in_whitespace_delimited_values(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
                Mem_Buf_Limit 50MB
                DB /tmp/flb.db

            [FILTER]
                Name grep
                Match app.*
                Regex level ^foo=bar$

            [OUTPUT]
                Name stdout
                Match *
                Retry_Limit 3
            """,
            check="sections",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertFalse(
            any("has neither Regex nor Exclude" in warning for warning in summary["warnings"])
        )

    def test_tag_check_handles_rewrite_tag_generated_tags(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.raw

            [FILTER]
                Name rewrite_tag
                Match app.raw
                Rule $log ^.*$ app.processed false

            [OUTPUT]
                Name stdout
                Match app.processed
                Retry_Limit 3
            """,
            check="tags",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertFalse(
            any("app.processed" in warning for warning in summary["warnings"])
        )

    def test_tag_check_handles_non_final_rewrite_tag_rule(self):
        """OUTPUT should match tags generated by any rewrite_tag Rule, not only the last one."""
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.raw

            [FILTER]
                Name rewrite_tag
                Match app.raw
                Rule $log ^ERROR$ app.error false
                Rule $log ^INFO$ app.info false

            [OUTPUT]
                Name stdout
                Match app.error
                Retry_Limit 3
            """,
            check="tags",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertFalse(
            any("app.error" in warning and "doesn't match any INPUT/FILTER tags" in warning
                for warning in summary["warnings"])
        )

    def test_tag_check_wildcard_input_tag_can_overlap_with_match_regex(self):
        """Wildcard tag patterns should not produce false Match_Regex mismatch warnings."""
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.*

            [OUTPUT]
                Name stdout
                Match_Regex ^app\\.[0-9]+$
                Retry_Limit 3
            """,
            check="tags",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertFalse(
            any("^app\\.[0-9]+$" in warning and "doesn't match any INPUT/FILTER tags" in warning
                for warning in summary["warnings"])
        )

    def test_tag_check_true_match_regex_mismatch_still_warns(self):
        """Provable Match_Regex mismatches should still be reported."""
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs

            [OUTPUT]
                Name stdout
                Match_Regex ^infra\\..*$
                Retry_Limit 3
            """,
            check="tags",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertTrue(
            any("^infra\\..*$" in warning and "doesn't match any INPUT/FILTER tags" in warning
                for warning in summary["warnings"])
        )

    def test_unknown_output_plugin_is_reported(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
                Mem_Buf_Limit 50MB
                DB /tmp/flb.db

            [OUTPUT]
                Name madeup_output
                Match *
                Retry_Limit 3
            """,
            check="sections",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertTrue(
            any("Unknown OUTPUT plugin" in warning for warning in summary["warnings"])
        )

    def test_best_practices_reports_retry_db_and_mem_buf_gaps(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
                HTTP_Server On
                storage.metrics on

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs

            [OUTPUT]
                Name stdout
                Match *
            """,
            check="best-practices",
        )
        self.assertEqual(proc.returncode, 0)
        self.assertTrue(
            any(
                "DB parameter to all tail INPUTs" in recommendation
                for recommendation in summary["recommendations"]
            )
        )
        self.assertTrue(
            any(
                "Mem_Buf_Limit to all tail INPUTs" in recommendation
                for recommendation in summary["recommendations"]
            )
        )
        self.assertTrue(
            any(
                "Retry_Limit on all OUTPUTs" in recommendation
                for recommendation in summary["recommendations"]
            )
        )

    def test_fail_on_warning_changes_exit_code_and_valid_field(self):
        config = """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs

            [OUTPUT]
                Name es
                Match *
                Host elasticsearch.default.svc
                HTTP_Passwd hardcoded-password
                tls Off
                Retry_Limit 3
            """

        normal_proc, normal_summary = self.run_validator(config, check="security")
        strict_proc, strict_summary = self.run_validator(
            config, check="security", fail_on_warning=True
        )

        self.assertEqual(normal_proc.returncode, 0)
        self.assertTrue(normal_summary["valid"])
        self.assertNotEqual(strict_proc.returncode, 0)
        self.assertFalse(strict_summary["valid"])

    def test_missing_fluent_bit_is_recommendation_by_default(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
                Mem_Buf_Limit 50MB
                DB /tmp/flb.db

            [OUTPUT]
                Name stdout
                Match *
                Retry_Limit 3
            """,
            check="dry-run",
            env={"PATH": "/nonexistent"},
        )

        self.assertEqual(proc.returncode, 0)
        self.assertEqual(summary["errors"], [])
        self.assertTrue(
            any(
                "Dry-run skipped because fluent-bit binary is not available in PATH"
                in recommendation
                for recommendation in summary["recommendations"]
            )
        )

    def test_require_dry_run_escalates_missing_binary_to_error(self):
        proc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5

            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
                Mem_Buf_Limit 50MB
                DB /tmp/flb.db

            [OUTPUT]
                Name stdout
                Match *
                Retry_Limit 3
            """,
            check="dry-run",
            require_dry_run=True,
            env={"PATH": "/nonexistent"},
        )

        self.assertNotEqual(proc.returncode, 0)
        self.assertTrue(
            any(
                "Dry-run skipped because fluent-bit binary is not available in PATH"
                in error
                for error in summary["errors"]
            )
        )

    def test_text_report_uses_recommendation_label(self):
        with tempfile.NamedTemporaryFile(mode="w", suffix=".conf", delete=False) as config_file:
            config_file.write(
                textwrap.dedent(
                    """
                    [SERVICE]
                        Flush 5

                    [INPUT]
                        Name tail
                        Path /var/log/*.log
                        Tag app.logs
                        Mem_Buf_Limit 50MB
                        DB /tmp/flb.db

                    [OUTPUT]
                        Name stdout
                        Match *
                        Retry_Limit 3
                    """
                ).strip()
                + "\n"
            )
            config_path = config_file.name

        cmd = [
            sys.executable,
            str(VALIDATOR),
            "--file",
            config_path,
            "--check",
            "dry-run",
        ]
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=False,
            env={**os.environ, "PATH": "/nonexistent"},
        )
        Path(config_path).unlink(missing_ok=True)

        self.assertEqual(proc.returncode, 0)
        self.assertIn("Recommendation:", proc.stdout)
        self.assertNotIn("Info", proc.stdout)

    def test_valid_basic_sample_is_clean_static_baseline(self):
        """The shipped valid sample should only emit the environment-driven dry-run recommendation."""
        proc, summary = self.run_validator_file(
            SKILL_DIR / "tests" / "valid-basic.conf",
            check="all",
            env={"PATH": "/nonexistent"},
        )

        self.assertEqual(proc.returncode, 0)
        self.assertEqual(summary["errors"], [])
        self.assertEqual(summary["warnings"], [])
        self.assertEqual(
            summary["recommendations"],
            [
                "Dry-run skipped because fluent-bit binary is not available in PATH; run dry-run in CI or a Fluent Bit runtime image."
            ],
        )


    # -------------------------------------------------------------------------
    # Regression tests for Bug 1: case-insensitive param key lookups
    # -------------------------------------------------------------------------

    def test_security_detects_tls_uppercase(self):
        """TLS Off written with uppercase key must still trigger a security warning."""
        _, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name es
                Match *
                Host es.example.com
                TLS Off
                Retry_Limit 3
            """,
            check="security",
        )
        self.assertTrue(
            any("TLS disabled" in w for w in summary["warnings"]),
            "Expected TLS warning for uppercase 'TLS Off'",
        )

    def test_security_detects_credential_lowercase_key(self):
        """Hardcoded credential written with lowercase key must still be detected."""
        _, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name es
                Match *
                Host es.example.com
                http_passwd mysecret
                Retry_Limit 3
            """,
            check="security",
        )
        self.assertTrue(
            any("HTTP_Passwd" in w for w in summary["warnings"]),
            "Expected credential warning for lowercase 'http_passwd'",
        )

    def test_s3_output_accepts_capitalized_params(self):
        """S3 output with Bucket/Region (capitalized) must not produce false errors."""
        rc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name s3
                Match *
                Bucket my-bucket
                Region us-east-1
                Retry_Limit 3
            """,
            check="sections",
        )
        self.assertFalse(
            any("missing required parameter 'bucket'" in e for e in summary["errors"]),
            "Capitalized 'Bucket' should be accepted for S3 output",
        )
        self.assertFalse(
            any("missing required parameter 'region'" in e for e in summary["errors"]),
            "Capitalized 'Region' should be accepted for S3 output",
        )

    def test_cloudwatch_accepts_capitalized_log_group_name(self):
        """CloudWatch with Log_Group_Name (capitalized) must not produce a false error."""
        _, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name cloudwatch_logs
                Match *
                region us-east-1
                Log_Group_Name /my/log/group
                Retry_Limit 3
            """,
            check="sections",
        )
        self.assertFalse(
            any("missing required parameter 'log_group_name'" in e for e in summary["errors"]),
            "Capitalized 'Log_Group_Name' should be accepted for CloudWatch output",
        )

    # -------------------------------------------------------------------------
    # Regression test for Bug 2: duplicate TLS warnings for OpenTelemetry
    # -------------------------------------------------------------------------

    def test_opentelemetry_no_duplicate_tls_warnings(self):
        """TLS Off on an OpenTelemetry output must produce exactly one TLS warning."""
        _, summary = self.run_validator(
            """
            [SERVICE]
                Flush 5
                HTTP_Server On
                storage.metrics on
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
                Mem_Buf_Limit 50MB
                DB /tmp/flb.db
            [OUTPUT]
                Name opentelemetry
                Match *
                Host otel-collector.example.com
                tls Off
                Retry_Limit 3
            """,
            check="all",
        )
        tls_warnings = [w for w in summary["warnings"] if "TLS disabled" in w]
        self.assertEqual(
            len(tls_warnings),
            1,
            f"Expected exactly one TLS warning, got {len(tls_warnings)}: {tls_warnings}",
        )

    # -------------------------------------------------------------------------
    # Regression test for Bug 3: Flush float false positive error
    # -------------------------------------------------------------------------

    def test_flush_float_is_not_an_error(self):
        """Flush 0.5 (valid sub-second float) must not produce an error."""
        rc, summary = self.run_validator(
            """
            [SERVICE]
                Flush 0.5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name stdout
                Match *
            """,
            check="sections",
        )
        self.assertFalse(
            any("Flush must be a number" in e for e in summary["errors"]),
            "Flush 0.5 should not raise a parse error",
        )
        self.assertTrue(
            any("Flush interval < 1 second" in w for w in summary["warnings"]),
            "Flush 0.5 should produce a sub-second performance warning",
        )

    # -------------------------------------------------------------------------
    # Regression tests for Gap 1: @INCLUDE and @SET directives
    # -------------------------------------------------------------------------

    def test_include_directive_does_not_error(self):
        """@INCLUDE directive must not produce a parse error."""
        _, summary = self.run_validator(
            """
            @INCLUDE /etc/fluent-bit/parsers.conf

            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name stdout
                Match *
            """,
            check="structure",
        )
        self.assertFalse(
            any("outside of a section" in e for e in summary["errors"]),
            "@INCLUDE should not be reported as 'Parameter outside of a section'",
        )

    def test_set_directive_does_not_error(self):
        """@SET directive must not produce a parse error."""
        _, summary = self.run_validator(
            """
            @SET my_path=/var/log/*.log

            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name stdout
                Match *
            """,
            check="structure",
        )
        self.assertFalse(
            any("outside of a section" in e for e in summary["errors"]),
            "@SET should not be reported as 'Parameter outside of a section'",
        )

    def test_include_directive_emits_recommendation(self):
        """@INCLUDE directive must produce a recommendation to validate the included file."""
        _, summary = self.run_validator(
            """
            @INCLUDE parsers.conf

            [SERVICE]
                Flush 5
            [INPUT]
                Name tail
                Path /var/log/*.log
                Tag app.logs
            [OUTPUT]
                Name stdout
                Match *
            """,
            check="structure",
        )
        self.assertTrue(
            any("@INCLUDE" in r and "parsers.conf" in r for r in summary["recommendations"]),
            "Expected a recommendation about the @INCLUDE directive",
        )


if __name__ == "__main__":
    unittest.main(verbosity=2)