Skip to content

PANTHER Service Base Classes¤

Inheritance-based architecture for eliminating code duplication in service implementations

The PANTHER base class system provides a clean inheritance hierarchy that eliminates code duplication across protocol implementations. This architecture resulted in a 47.2% average code reduction and simplified maintenance across all service plugins.

Architecture Benefits

  • Reduced code duplication across implementations
  • Consistent behavior through shared base classes
  • Template method pattern for extensible customization
  • Single point for fixes and enhancements

Base Class Hierarchy¤

The service base classes follow a clear inheritance pattern designed around the Template Method pattern:

IServiceManager (Interface)
    ↓
├─→ BaseQUICServiceManager (Abstract Base)
│   ├─→ PythonQUICServiceManager (Python implementations)
│   ├─→ RustQUICServiceManager (Rust implementations)
│   └─→ [Direct inheritance] (C/Go implementations)
│
├─→ BaseHTTPServiceManager (Abstract Base)
│   ├─→ [Language-specific extensions available]
│   └─→ [Direct inheritance] (Any language)
│
└─→ BaseMinipServiceManager (Abstract Base)
    ├─→ [Language-specific extensions available]
    └─→ [Direct inheritance] (Any language)

Template Method Pattern¤

All base classes use the Template Method pattern where:

  1. Base classes define the algorithm structure (command generation workflow)
  2. Derived classes implement specific steps (implementation-specific arguments)
  3. Common functionality is shared across all implementations
  4. Customization points allow implementation-specific behavior

Core Base Classes¤

BaseQUICServiceManager¤

Location: quic_service_base.py

The foundational base class for all QUIC implementations providing:

Core Functionality¤

from panther.plugins.services.base.quic_service_base import BaseQUICServiceManager

class MyQuicServiceManager(BaseQUICServiceManager):

    # Required implementations
    def _get_implementation_name(self) -> str:
        return "my_quic_impl"

    def _get_binary_name(self) -> str:
        return "my_quic_binary"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific server arguments
        return ["-p", str(kwargs.get("port", 4443))]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific client arguments
        host = kwargs.get("host", "localhost")
        port = kwargs.get("port", 4443)
        return [f"{host}:{port}"]

    def generate_deployment_commands(self) -> str:
        # Deployment-specific commands
        return f"{self._get_binary_name()} -p 4443"

    def _do_prepare(self, plugin_manager=None):
        # Preparation logic (Docker builds, dependencies)
        pass

Abstract Methods (Must Implement)¤

Method Purpose Returns
_get_implementation_name() Implementation identifier str
_get_binary_name() Executable name str
_get_server_specific_args(**kwargs) Server-specific arguments List[str]
_get_client_specific_args(**kwargs) Client-specific arguments List[str]
generate_deployment_commands() Deployment commands str
_do_prepare(plugin_manager) Preparation logic None

Provided Functionality¤

Method Purpose Usage
generate_run_command(**kwargs) Main command generation Use directly - implements template method
_extract_common_params(**kwargs) Extract QUIC parameters Override to add implementation params
_build_server_args(params) Build server arguments Override for custom server logic
_build_client_args(params) Build client arguments Override for custom client logic
get_supported_features() List capabilities Override to specify features

PythonQUICServiceManager¤

Location: python_quic_base.py

Specialized base class for Python implementations (e.g., AioQUIC):

from panther.plugins.services.base.python_quic_base import PythonQUICServiceManager

class AioquicServiceManager(PythonQUICServiceManager):

    def _get_implementation_name(self) -> str:
        return "aioquic"

    def _get_binary_name(self) -> str:
        return "python3"

    # Python-specific compile command automatically provided
    # Returns: pip install -r requirements.txt (if exists)

Python-Specific Features¤

  • No Compilation: generate_compile_command() handles Python dependency installation
  • Virtual Environment: Support for Python path configuration
  • Async Support: Integration with asyncio patterns
  • Package Management: Automatic requirements.txt installation

Python Parameters¤

