lsquic

lsquic

Classes:

Name	Description
LsquicServiceManager

LsquicServiceManager

LsquicServiceManager(
    service_config_to_test: LsquicConfig,
    service_type: str,
    protocol: ProtocolConfig,
    implementation_name: str,
)

Bases: IImplementationManager

Methods:

Name	Description
add_command	Add a command to a specific phase of execution
build_command_args	Builds a list of command arguments with proper escaping
build_env_vars	Builds a dictionary of environment variables with proper escaping
generate_compile_commands	This method generates and returns a list of compile commands.
generate_deployment_commands	Generates a structured list of deployment command arguments for the LSQUIC service.
generate_post_compile_commands	Generate a list of post-compile commands.
generate_post_run_commands	Generates post-run commands.
generate_pre_compile_commands	Generates a list of shell commands to be executed before compilation.
generate_pre_run_commands	Generates a list of pre-run commands.
generate_run_command	Generates the run command for the LSQUIC service.
initialize_commands	Initializes and generates a dictionary of commands to be executed at different stages
prepare	Prepare the service manager for use.
render_commands	Renders a command using a Jinja2 template with the provided parameters.
render_template_with_structured_args	Renders a template with structured parameters for proper quoting

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/iut/quic/lsquic/lsquic.py

def __init__(
    self,
    service_config_to_test: LsquicConfig,
    service_type: str,
    protocol: ProtocolConfig,
    implementation_name: str,
):
    super().__init__(service_config_to_test, service_type, protocol, implementation_name)
    self.logger.debug("Initializing Lsquic service manager for '%s'", implementation_name)
    self.logger.debug("Loaded Lsquic configuration: %s", self.service_config_to_test)
    self.initialize_commands()

add_command

add_command(
    phase,
    command,
    description=None,
    is_function_definition=False,
    is_multiline=False,
    is_critical=True,
    working_dir=None,
    environment=None,
    timeout=None,
)

Add a command to a specific phase of execution

Parameters:

Name	Type	Description	Default
phase		The phase to add the command to (pre_compile, compile, post_compile, pre_run, run, post_run)	required
command		The command to add (str or ShellCommand)	required
description		Optional description of the command	None
is_function_definition		Whether this command is a shell function definition	False
is_multiline		Whether this command spans multiple lines	False
is_critical		Whether failure of this command should halt execution	True
working_dir		Working directory for the command execution	None
environment		Environment variables for the command	None
timeout		Command timeout in seconds	None

Returns:

Type	Description
	None

Raises:

Type	Description
ValueError	If the phase is invalid

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def add_command(
    self,
    phase,
    command,
    description=None,
    is_function_definition=False,
    is_multiline=False,
    is_critical=True,
    working_dir=None,
    environment=None,
    timeout=None,
):
    """
    Add a command to a specific phase of execution

    Args:
        phase: The phase to add the command to (pre_compile, compile, post_compile, pre_run, run, post_run)
        command: The command to add (str or ShellCommand)
        description: Optional description of the command
        is_function_definition: Whether this command is a shell function definition
        is_multiline: Whether this command spans multiple lines
        is_critical: Whether failure of this command should halt execution
        working_dir: Working directory for the command execution
        environment: Environment variables for the command
        timeout: Command timeout in seconds

    Returns:
        None

    Raises:
        ValueError: If the phase is invalid
    """
    valid_phases = ["pre_compile", "compile", "post_compile", "pre_run", "post_run"]
    if phase not in valid_phases:
        raise ValueError(
            f"Invalid command phase: {phase}. Must be one of: {', '.join(valid_phases)}"
        )

    command_key = f"{phase}_cmds"
    if command_key not in self.run_cmd:
        self.run_cmd[command_key] = []

    # Check if the command is already a ShellCommand object
    if isinstance(command, ShellCommand):
        # Use the existing ShellCommand object
        self.logger.debug("Adding %s to %s phase", command.description, phase)
        self.run_cmd[command_key].append(command)
    elif isinstance(command, list):
        # Convert each item in the list to a ShellCommand if it's a string
        for cmd in command:
            if isinstance(cmd, ShellCommand):
                self.run_cmd[command_key].append(cmd)
            else:
                # Create a new ShellCommand object
                shell_cmd = ShellCommand(
                    command=cmd,
                    description=description
                    or f"Command: {cmd[:40]}{'...' if len(cmd) > 40 else ''}",
                    is_critical=is_critical,
                    is_multiline=is_multiline,
                    is_function_definition=is_function_definition,
                    working_dir=working_dir,
                    environment=environment,
                    timeout=timeout,
                )
                self.run_cmd[command_key].append(shell_cmd)
    else:
        # Create a new ShellCommand object for a string command
        if description:
            self.logger.debug("Adding %s to %s phase", description, phase)

        shell_cmd = ShellCommand(
            command=command,
            description=description
            or f"Command: {command[:40]}{'...' if len(command) > 40 else ''}",
            is_critical=is_critical,
            is_multiline=is_multiline,
            is_function_definition=is_function_definition,
            working_dir=working_dir,
            environment=environment,
            timeout=timeout,
        )
        self.run_cmd[command_key].append(shell_cmd)

