panther_cli

This script is the main entry point for the PANTHER CLI. It provides a command-line interface to manage the PANTHER tool.

build_docker_visualizer

build_docker_visualizer(push=False)

summary

Parameters:

• push (bool, default: False ) –

  description. Defaults to False.

Source code in panther/panther_cli.py

def build_docker_visualizer(push=False):
    """_summary_

    Args:
        push (bool, optional): _description_. Defaults to False.
    """
    client = docker.from_env()
    logger.info("Building Docker image visualizer")
    client.images.build(
        path="panther_webapp/tools/",
        rm=True,
        dockerfile="Dockerfile.visualizer",
        tag="ivy-visualizer",
        network_mode="host",
    )
    if push:
        push_image_to_registry("ivy-visualizer")

build_webapp

build_webapp(push=False)

summary

Parameters:

• push (bool, default: False ) –

  description. Defaults to False.

Source code in panther/panther_cli.py

def build_webapp(push=False):
    """_summary_

    Args:
        push (bool, optional): _description_. Defaults to False.
    """
    client = docker.from_env()
    logger.info("Building Docker image panther-webapp")
    execute_command("sudo chown -R $USER:$GROUPS $PWD/panther_webapp/")
    image_obj, log_generator = client.images.build(
        path="panther_webapp",
        dockerfile="Dockerfile.ivy_webapp",
        tag="panther-webapp",
        network_mode="host",
        rm=True,
        quiet=False,
    )  # squash=True,
    log_docker_output(log_generator, "Building Docker image panther-webap")
    if push:
        push_image_to_registry("panther-webapp")

build_worker

build_worker(implem, config, push=False)

summary

Parameters:

• implem (_type_) –

  description

• config (_type_) –

  description

• push (bool, default: False ) –

  description. Defaults to False.

Source code in panther/panther_cli.py

def build_worker(implem, config, push=False):
    """_summary_

    Args:
        implem (_type_): _description_
        config (_type_): _description_
        push (bool, optional): _description_. Defaults to False.
    """
    stop_tool()
    execute_command("git clean -f -d panther_worker/panther-ivy;")
    client = docker.from_env()

    implem_build_commands = dict(config.items('implem_build_commands'))
    shadow_support = config['shadow_support']
    tag, path, dockerfile = eval(implem_build_commands[implem])

    logger.info(f"Building Docker image {tag} from {dockerfile}")
    # Build the base ubuntu-panther image
    logger.debug("Building Docker image ubuntu-panther")
    image_obj, log_generator = client.images.build(
        path="panther_worker/",
        dockerfile="Dockerfile.ubuntu",
        tag="ubuntu-panther",
        rm=True,
        network_mode="host",
    )
    log_docker_output(log_generator, "Building Docker image ubuntu-panther")

    # Build the first ivy image
    logger.debug("Building Docker image ivy")
    image_obj, log_generator = client.images.build(
        path="panther_worker/",
        dockerfile="Dockerfile.ivy_1",
        tag="ivy",
        rm=True,
        # buildargs={"CACHEBUST": str(time.time())}, # Cache invalidation
        network_mode="host",
    )
    log_docker_output(log_generator, "Building Docker image ivy")

    # Check if shadow build is needed
    shadow_tag = None
    final_tag = f"{tag}-panther"

    if shadow_support.getboolean(implem):
        logger.debug("Building Docker image shadow-panther")
        image_obj, log_generator = client.images.build(
            path="panther_worker/",
            dockerfile="Dockerfile.shadow",
            tag="shadow-panther",
            rm=True,
            network_mode="host",
        )
        log_docker_output(log_generator, "Building Docker image shadow-panther")
        shadow_tag = "shadow-panther"

        # Build the picotls image
        build_args = {"image": shadow_tag}
        itag = "shadow-panther-picotls"
        logger.debug(f"Building Docker image {itag} from tag {build_args}")
        image_obj, log_generator = client.images.build(
            path="panther_worker/app/implementations/quic-implementations/picotls/",
            dockerfile="Dockerfile.picotls",
            tag=itag,
            rm=True,
            network_mode="host",
            buildargs=build_args,
        )
        log_docker_output(log_generator, "Building Docker image shadow-panther-picotls")
    else:
        # Build the picotls image
        build_args = {"image": "ivy"}
        itag = "panther-picotls"
        logger.debug(f"Building Docker image {itag} from tag {build_args}")
        image_obj, log_generator = client.images.build(
            path="panther_worker/app/implementations/quic-implementations/picotls/",
            dockerfile="Dockerfile.picotls",
            tag=itag,
            rm=True,
            network_mode="host",
            buildargs=build_args,
        )
        log_docker_output(log_generator, "Building Docker image panther-picotls")

    # Build the specified implementation image
    build_args = (
        {"image": "shadow-panther-picotls"} if shadow_tag else {"image": "panther-picotls"}
    )
    logger.debug(f"Building Docker image {tag} from tag {build_args}")
    image_obj, log_generator = client.images.build(
        path=path,
        dockerfile=dockerfile,
        tag=tag,
        rm=True,
        network_mode="host",
        buildargs=build_args,
    )

    log_docker_output(log_generator, f"Building Docker image {tag}")
    # Build the final implementation-ivy image
    build_args = {"image": tag}
    logger.debug(f"Building Docker image {final_tag} from tag {build_args}")
    image_obj, log_generator = client.images.build(
        path="panther_worker/",
        dockerfile="Dockerfile.ivy_2",
        tag=final_tag,
        rm=True,
        network_mode="host",
        buildargs=build_args,
    )
    log_docker_output(log_generator, f"Building Docker image {final_tag}")

    if push:
        push_image_to_registry(final_tag)