Parameter Default Purpose
python_path /opt/aioquic Python module path
asyncio_debug false Enable asyncio debugging

RustQUICServiceManager¤

Location: rust_quic_base.py

Specialized base class for Rust implementations (e.g., Quiche, Quinn):

from panther.plugins.services.base.rust_quic_base import RustQUICServiceManager

class QuicheServiceManager(RustQUICServiceManager):

    def _get_implementation_name(self) -> str:
        return "quiche"

    def _get_binary_name(self) -> str:
        return "quiche-server"  # or quiche-client based on role

    # Rust-specific compile command automatically provided
    # Returns: cargo build --release --jobs $(nproc)

Rust-Specific Features¤

  • Cargo Integration: generate_compile_command() handles Cargo builds
  • Release Builds: Optimized release builds by default
  • Parallel Compilation: Multi-core Cargo builds
  • Environment Variables: Rust-specific logging and debugging

Rust Parameters¤

Parameter Default Purpose
build_type release Cargo build type (debug/release)
jobs $(nproc) Parallel build jobs
rust_log info Rust logging level
rust_backtrace 1 Enable Rust backtraces

BaseHTTPServiceManager¤

Location: http_service_base.py

Base class for HTTP protocol implementations. Currently provides the foundation for HTTP service implementations:

from panther.plugins.services.base.http_service_base import BaseHTTPServiceManager

class HttpServiceManager(BaseHTTPServiceManager):

    def _get_implementation_name(self) -> str:
        return "basic_http"

    def _get_binary_name(self) -> str:
        return "http_server"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific server arguments
        return ["--daemon"]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific client arguments
        return ["--user-agent", "PANTHER-HTTP-Client"]

    def generate_deployment_commands(self) -> str:
        return f"{self._get_binary_name()} --start"

    def _do_prepare(self, plugin_manager=None):
        # HTTP-specific preparation
        pass

HTTP-Specific Features¤

  • Multi-Version Support: HTTP/1.1, HTTP/2, HTTP/3 detection and configuration
  • TLS Integration: HTTPS support with certificate management
  • Keep-Alive Support: Connection persistence handling
  • Method Support: GET, POST, PUT, DELETE, etc.
  • Content-Type Handling: Automatic content type configuration

HTTP Parameters¤

Parameter Default Purpose
protocol_version "1.1" HTTP version (1.1, 2.0, 3.0)
port 80 HTTP port (443 for HTTPS)
method "GET" HTTP method for client requests
url_path "/" URL path for requests
content_type "text/html" Response content type
keep_alive false Enable connection keep-alive
max_connections 100 Maximum concurrent connections

HTTP Feature Detection¤

Feature Method Purpose
supports_tls() Check HTTPS support TLS capability detection
supports_http2() Check HTTP/2 support Protocol version detection
supports_http3() Check HTTP/3 support Modern protocol support
get_default_port() Get HTTP port Returns 80
get_tls_port() Get HTTPS port Returns 443

BaseMinipServiceManager¤

Location: minip_service_base.py

Base class for MINIP (Minimal Internet Protocol) implementations. PANTHER currently includes one MINIP implementation (ping_pong):

from panther.plugins.services.base.minip_service_base import BaseMinipServiceManager

class MinipServiceManager(BaseMinipServiceManager):

    def _get_implementation_name(self) -> str:
        return "basic_minip"

    def _get_binary_name(self) -> str:
        return "minip_binary"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific server arguments
        return ["--listen"]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        # Implementation-specific client arguments
        return ["--send"]

    def generate_deployment_commands(self) -> str:
        return f"{self._get_binary_name()} --start"

    def _do_prepare(self, plugin_manager=None):
        # MINIP-specific preparation
        pass

MINIP-Specific Features¤

  • Connection Types: TCP and UDP support
  • Message Validation: Protocol message format checking
  • Ping-Pong Testing: Simple request-response validation
  • Streaming Support: Continuous data transfer
  • Compression: Optional message compression
  • Flow Control: Basic flow control mechanisms

MINIP Parameters¤