build_command_args

build_command_args(command_args)

Builds a list of command arguments with proper escaping

Parameters:

Name	Type	Description	Default
command_args		List of command arguments or a single string. If a string is provided, it will be split on spaces, respecting quoted sections.	required

Returns:

Name	Type	Description
list		A properly escaped list of command arguments

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def build_command_args(self, command_args):
    """
    Builds a list of command arguments with proper escaping

    Args:
        command_args: List of command arguments or a single string.
            If a string is provided, it will be split on spaces, respecting quoted sections.

    Returns:
        list: A properly escaped list of command arguments
    """
    if command_args is None:
        return []

    if isinstance(command_args, str):
        # Split the string on spaces, respecting quoted sections
        try:
            args = [arg for arg in shlex.split(command_args) if arg.strip()]
        except ValueError as e:
            self.logger.warning("Error splitting command: %s. Using as-is.", e)
            args = [command_args]
    elif isinstance(command_args, list):
        # Ensure all items in the list are strings
        args = [str(arg) for arg in command_args if arg is not None]
    else:
        args = [str(command_args)]

    return args

build_env_vars

build_env_vars(env_dict)

Builds a dictionary of environment variables with proper escaping

Parameters:

Name	Type	Description	Default
env_dict		Dictionary of environment variables	required

Returns:

Name	Type	Description
dict		A properly escaped dictionary of environment variables

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def build_env_vars(self, env_dict):
    """
    Builds a dictionary of environment variables with proper escaping

    Args:
        env_dict: Dictionary of environment variables

    Returns:
        dict: A properly escaped dictionary of environment variables
    """
    if not isinstance(env_dict, dict):
        self.logger.warning(
            "Expected dict for env_vars, got %s. Using empty dict.", type(env_dict)
        )
        return {}

    # Ensure all values are strings
    return {k: str(v) for k, v in env_dict.items()}

generate_compile_commands

generate_compile_commands() -> list[str]

This method generates and returns a list of compile commands. Generates compile commands.

Returns:

Name	Type	Description
list	list[str]	An empty list representing the compile commands.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def generate_compile_commands(self) -> list[str]:
    """
    This method generates and returns a list of compile commands.
    Generates compile commands.

    Returns:
        list: An empty list representing the compile commands.
    """
    return []

generate_deployment_commands

generate_deployment_commands() -> list

Generates a structured list of deployment command arguments for the LSQUIC service.

This method constructs the command arguments using the structured approach, creating a list of arguments rather than concatenating strings, which ensures proper escaping and handling of special characters.

Returns:

Name	Type	Description
list	list	The list of command arguments.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_deployment_commands(self) -> list:
    """
    Generates a structured list of deployment command arguments for the LSQUIC service.

    This method constructs the command arguments using the structured approach,
    creating a list of arguments rather than concatenating strings, which
    ensures proper escaping and handling of special characters.

    Returns:
        list: The list of command arguments.
    """
    self.logger.debug(
        f"Generating deployment commands for service: {self.service_name} with service parameters: {self.service_config_to_test}"
    )

    # Get appropriate parameters based on role
    if self.role == RoleEnum.server:
        params = self.service_config_to_test.implementation.version.server
    else:  # client
        params = self.service_config_to_test.implementation.version.client

    target = self.service_config_to_test.protocol.target

    # Initialize the command argument list
    cmd_args = []

    # Add certificate parameters
    if "certificates" in params and params["certificates"]:
        certs = params["certificates"]
        if "cert_param" in certs and "cert_file" in certs:
            cmd_args.extend([certs["cert_param"], certs["cert_file"]])
        if "key_param" in certs and "key_file" in certs:
            cmd_args.extend([certs["key_param"], certs["key_file"]])

    # Add ticket file parameters for client
    if self.role == RoleEnum.client and "ticket_file" in params:
        ticket = params["ticket_file"]
        if "param" in ticket and "file" in ticket:
            cmd_args.extend([ticket["param"], ticket["file"]])

    # Add protocol parameters (ALPN)
    if "protocol" in params and params["protocol"]:
        proto = params["protocol"]
        if "alpn" in proto and proto["alpn"]:
            cmd_args.extend([proto["alpn"]["param"], proto["alpn"]["value"]])

        # Add additional parameters
        if "additional_parameters" in proto and proto["additional_parameters"]:
            # Split additional parameters into separate arguments
            additional_params = self.build_command_args(proto["additional_parameters"])
            cmd_args.extend(additional_params)

    # Add network interface if specified
    if "network" in params and params["network"] and "interface" in params["network"]:
        network = params["network"]
        cmd_args.extend([network["interface"]["param"], network["interface"]["value"]])

    # Add server address parameter
    if "network" in params and "port" in params["network"]:
        cmd_args.extend(["-s", f"{target}:{params['network']['port']}"])

    # Add logging redirection
    if "logging" in params and params["logging"]:
        cmd_args.extend(
            [">", params["logging"]["log_path"], "2>", params["logging"]["err_path"]]
        )

    self.logger.debug("Generated command arguments: %s", cmd_args)
    return cmd_args

generate_post_compile_commands

generate_post_compile_commands() -> list[str]

Generate a list of post-compile commands. This method returns an empty list of strings representing commands to be executed after the compilation process. Returns: List[str]: An empty list of post-compile commands.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def generate_post_compile_commands(self) -> list[str]:
    """
    Generate a list of post-compile commands.
    This method returns an empty list of strings representing commands
    to be executed after the compilation process.
    Returns:
        List[str]: An empty list of post-compile commands.
    """

    return []

generate_post_run_commands

generate_post_run_commands()

Generates post-run commands.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_post_run_commands(self):
    """
    Generates post-run commands.
    """
    return super().generate_post_run_commands() + ["cp /opt/lsquic/bin /app/logs/;"]

generate_pre_compile_commands

generate_pre_compile_commands() -> list

Generates a list of shell commands to be executed before compilation.

Returns:

Name	Type	Description
list	list	A list of either string commands or ShellCommand objects if available

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def generate_pre_compile_commands(self) -> list:
    """
    Generates a list of shell commands to be executed before compilation.

    Returns:
        list: A list of either string commands or ShellCommand objects if available
    """
    # ShellCommand is imported at the top of the file, so we use it directly
    # for better shell command representation with metadata and proper escaping

    try:
        # Using ShellCommand objects for better structure, error handling, and debugging support
        return [
            ShellCommand(
                command="set -x;", description="Enable command tracing", is_critical=True
            ),
            # Set PS4 with the file and line number pattern for better debugging
            # Use a single properly formatted command to avoid syntax errors
            ShellCommand(
                command='export PS4="+ [${BASH_SOURCE:-sh}:${LINENO}] "',
                description="Configure PS4 variable for enhanced debug output",
                is_critical=True,
            ),
            ShellCommand(
                command="export SHELLOPTS",
                description="Export shell options for subshells",
                is_critical=True,
            ),
            ShellCommand(
                command="export PATH=$PATH:$ADDITIONAL_PATH;",
                description="Set PATH environment variable",
                is_critical=False,  # Non-critical as ADDITIONAL_PATH might be empty
            ),
            ShellCommand(
                command="export PYTHONPATH=$PYTHONPATH:$ADDITIONAL_PYTHONPATH;",
                description="Set PYTHONPATH environment variable",
                is_critical=False,  # Non-critical as ADDITIONAL_PYTHONPATH might be empty
            ),
            ShellCommand(
                command="env >> /app/logs/ivy_setup.log;",
                description="Log environment variables for debugging",
                is_critical=False,
            ),
        ]
    except (ImportError, AttributeError) as e:
        # Fallback to plain string commands if ShellCommand can't be used
        self.logger.warning(
            "Error using ShellCommand objects: %s. Falling back to legacy string commands.",
            str(e),
        )
        return [
            "set -x;",
            'export PS4="+ [${BASH_SOURCE:-sh}:${LINENO}] "',
            "export SHELLOPTS",
            "export PATH=$PATH:$ADDITIONAL_PATH;",
            "export PYTHONPATH=$PYTHONPATH:$ADDITIONAL_PYTHONPATH;",
            "env >> /app/logs/ivy_setup.log;",
        ]