clean_tool

clean_tool(config)

summary

Parameters:

• config (_type_) –

  description

Source code in panther/panther_cli.py

def clean_tool(config):
    """_summary_

    Args:
        config (_type_): _description_
    """
    client = docker.from_env()
    docker_containers = client.containers.list(all=True)
    for dc in docker_containers:
        dc.remove(force=True)
    logger.info(client.containers.prune())
    logger.info(client.images.prune(filters={"dangling": False}))
    logger.info(client.networks.prune())
    logger.info(client.volumes.prune())

execute_command

execute_command(command, tmux=None, cwd=None)

summary

Parameters:

• command (_type_) –

  description

• tmux (_type_, default: None ) –

  description. Defaults to None.

• cwd (_type_, default: None ) –

  description. Defaults to None.

Raises:

• CalledProcessError –

  description

Source code in panther/panther_cli.py

def execute_command(command, tmux=None, cwd=None):
    """_summary_

    Args:
        command (_type_): _description_
        tmux (_type_, optional): _description_. Defaults to None.
        cwd (_type_, optional): _description_. Defaults to None.

    Raises:
        subprocess.CalledProcessError: _description_
    """
    logger.debug(f"Executing command: {command}")

    if tmux:
        logger.info(f"Executing command in tmux with log file:  \"{command}\"")
        session_name = subprocess.check_output(['tmux', 'display-message', '-p', '#S']).strip().decode('utf-8')
        os.system(f"tmux split-window -h -l 80%; tmux send-keys -t  {session_name}:0.1 \"{command}\" C-m;")
    else:
        if cwd:
            result = subprocess.run(command, shell=True, cwd=cwd)
        else:
            result = subprocess.run(command, shell=True)
        if result.returncode != 0:
            raise subprocess.CalledProcessError(result.returncode, command)

get_current_branch

get_current_branch()

summary

Returns:

• _type_ –

  description

Source code in panther/panther_cli.py

def get_current_branch():
    """_summary_

    Returns:
        _type_: _description_
    """
    result = subprocess.run(
        f"git rev-parse --abbrev-ref HEAD",
        shell=True,
        stdout=subprocess.PIPE,
        text=True,
    )
    logger.info(f"Current branch: {result.stdout.strip()}")
    return result.stdout.strip()

install_tool

install_tool(config, branch=None)

summary

Parameters:

• config (_type_) –

  description

• branch (_type_, default: None ) –

  description. Defaults to None.

Source code in panther/panther_cli.py

def install_tool(config, branch=None):
    """_summary_

    Args:
        config (_type_): _description_
        branch (_type_, optional): _description_. Defaults to None.
    """
    # Pre-installation commands
    logger.info("Running pre-installation commands")

    # Create necessary directories
    # TODO already done in other scripts
    for folder in config["directories"]:
        logger.info(f"Creating directory: {config['directories'][folder]}")
        # Create build/ and test/temp/ directories inside folder
        os.makedirs(os.path.join(folder, "build"), exist_ok=True)
        os.makedirs(os.path.join(folder, "test", "temp"), exist_ok=True)

    # Install modules
    if config["modules"].getboolean("checkout_git"):
        logger.info("Checking out git repositories")
        if branch is not None:
            execute_command(f"git checkout {branch}")
        current_branch = get_current_branch()
        execute_command("git submodule update --init --recursive")
        # TODO cd not working -> chdir
        execute_command(
            f"git fetch",
            cwd="panther_worker/panther-ivy/"
        )
        execute_command(
            f"git checkout {current_branch}",
            cwd="panther_worker/panther-ivy/"
        )
        execute_command(
            f"git pull",
            cwd="panther_worker/panther-ivy/"
        )
        execute_command(
            f"git submodule update --init --recursive",
            cwd="panther_worker/panther-ivy/"
        )
        execute_command(
            f"git pull",
            cwd="panther_worker/panther-ivy/"
        )
        # execute_command(
        #     "cd panther_worker/app/implementations/quic-implementations/picotls-implem;" + \  
        #     "git checkout 047c5fe20bb9ea91c1caded8977134f19681ec76;" + \
        #     "git submodule update --init --recursive" + \
        # )

    if config["modules"].getboolean("build_webapp"):
        build_webapp()

    if config["modules"].getboolean("build_worker"):
        for implem, should_build in config["implems"].items():
            if should_build.lower() == "true":
                if not "shadow" in implem:
                    build_worker(implem, config)
                elif config["modules"]["build_shadow"].lower() == "true":
                    build_worker(implem, config)

    if config["modules"].getboolean("build_visualizer"):
        build_docker_visualizer()

    update_docker_compose(config)