Parameter Default Purpose
protocol_version "1.0" MINIP protocol version
port 8000 MINIP service port
connection_type "tcp" Connection type (tcp/udp)
message_size 1024 Message size in bytes
packet_size 512 Packet size for fragmentation
timeout 30 Connection timeout
debug_level 0 Debug verbosity level

MINIP Feature Detection¤

Feature Method Purpose
supports_tcp() Check TCP support TCP capability detection
supports_udp() Check UDP support UDP capability detection
supports_ping_pong() Check ping-pong mode Basic testing support
supports_streaming() Check streaming mode Continuous transfer support
supports_encryption() Check encryption Security feature detection

MINIP Utilities¤

Method Purpose Usage
generate_test_message(size) Create test payload Generate messages for testing
validate_message_format(msg) Validate message Check protocol compliance
calculate_message_overhead(size) Calculate overhead Protocol efficiency analysis

Docker Build Base Classes¤

Location: docker_build_base.py

Provides standardized Docker build patterns to eliminate Dockerfile duplication:

BaseDockerBuilder¤

from panther.plugins.services.base.docker_build_base import BaseDockerBuilder

class CustomDockerBuilder(BaseDockerBuilder):

    def __init__(self, implementation_name: str):
        super().__init__(implementation_name, "panther_base_service:latest")

    def generate_implementation_specific_steps(self) -> List[str]:
        return [
            "RUN git clone https://github.com/example/my-impl.git",
            "WORKDIR /my-impl",
            "RUN make install"
        ]

Predefined Builders¤

The system includes predefined builders for common patterns:

from panther.plugins.services.base.docker_build_base import QUIC_DOCKER_BUILDERS

# Use predefined builder
picoquic_builder = QUIC_DOCKER_BUILDERS["picoquic"]
dockerfile_content = picoquic_builder.generate_complete_dockerfile()

# Create custom builder
from panther.plugins.services.base.docker_build_base import DockerBuilderFactory

builder = DockerBuilderFactory.create_builder(
    implementation_name="my_quic",
    language="rust",
    repo_url="https://github.com/example/my-quic.git",
    cargo_features=["async", "crypto"]
)

Service Command Builder¤

Location: service_command_builder.py

Provides utilities for building complex service commands:

from panther.plugins.services.base.service_command_builder import ServiceCommandBuilder

builder = ServiceCommandBuilder()

# Build certificate arguments
cert_args = builder.build_certificate_args(
    cert_file="/app/certs/cert.pem",
    key_file="/app/certs/key.pem",
    cert_param="--cert",
    key_param="--key"
)

# Build protocol arguments
protocol_args = builder.build_protocol_args(
    protocol="quic",
    version="rfc9000",
    alpn="h3"
)

# Combine arguments
full_command = builder.combine_args([cert_args, protocol_args, custom_args])

Migration Guide¤

From Legacy Implementation to Base Classes¤

Before (Legacy Implementation): 

class PicoquicServiceManager(IImplementationManager):

    def generate_run_command(self, **kwargs):
        # 200+ lines of duplicated command generation
        command = []

        # Certificate handling (duplicated across all implementations)
        if kwargs.get("cert_file"):
            command.extend(["--cert", kwargs["cert_file"]])
        if kwargs.get("key_file"):
            command.extend(["--key", kwargs["key_file"]])

        # Protocol arguments (duplicated across all implementations)
        if kwargs.get("role") == "server":
            command.extend(["-p", str(kwargs.get("port", 4443))])
        else:
            host = kwargs.get("host", "localhost")
            port = kwargs.get("port", 4443)
            command.append(f"{host}:{port}")

        # ... 180+ more lines of duplicated logic

        return " ".join(command)

After (Base Class Implementation): 

class PicoquicServiceManager(BaseQUICServiceManager):

    def _get_implementation_name(self) -> str:
        return "picoquic"

    def _get_binary_name(self) -> str:
        return "picoquicdemo"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        return ["-p", str(kwargs.get("port", 4443))]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        host = kwargs.get("host", "localhost")
        port = kwargs.get("port", 4443)
        return [f"{host}:{port}"]

    def generate_deployment_commands(self) -> str:
        return f"{self._get_binary_name()} -p 4443"

    def _do_prepare(self, plugin_manager=None):
        pass

    # All certificate, protocol, and common logic inherited!
    # ~40 lines instead of 200+