generate_pre_run_commands

generate_pre_run_commands() -> list[str]

Generates a list of pre-run commands. This method returns an empty list of strings, which can be overridden by subclasses to provide specific pre-run commands required for their execution context. Returns: List[str]: An empty list of strings representing pre-run commands.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def generate_pre_run_commands(self) -> list[str]:
    """
    Generates a list of pre-run commands.
    This method returns an empty list of strings, which can be overridden by subclasses
    to provide specific pre-run commands required for their execution context.
    Returns:
        List[str]: An empty list of strings representing pre-run commands.
    """

    return []

generate_run_command

generate_run_command()

Generates the run command for the LSQUIC service.

This method constructs a complete run command configuration using the structured approach for proper quoting and escaping of all command arguments and environment variables.

Returns:

Name	Type	Description
dict		The run command configuration with all necessary components.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/iut/quic/lsquic/lsquic.py

def generate_run_command(self):
    """
    Generates the run command for the LSQUIC service.

    This method constructs a complete run command configuration using
    the structured approach for proper quoting and escaping of all
    command arguments and environment variables.

    Returns:
        dict: The run command configuration with all necessary components.
    """
    if self.role == RoleEnum.server:
        params = self.service_config_to_test.implementation.version.server
    else:  # client
        params = self.service_config_to_test.implementation.version.client

    self.working_dir = params["binary"]["dir"]

    # Build command arguments as list
    command_args = self.generate_deployment_commands()

    # Environment variables
    env_vars = {
        "LD_LIBRARY_PATH": "/opt/lsquic/lib",
        "LSQUIC_LOGS_DIR": "/app/logs/lsquic",
    }

    # Add any additional environment variables from config
    if "environment" in params:
        for key, value in params["environment"].items():
            env_vars[key] = value

    # Try to render with structured template, fall back to original if needed
    try:
        template_name = f"{str(self.role.name)}_command_structured.jinja"
        rendered_command = self.render_template_with_structured_args(
            template_name, params, command_args, env_vars
        )
        # For structured templates, we'll use the rendered command as a string
        command_args = rendered_command
    except Exception as e:
        self.logger.warning(
            f"Failed to use structured template for {self.service_name}: {e}. "
            "Falling back to default command generation."
        )
        # Keep command_args as is if the structured template fails

    return {
        "working_dir": self.working_dir,
        "command_binary": params["binary"]["name"],
        "command_args": command_args,
        "timeout": self.service_config_to_test.timeout,
        "command_env": env_vars,
    }

initialize_commands

initialize_commands() -> dict

Initializes and generates a dictionary of commands to be executed at different stages of the process (pre-compile, compile, post-compile, pre-run, run, post-run).

The dictionary keys are

• "pre_compile_cmds": Commands to be executed before compilation.
• "compile_cmds": Commands to be executed during compilation.
• "post_compile_cmds": Commands to be executed after compilation.
• "pre_run_cmds": Commands to be executed before running.
• "run_cmd": Command to be executed to run the main process.
• "post_run_cmds": Commands to be executed after running.

Returns:

Name	Type	Description
dict	dict	A dictionary containing the commands for each stage.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

@validate_cmd
def initialize_commands(self) -> dict:
    """
    Initializes and generates a dictionary of commands to be executed at different stages
    of the process (pre-compile, compile, post-compile, pre-run, run, post-run).

    The dictionary keys are:
        - "pre_compile_cmds": Commands to be executed before compilation.
        - "compile_cmds": Commands to be executed during compilation.
        - "post_compile_cmds": Commands to be executed after compilation.
        - "pre_run_cmds": Commands to be executed before running.
        - "run_cmd": Command to be executed to run the main process.
        - "post_run_cmds": Commands to be executed after running.

    Returns:
        dict: A dictionary containing the commands for each stage.
    """
    self.run_cmd = {
        "pre_compile_cmds": self.generate_pre_compile_commands(),
        "compile_cmds": self.generate_compile_commands(),
        "post_compile_cmds": self.generate_post_compile_commands(),
        "pre_run_cmds": self.generate_pre_run_commands(),
        "run_cmd": self.generate_run_command(),
        "post_run_cmds": self.generate_post_run_commands(),
    }
    self.logger.debug("Run commands: %s", self.run_cmd)
    return self.run_cmd