is_tmux_session

is_tmux_session()

Check if running inside a tmux session.

Source code in panther/panther_cli.py

def is_tmux_session():
    """Check if running inside a tmux session."""
    return 'TMUX' in subprocess.run(['env'], capture_output=True, text=True).stdout

load_config

load_config(config_path)

summary

Parameters:

• config_path (_type_) –

  description

Returns:

• _type_ –

  description

Source code in panther/panther_cli.py

def load_config(config_path):
    """_summary_

    Args:
        config_path (_type_): _description_

    Returns:
        _type_: _description_
    """
    config = configparser.ConfigParser(interpolation=configparser.ExtendedInterpolation())
    config.read(config_path)
    return config

start_bash_container

start_bash_container(implem)

summary Start a Docker container with the specified parameters.

Parameters:

• implem (_type_) –

  description

Returns:

• _type_ –

  description

Source code in panther/panther_cli.py

def start_bash_container(implem):
    """_summary_
    Start a Docker container with the specified parameters.

    Args:
        implem (_type_): _description_

    Returns:
        _type_: _description_
    """
    client = docker.from_env()
    pwd = os.getcwd()
    def get_nproc():
        """Get the number of processors available."""
        try:
            result = subprocess.run(["nproc"], capture_output=True, text=True, check=True)
            return result.stdout.strip()
        except subprocess.CalledProcessError as e:
            print(f"Error getting the number of processors: {e}")
            return "1"
    nproc = get_nproc()
    cpus = f"{nproc}.0"

    container_name = f"{implem}-panther"

    volumes = {
        f"{pwd}/tls-keys": {"bind": "/app/tls-keys", "mode": "rw"},
        f"{pwd}/tickets":  {"bind": "/app/tickets", "mode": "rw"},
        f"{pwd}/qlogs":    {"bind": "/app/qlogs", "mode": "rw"},
        f"{pwd}/panther_worker/app/panther-ivy/protocol-testing/": {
            "bind": "/app/panther-ivy/protocol-testing/",
            "mode": "rw",
        },
        f"{pwd}/panther_worker/app/panther-ivy/ivy/include/1.7": {
            "bind": "/app/panther-ivy/ivy/include/1.7",
            "mode": "rw",
        },
    }

    try:
        container = client.containers.run(
            image=container_name,
            command="bash",
            privileged=True,
            cpus=cpus,
            mem_limit="10g",
            mem_reservation="9.5g",
            volumes=volumes,
            tty=True,
            stdin_open=True,
            detach=True,
        )
        print(f"Started container {container.id} ({container_name})")
    except Exception as e:
        print(f"Error starting the container: {e}")

start_tool

start_tool(config, swarm=False)

summary

Parameters:

• config (_type_) –

  description

• swarm (bool, default: False ) –

  description. Defaults to False.

Source code in panther/panther_cli.py

def start_tool(config, swarm=False):
    """_summary_

    Args:
        config (_type_): _description_
        swarm (bool, optional): _description_. Defaults to False.
    """
    client = docker.from_env()

    create_docker_network()

    execute_command("sudo chown -R $USER:$GROUPS $PWD/")
    execute_command("xhost +")

    if swarm:
        execute_command("docker swarm init")
        yaml_path, defined_services = update_docker_swarm(config)
    else:
        yaml_path, defined_services = update_docker_compose(config)

    execute_command(f"cat {yaml_path}")

    if swarm:
        execute_command(f"docker stack rm panther")
        execute_command(f"docker stack -c {yaml_path} panther")
    else:
        execute_command(f"docker compose -f {yaml_path} up -d")

    execute_command("clear")

    docker_to_monitor = []
    for container_name in defined_services:
        if container_exists(client, container_name):
            docker_to_monitor.append(container_name)
            ip_address = get_container_ip(client, container_name)
            if ip_address:
                entry = f"{ip_address} {container_name}\n"
                append_to_hosts_file(entry)
        else:
            logger.info(f"Container '{container_name}' does not exist.")

    thread = threading.Thread(target=monitor_docker_usage, args=([docker_to_monitor, 1, -1])) 
    thread.start()

    if swarm:
        compose_log = f"\'logs/swarm_{datetime.now()}.log\'"
        execute_command(f"docker stack services panther | tee {compose_log}", tmux=compose_log)
    else:
        compose_log = f"\'logs/compose_{datetime.now()}.log\'"
        execute_command(f"docker compose -f {yaml_path} logs -f | tee {compose_log}", tmux=compose_log)

    # TODO: should split the first 
    session_name = subprocess.check_output(['tmux', 'display-message', '-p', '#S']).strip().decode('utf-8')
    os.system(f"tmux split-window -v -l 10%; tmux send-keys -t  {session_name}:0.0 \"bash\" C-m;")

stop_tool

stop_tool()

summary

Source code in panther/panther_cli.py

def stop_tool():
    """_summary_
    """
    client = docker.from_env()
    docker_containers = client.containers.list(all=True)
    for dc in docker_containers:
        dc.stop()