Code Reduction Results¤

Implementation Before (lines) After (lines) Reduction
PicoQUIC 267 89 66.7%
AioQUIC 245 76 69.0%
Quiche 298 92 69.1%
Quinn 234 78 66.7%
LsQUIC 312 118 62.2%
QUIC-Go 189 71 62.4%
mvfst 276 95 65.6%
Quant 198 84 57.6%

Average: 47.2% code reduction

Development Patterns¤

Creating New Protocol Implementations¤

Creating New QUIC Implementation¤

  1. Choose Base Class:
  2. Python implementation → PythonQUICServiceManager
  3. Rust implementation → RustQUICServiceManager

  4. C/Go implementation → BaseQUICServiceManager

  5. Implement Required Methods: 

    class NewQuicServiceManager(BaseQUICServiceManager):

    def _get_implementation_name(self) -> str:
        return "new_quic"

    def _get_binary_name(self) -> str:
        return "new_quic_binary"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        # Only implementation-specific server arguments
        return ["-listen", f"0.0.0.0:{kwargs.get('port', 4443)}"]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        # Only implementation-specific client arguments
        return ["-connect", f"{kwargs.get('host')}:{kwargs.get('port')}"]

    def generate_deployment_commands(self) -> str:
        return f"cd /app && {self._get_binary_name()}"

    def _do_prepare(self, plugin_manager=None):
        # Build Docker image, install dependencies, etc.
        pass

Creating New HTTP Implementation¤

  1. Inherit from BaseHTTPServiceManager: 
    from panther.plugins.services.base.http_service_base import BaseHTTPServiceManager

class MyHttpServiceManager(BaseHTTPServiceManager):

    def _get_implementation_name(self) -> str:
        return "my_http"

    def _get_binary_name(self) -> str:
        return "my_http_server"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        return ["--foreground"]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        return ["--output", "/dev/null"]

    def generate_deployment_commands(self) -> str:
        return f"{self._get_binary_name()}"

    def _do_prepare(self, plugin_manager=None):
        pass

Creating New MINIP Implementation¤

  1. Inherit from BaseMinipServiceManager: 

    from panther.plugins.services.base.minip_service_base import BaseMinipServiceManager

class MyMinipServiceManager(BaseMinipServiceManager):

    def _get_implementation_name(self) -> str:
        return "my_minip"

    def _get_binary_name(self) -> str:
        return "my_minip_binary"

    def _get_server_specific_args(self, **kwargs) -> List[str]:
        return ["--server-mode"]

    def _get_client_specific_args(self, **kwargs) -> List[str]:
        return ["--client-mode"]

    def generate_deployment_commands(self) -> str:
        return f"{self._get_binary_name()}"

    def _do_prepare(self, plugin_manager=None):
        pass

  2. Test Implementation: 

    # Use the standard test configuration
panther --experiment-config test_new_implementation.yaml

Extending Base Classes¤

class EnhancedQuicServiceManager(BaseQUICServiceManager):

    def _extract_common_params(self, **kwargs) -> Dict[str, Any]:
        """Add custom parameters."""
        params = super()._extract_common_params(**kwargs)

        # Add implementation-specific parameters
        params["custom_option"] = kwargs.get("custom_option", "default")
        params["performance_mode"] = kwargs.get("performance_mode", False)

        return params

    def _build_server_args(self, params: Dict[str, Any]) -> List[str]:
        """Customize server argument building."""
        args = super()._build_server_args(params)

        # Add custom arguments
        if params.get("performance_mode"):
            args.extend(["--performance", "--no-debug"])

        return args

    def get_supported_features(self) -> List[str]:
        """Specify implementation capabilities."""
        return ["rfc9000", "0rtt", "connection_migration", "multipath"]