prepare

prepare(plugin_loader: PluginLoader | None = None)

Prepare the service manager for use.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/iut/quic/lsquic/lsquic.py

def prepare(self, plugin_loader: PluginLoader | None = None):
    """
    Prepare the service manager for use.
    """
    self.logger.debug("Preparing Lsquic service manager...")
    plugin_loader.build_docker_image_from_path(
        Path(
            os.path.join(
                os.getcwd(),
                "panther",
                "plugins",
                "services",
                "Dockerfile",
            )
        ),
        "panther_base",
        "service",
    )
    plugin_loader.build_docker_image(
        self.get_implementation_name(),
        self.service_config_to_test.implementation.version,
    )

render_commands

render_commands(
    params,
    template_name,
    command_args=None,
    env_vars=None,
    extra_fields=None,
)

Renders a command using a Jinja2 template with the provided parameters.

Parameters:

Name	Type	Description	Default
params		Dictionary containing regular parameters for template rendering.	required
template_name		Name of the template to render.	required
command_args		List of command arguments for structured command generation.	None
env_vars		Dictionary of environment variables for structured command generation.	None
extra_fields		Additional YAML fragments as a string for structured templates.	None

Returns:

Name	Type	Description
str		The rendered command string.

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def render_commands(
    self, params, template_name, command_args=None, env_vars=None, extra_fields=None
):
    """
    Renders a command using a Jinja2 template with the provided parameters.

    Args:
        params: Dictionary containing regular parameters for template rendering.
        template_name: Name of the template to render.
        command_args: List of command arguments for structured command generation.
        env_vars: Dictionary of environment variables for structured command generation.
        extra_fields: Additional YAML fragments as a string for structured templates.

    Returns:
        str: The rendered command string.
    """
    self.logger.debug(
        f"Rendering command using template '{template_name}' with parameters: {params}"
    )

    # Register the quoting filters for shell and YAML
    self.jinja_env.filters["quote_shell"] = quote_shell
    self.jinja_env.filters["quote_yaml"] = quote_yaml

    template = self.jinja_env.get_template(template_name)

    # Enhance the params dict with structured command_args, env_vars and extra_fields if provided
    render_params = dict(params)
    if command_args is not None:
        render_params["command_args"] = command_args
    if env_vars is not None:
        render_params["env_vars"] = env_vars
    if extra_fields is not None:
        render_params["extra_fields"] = extra_fields

    command = template.render(**render_params)

    # Clean up the command string, but preserve newlines for multiline commands
    if "\n" not in command:
        command_str = command.replace("\t", " ").strip()
    else:
        command_str = command

    service_name = self.service_config_to_test.name
    self.logger.debug("Generated command for '%s': %s", service_name, command_str)
    return command_str

render_template_with_structured_args

render_template_with_structured_args(
    template_name,
    params=None,
    command_args=None,
    env_vars=None,
    extra_fields=None,
)

Renders a template with structured parameters for proper quoting

Parameters:

Name	Type	Description	Default
template_name		Name of the template file	required
params		Additional parameters to pass to the template	None
command_args		List of command arguments or a string to be split	None
env_vars		Dictionary of environment variables	None
extra_fields		String with additional YAML fragments	None

Returns:

Name	Type	Description
str		The rendered template with properly quoted values

Source code in .venv/lib/python3.10/site-packages/panther/plugins/services/services_interface.py

def render_template_with_structured_args(
    self,
    template_name,
    params=None,
    command_args=None,
    env_vars=None,
    extra_fields=None,
):
    """
    Renders a template with structured parameters for proper quoting

    Args:
        template_name: Name of the template file
        params: Additional parameters to pass to the template
        command_args: List of command arguments or a string to be split
        env_vars: Dictionary of environment variables
        extra_fields: String with additional YAML fragments

    Returns:
        str: The rendered template with properly quoted values
    """
    params = params or {}
    processed_args = self.build_command_args(command_args) if command_args is not None else None
    processed_env = self.build_env_vars(env_vars) if env_vars is not None else None

    return self.render_commands(
        params, template_name, processed_args, processed_env, extra_fields
    )