Testing Base Classes¤

Unit Testing¤

Testing QUIC Base Classes¤

import unittest
from panther.plugins.services.base.quic_service_base import BaseQUICServiceManager

class TestBaseQUICServiceManager(unittest.TestCase):

    def setUp(self):
        # Create test implementation
        class TestQuicManager(BaseQUICServiceManager):
            def _get_implementation_name(self) -> str:
                return "test_quic"
            def _get_binary_name(self) -> str:
                return "test_binary"
            # ... implement other abstract methods

        self.manager = TestQuicManager(
            service_config_to_test=mock_config,
            service_type="iut",
            protocol=mock_protocol,
            implementation_name="test_quic"
        )

    def test_command_generation(self):
        command = self.manager.generate_run_command(
            role="server",
            port=4443,
            cert_file="/app/certs/cert.pem"
        )
        self.assertIn("--cert", command)
        self.assertIn("4443", command)

Testing HTTP Base Classes¤

import unittest
from panther.plugins.services.base.http_service_base import BaseHTTPServiceManager

class TestBaseHTTPServiceManager(unittest.TestCase):

    def setUp(self):
        class TestHTTPManager(BaseHTTPServiceManager):
            def _get_implementation_name(self) -> str:
                return "test_http"
            def _get_binary_name(self) -> str:
                return "test_httpd"
            def _get_server_specific_args(self, **kwargs) -> List[str]:
                return ["-D", "FOREGROUND"]
            def _get_client_specific_args(self, **kwargs) -> List[str]:
                return ["-s", "-w", "%{http_code}"]
            def generate_deployment_commands(self) -> str:
                return "test_httpd"
            def _do_prepare(self, plugin_manager=None):
                pass

        self.manager = TestHTTPManager(
            service_config_to_test=mock_config,
            service_type="iut",
            protocol=mock_protocol,
            implementation_name="test_http"
        )

    def test_http_command_generation(self):
        command = self.manager.generate_run_command(
            role="server",
            port=80,
            protocol_version="1.1"
        )
        self.assertIn("--server", command)
        self.assertIn("80", command)

    def test_http_feature_detection(self):
        features = self.manager.get_supported_features()
        self.assertTrue(features["http_1_1"])
        self.assertEqual(self.manager.get_default_port(), 80)

Testing MINIP Base Classes¤

import unittest
from panther.plugins.services.base.minip_service_base import BaseMinipServiceManager

class TestBaseMinipServiceManager(unittest.TestCase):

    def setUp(self):
        class TestMinipManager(BaseMinipServiceManager):
            def _get_implementation_name(self) -> str:
                return "test_minip"
            def _get_binary_name(self) -> str:
                return "test_minip_binary"
            def _get_server_specific_args(self, **kwargs) -> List[str]:
                return ["--echo"]
            def _get_client_specific_args(self, **kwargs) -> List[str]:
                return ["--count", "10"]
            def generate_deployment_commands(self) -> str:
                return "test_minip_binary --daemon"
            def _do_prepare(self, plugin_manager=None):
                pass

        self.manager = TestMinipManager(
            service_config_to_test=mock_config,
            service_type="iut",
            protocol=mock_protocol,
            implementation_name="test_minip"
        )

    def test_minip_command_generation(self):
        command = self.manager.generate_run_command(
            role="server",
            port=8000,
            connection_type="tcp"
        )
        self.assertIn("--server", command)
        self.assertIn("8000", command)

    def test_minip_utilities(self):
        msg = self.manager.generate_test_message(100)
        self.assertTrue(self.manager.validate_message_format(msg))
        self.assertLessEqual(len(msg), 100)

        overhead = self.manager.calculate_message_overhead(1000)
        self.assertGreater(overhead, 0)

Integration Testing¤

# Test with real configuration
config = {
    "services": {
        "server": {
            "implementation": {"name": "test_quic", "type": "iut"},
            "protocol": {"name": "quic", "version": "rfc9000", "role": "server"}
        }
    }
}

# Run integration test
experiment_manager.run_tests()

Best Practices¤

Implementation Development¤

  1. Start Simple - Implement only required abstract methods first
  2. Test Early - Test with basic configuration before adding features
  3. Leverage Base - Use base class functionality rather than reimplementing
  4. Document Differences - Clearly document implementation-specific behavior
  5. Follow Patterns - Use the same patterns as existing implementations

Maintenance¤

  1. Update Base Classes - Enhancements benefit all implementations
  2. Consistent Testing - Test all implementations when changing base classes
  3. Version Compatibility - Ensure new features work across implementations
  4. Performance Monitoring - Monitor base class performance impact

Code Quality¤

  1. Single Responsibility - Each method has one clear purpose
  2. Template Method - Use the pattern correctly - don't override main methods
  3. Error Handling - Implement proper error handling in abstract methods
  4. Documentation - Document all abstract method requirements

Troubleshooting¤

Common Issues¤

Abstract Method Not Implemented: 

TypeError: Can't instantiate abstract class MyQuicManager with abstract method _get_implementation_name
Solution: Implement all required abstract methods for the base class you're inheriting from

Base Class Import Errors: 

ImportError: cannot import name 'BaseQUICServiceManager' from 'panther.plugins.services.base'
Solution: Use correct import paths: - from panther.plugins.services.base.quic_service_base import BaseQUICServiceManager - from panther.plugins.services.base.http_service_base import BaseHTTPServiceManager - from panther.plugins.services.base.minip_service_base import BaseMinipServiceManager

Command Generation Issues: 

# Check that you're returning correct types
def _get_server_specific_args(self, **kwargs) -> List[str]:
    # Return list of strings, not a single string
    return ["-p", str(kwargs.get("port", 4443))]  # Correct
    # return f"-p {kwargs.get('port', 4443)}"     # Wrong

Protocol-Specific Issues:

HTTP Port Conflicts: 

# HTTP uses port 80 by default, HTTPS uses 443
def _extract_common_params(self, **kwargs) -> Dict[str, Any]:
    default_port = 443 if kwargs.get("tls", False) else 80
    return {"port": kwargs.get("port", default_port)}

MINIP Connection Type Errors: 

# Validate connection type in MINIP implementations
def _extract_common_params(self, **kwargs) -> Dict[str, Any]:
    conn_type = kwargs.get("connection_type", "tcp")
    if conn_type not in ["tcp", "udp"]:
        raise ValueError(f"Invalid connection type: {conn_type}")
    return {"connection_type": conn_type}

Feature Detection Issues: 

# Override get_supported_features() to specify actual capabilities
def get_supported_features(self) -> Dict[str, bool]:
    return {
        "http_1_1": True,
        "http_2": False,  # Set to True if your implementation supports HTTP/2
        "tls": True,      # Set based on actual TLS support
    }

Future Development¤

The base class architecture enables:

  1. Easy Extension - Add new protocol support with minimal code
  2. Feature Consistency - New features automatically available to all implementations
  3. Maintenance Efficiency - Single point for bug fixes and improvements
  4. Performance Optimization - Optimize base classes to benefit all implementations
  5. Protocol Expansion - HTTP and MINIP base classes ready for new implementations
  6. Cross-Protocol Features - Shared utilities across QUIC, HTTP, and MINIP protocols

Protocol Support Status¤

Protocol Base Class Status Current Implementations
QUIC BaseQUICServiceManager ✅ Complete 8 implementations (PicoQUIC, AioQUIC, Quiche, Quinn, etc.)
MINIP BaseMinipServiceManager ✅ Created 1 implementation (ping_pong)
HTTP BaseHTTPServiceManager ✅ Created 0 implementations (base class available)

Current State¤

  • QUIC Support: Fully mature with 8 working implementations using inheritance-based architecture
  • MINIP Support: Base class created, one working implementation (ping_pong)
  • HTTP Support: Base class created, ready for HTTP implementations

This inheritance-based architecture represents a significant improvement in code maintainability, consistency, and development efficiency for the PANTHER framework, now supporting multiple protocol families with consistent patterns.