Files
ethanseow2002@gmail.com 08c282335a first commit
2026-01-27 17:23:23 +08:00

836 lines
30 KiB
Python

"""
Malleable - Python Server Implementation
Framework-agnostic HTTP transformation library implementing Cobalt Strike's
Malleable C2 profile format for security research and testing.
"""
import re
import base64
import struct
from typing import Optional, Dict, List, Tuple, Any
import binascii
from dataclasses import dataclass
from enum import Enum
def _strip_comments(content: str) -> str:
result = []
in_string = False
i = 0
while i < len(content):
ch = content[i]
if in_string:
if ch == '\\' and i + 1 < len(content):
result.append(ch)
result.append(content[i + 1])
i += 2
continue
if ch == '"':
in_string = False
result.append(ch)
else:
if ch == '"':
in_string = True
result.append(ch)
elif ch == '#':
while i < len(content) and content[i] != '\n':
i += 1
result.append('\n' if i < len(content) and content[i] == '\n' else '')
else:
result.append(ch)
i += 1
return ''.join(result)
def _decode_string_str(s: str) -> str:
return TransformEngine._decode_string(s).decode('utf-8', errors='replace')
# =========================================================================
# DATA STRUCTURES
# =========================================================================
class TransformType(Enum):
"""Transform types supported in malleable profiles"""
BASE64 = "base64"
BASE64URL = "base64url"
NETBIOS = "netbios"
NETBIOSU = "netbiosu"
MASK = "mask"
PREPEND = "prepend"
APPEND = "append"
class TerminationType(Enum):
"""Termination statement types"""
HEADER = "header"
PARAMETER = "parameter"
PRINT = "print"
URI_APPEND = "uri-append"
@dataclass
class Transform:
"""A single transform step"""
type: TransformType
argument: Optional[str] = None
@dataclass
class Termination:
"""Termination statement"""
type: TerminationType
target: Optional[str] = None
@dataclass
class TransformChain:
"""Complete data transformation pipeline"""
transforms: List[Transform]
termination: Termination
@dataclass
class HttpConfig:
"""HTTP transaction configuration"""
headers: Dict[str, str]
parameters: Dict[str, str]
metadata: Optional[TransformChain] = None
id: Optional[TransformChain] = None
output: Optional[TransformChain] = None
@dataclass
class HttpTransaction:
"""Complete HTTP transaction definition"""
variant: str
method: str # GET or POST
uris: List[str]
client: HttpConfig
server: HttpConfig
@dataclass
class HttpRequest:
"""HTTP request structure"""
method: str
uri: str
path: str
query: Dict[str, str]
headers: Dict[str, str]
body: bytes
@dataclass
class HttpResponse:
"""HTTP response structure"""
status_code: int
headers: Dict[str, str]
body: bytes
class ProfileMismatchError(Exception):
"""Raised when a request does not match the profile or transforms fail."""
def __init__(self, reason: str, details: str):
super().__init__(f"{reason}: {details}")
self.reason = reason
self.details = details
class MalleableProfile:
"""
Malleable C2 profile
This class parses profiles from memory (string). If you need to load from a file,
read it into a string first and pass it to this constructor.
Example - Embedded profile:
profile_str = 'set sample_name "test"; ...'
profile = MalleableProfile(profile_str)
Example - Load from file (user-managed):
with open('config.profile', 'r') as f:
profile = MalleableProfile(f.read())
"""
def __init__(self, profile_content: str):
"""
Initialize malleable profile from string
Args:
profile_content: Profile content as string
"""
self.profile_name: Optional[str] = None
self.useragent: Optional[str] = None
self.headers_remove: List[str] = []
self.http_get_transactions: Dict[str, HttpTransaction] = {}
self.http_post_transactions: Dict[str, HttpTransaction] = {}
self._parse(profile_content)
def _parse(self, content: str):
"""Parse malleable profile content"""
# Remove comments
original_len = len(content)
hash_in_quotes = 0
in_string = False
i = 0
while i < len(content):
ch = content[i]
if in_string:
if ch == '\\' and i + 1 < len(content):
i += 2
continue
if ch == '"':
in_string = False
elif ch == '#':
hash_in_quotes += 1
else:
if ch == '"':
in_string = True
i += 1
content = _strip_comments(content)
# Extract profile name
match = re.search(r'set\s+sample_name\s+"((?:[^"\\]|\\.)*)"', content)
if match:
self.profile_name = _decode_string_str(match.group(1))
# Extract useragent
match = re.search(r'set\s+useragent\s+"((?:[^"\\]|\\.)*)"', content)
if match:
self.useragent = _decode_string_str(match.group(1))
match = re.search(r'set\s+headers_remove\s+"((?:[^"\\]|\\.)*)"', content)
if match:
raw_headers = _decode_string_str(match.group(1))
self.headers_remove = [h.strip() for h in raw_headers.split(',') if h.strip()]
# Parse http-get blocks
for variant, block_content in self._find_named_blocks(content, "http-get"):
transaction = self._parse_http_transaction("GET", variant, block_content)
if transaction:
self.http_get_transactions[variant] = transaction
# Parse http-post blocks
for variant, block_content in self._find_named_blocks(content, "http-post"):
transaction = self._parse_http_transaction("POST", variant, block_content)
if transaction:
self.http_post_transactions[variant] = transaction
def _parse_http_transaction(self, method: str, variant: str, content: str) -> Optional[HttpTransaction]:
"""Parse a single HTTP transaction block"""
uris = []
verb = method
client_config = HttpConfig(headers={}, parameters={})
server_config = HttpConfig(headers={}, parameters={})
# Extract URIs
match = re.search(r'set\s+uri\s+"((?:[^"\\]|\\.)*)"', content)
if match:
uri_str = _decode_string_str(match.group(1))
uris = uri_str.split()
match = re.search(r'set\s+verb\s+"((?:[^"\\]|\\.)*)"', content)
if match:
verb = _decode_string_str(match.group(1)).upper()
# Parse client block
client_content = self._extract_block_content(content, "client")
if client_content is not None:
self._parse_config_block(client_content, client_config)
# Parse server block
server_content = self._extract_block_content(content, "server")
if server_content is not None:
self._parse_config_block(server_content, server_config)
return HttpTransaction(
variant=variant,
method=verb,
uris=uris,
client=client_config,
server=server_config
)
def _parse_config_block(self, content: str, config: HttpConfig):
"""Parse client or server configuration block"""
# Parse headers
header_matches = list(re.finditer(r'header\s+"((?:[^"\\]|\\.)*)"\s+"((?:[^"\\]|\\.)*)"', content))
for match in header_matches:
name = _decode_string_str(match.group(1))
value = _decode_string_str(match.group(2))
config.headers[name] = value
# Parse parameters
param_matches = list(re.finditer(r'parameter\s+"((?:[^"\\]|\\.)*)"\s+"((?:[^"\\]|\\.)*)"', content))
for match in param_matches:
key = _decode_string_str(match.group(1))
value = _decode_string_str(match.group(2))
config.parameters[key] = value
# Parse metadata block
metadata_content = self._extract_block_content(content, "metadata")
if metadata_content is not None:
config.metadata = self._parse_transform_chain(metadata_content)
# Parse id block
id_content = self._extract_block_content(content, "id")
if id_content is not None:
config.id = self._parse_transform_chain(id_content)
# Parse output block
output_content = self._extract_block_content(content, "output")
if output_content is not None:
config.output = self._parse_transform_chain(output_content)
def _find_named_blocks(self, content: str, keyword: str) -> List[Tuple[str, str]]:
"""Find blocks like 'http-get \"variant\" { ... }' with brace matching."""
results = []
idx = 0
while True:
pos = content.find(keyword, idx)
if pos == -1:
break
if pos > 0 and content[pos - 1].isalnum():
idx = pos + len(keyword)
continue
i = pos + len(keyword)
while i < len(content) and content[i].isspace():
i += 1
variant = "default"
if i < len(content) and content[i] == '"':
i += 1
start = i
while i < len(content) and content[i] != '"':
if content[i] == '\\' and i + 1 < len(content):
i += 2
continue
i += 1
variant = content[start:i]
i += 1
while i < len(content) and content[i].isspace():
i += 1
if i >= len(content) or content[i] != '{':
idx = pos + len(keyword)
continue
block_content, end_pos = self._extract_braced_content(content, i)
if block_content is not None:
results.append((variant, block_content))
idx = end_pos
else:
idx = pos + len(keyword)
return results
def _extract_block_content(self, content: str, keyword: str) -> Optional[str]:
"""Extract first block content for a given keyword, using brace matching."""
pos = content.find(keyword)
if pos == -1:
return None
i = pos + len(keyword)
while i < len(content) and content[i].isspace():
i += 1
if i >= len(content) or content[i] != '{':
return None
block_content, _ = self._extract_braced_content(content, i)
return block_content
def _extract_braced_content(self, content: str, brace_pos: int) -> Tuple[Optional[str], int]:
"""Extract content inside braces starting at brace_pos (which must be '{')."""
depth = 0
i = brace_pos
start = brace_pos + 1
in_string = False
while i < len(content):
ch = content[i]
if in_string:
if ch == '\\' and i + 1 < len(content):
i += 2
continue
if ch == '"':
in_string = False
else:
if ch == '"':
in_string = True
elif ch == '{':
depth += 1
elif ch == '}':
depth -= 1
if depth == 0:
return content[start:i], i + 1
i += 1
return None, len(content)
def _parse_transform_chain(self, content: str) -> TransformChain:
"""Parse a transformation chain"""
transforms = []
termination = None
transform_arg_has_backslash = False
termination_target_has_backslash = False
# Split into statements, ignoring semicolons inside quotes
statements = self._split_statements(content)
for stmt in statements:
stmt = stmt.strip()
if not stmt:
continue
# Check for termination statements
if stmt.startswith('header'):
match = re.match(r'header\s+"((?:[^"\\]|\\.)*)"', stmt)
if match:
termination_target_has_backslash = "\\" in match.group(1)
termination = Termination(TerminationType.HEADER, _decode_string_str(match.group(1)))
break
elif stmt.startswith('parameter'):
match = re.match(r'parameter\s+"((?:[^"\\]|\\.)*)"', stmt)
if match:
termination_target_has_backslash = "\\" in match.group(1)
termination = Termination(TerminationType.PARAMETER, _decode_string_str(match.group(1)))
break
elif stmt == 'print':
termination = Termination(TerminationType.PRINT)
break
elif stmt == 'uri-append':
termination = Termination(TerminationType.URI_APPEND)
break
# Check for transform statements
elif stmt.startswith('prepend'):
match = re.match(r'prepend\s+"((?:[^"\\]|\\.)*)"', stmt)
if match:
transform_arg_has_backslash = transform_arg_has_backslash or ("\\" in match.group(1))
transforms.append(Transform(TransformType.PREPEND, match.group(1)))
elif stmt.startswith('append'):
match = re.match(r'append\s+"((?:[^"\\]|\\.)*)"', stmt)
if match:
transform_arg_has_backslash = transform_arg_has_backslash or ("\\" in match.group(1))
transforms.append(Transform(TransformType.APPEND, match.group(1)))
elif stmt == 'base64':
transforms.append(Transform(TransformType.BASE64))
elif stmt == 'base64url':
transforms.append(Transform(TransformType.BASE64URL))
elif stmt == 'netbios':
transforms.append(Transform(TransformType.NETBIOS))
elif stmt == 'netbiosu':
transforms.append(Transform(TransformType.NETBIOSU))
elif stmt == 'mask':
transforms.append(Transform(TransformType.MASK))
return TransformChain(transforms=transforms, termination=termination or Termination(TerminationType.PRINT))
def _split_statements(self, content: str) -> List[str]:
"""Split semicolon-delimited statements, respecting quoted strings."""
statements = []
buf = []
in_string = False
i = 0
while i < len(content):
ch = content[i]
if in_string:
if ch == '\\' and i + 1 < len(content):
buf.append(ch)
buf.append(content[i + 1])
i += 2
continue
if ch == '"':
in_string = False
buf.append(ch)
else:
if ch == '"':
in_string = True
buf.append(ch)
elif ch == ';':
stmt = ''.join(buf).strip()
if stmt:
statements.append(stmt)
buf = []
else:
buf.append(ch)
i += 1
tail = ''.join(buf).strip()
if tail:
statements.append(tail)
return statements
# =========================================================================
# TRANSFORMATION FUNCTIONS
# =========================================================================
class TransformEngine:
"""Handles data transformation according to malleable profiles"""
@staticmethod
def apply_transform(data: bytes, transform: Transform) -> bytes:
"""Apply a single transform (forward direction)"""
if transform.type == TransformType.BASE64:
return base64.b64encode(data)
elif transform.type == TransformType.BASE64URL:
return base64.urlsafe_b64encode(data).rstrip(b'=')
elif transform.type == TransformType.NETBIOS:
return TransformEngine._netbios_encode(data, b'a')
elif transform.type == TransformType.NETBIOSU:
return TransformEngine._netbios_encode(data, b'A')
elif transform.type == TransformType.MASK:
return TransformEngine._mask_encode(data)
elif transform.type == TransformType.PREPEND:
if transform.argument:
arg = TransformEngine._decode_string(transform.argument)
return arg + data
return data
elif transform.type == TransformType.APPEND:
if transform.argument:
arg = TransformEngine._decode_string(transform.argument)
return data + arg
return data
return data
@staticmethod
def reverse_transform(data: bytes, transform: Transform) -> bytes:
"""Reverse a single transform (backward direction)"""
if transform.type == TransformType.BASE64:
try:
return base64.b64decode(data)
except binascii.Error as exc:
raise ProfileMismatchError("transform", f"base64 decode failed: {exc}") from exc
elif transform.type == TransformType.BASE64URL:
# Add padding if needed
padding = (4 - len(data) % 4) % 4
data = data + b'=' * padding
try:
return base64.urlsafe_b64decode(data)
except binascii.Error as exc:
raise ProfileMismatchError("transform", f"base64url decode failed: {exc}") from exc
elif transform.type == TransformType.NETBIOS:
return TransformEngine._netbios_decode(data, b'a')
elif transform.type == TransformType.NETBIOSU:
return TransformEngine._netbios_decode(data, b'A')
elif transform.type == TransformType.MASK:
return TransformEngine._mask_decode(data)
elif transform.type == TransformType.PREPEND:
if transform.argument:
arg = TransformEngine._decode_string(transform.argument)
if data.startswith(arg):
return data[len(arg):]
raise ProfileMismatchError("transform", "prepend mismatch")
return data
elif transform.type == TransformType.APPEND:
if transform.argument:
arg = TransformEngine._decode_string(transform.argument)
if data.endswith(arg):
return data[:-len(arg)]
raise ProfileMismatchError("transform", "append mismatch")
return data
return data
@staticmethod
def apply_chain(data: bytes, chain: TransformChain) -> bytes:
"""Apply complete transformation chain (forward)"""
result = data
for transform in chain.transforms:
result = TransformEngine.apply_transform(result, transform)
return result
@staticmethod
def reverse_chain(data: bytes, chain: TransformChain) -> bytes:
"""Reverse complete transformation chain (backward)"""
result = data
# Apply transforms in REVERSE order
for transform in reversed(chain.transforms):
result = TransformEngine.reverse_transform(result, transform)
return result
@staticmethod
def _netbios_encode(data: bytes, base: bytes) -> bytes:
"""NetBIOS encoding"""
base_char = base[0]
result = bytearray()
for byte in data:
result.append(base_char + (byte >> 4))
result.append(base_char + (byte & 0x0F))
return bytes(result)
@staticmethod
def _netbios_decode(data: bytes, base: bytes) -> bytes:
"""NetBIOS decoding"""
base_char = base[0]
result = bytearray()
for i in range(0, len(data), 2):
if i + 1 < len(data):
high = data[i] - base_char
low = data[i + 1] - base_char
result.append((high << 4) | low)
return bytes(result)
@staticmethod
def _mask_encode(data: bytes) -> bytes:
"""XOR mask encoding with random key"""
import os
key = os.urandom(4)
masked = bytearray()
for i, byte in enumerate(data):
masked.append(byte ^ key[i % 4])
return key + bytes(masked)
@staticmethod
def _mask_decode(data: bytes) -> bytes:
"""XOR mask decoding"""
if len(data) < 4:
raise ProfileMismatchError("transform", "mask decode expects 4-byte key")
key = data[:4]
masked = data[4:]
result = bytearray()
for i, byte in enumerate(masked):
result.append(byte ^ key[i % 4])
return bytes(result)
@staticmethod
def _decode_string(s: str) -> bytes:
"""Decode string with escape sequences"""
result = bytearray()
i = 0
while i < len(s):
if s[i] == '\\' and i + 1 < len(s):
next_char = s[i + 1]
if next_char == 'n':
result.append(ord('\n'))
i += 2
elif next_char == 'r':
result.append(ord('\r'))
i += 2
elif next_char == 't':
result.append(ord('\t'))
i += 2
elif next_char == '\\':
result.append(ord('\\'))
i += 2
elif next_char == 'x' and i + 3 < len(s):
hex_str = s[i+2:i+4]
result.append(int(hex_str, 16))
i += 4
elif next_char == 'u' and i + 5 < len(s):
hex_str = s[i+2:i+6]
codepoint = int(hex_str, 16)
result.extend(chr(codepoint).encode('utf-8'))
i += 6
else:
result.append(ord(next_char))
i += 2
else:
result.append(ord(s[i]))
i += 1
return bytes(result)
# =========================================================================
# SERVER EXTRACTION FUNCTIONS
# =========================================================================
class MalleableServer:
"""Server-side extraction and response building"""
def __init__(self, profile: MalleableProfile):
self.profile = profile
self.engine = TransformEngine()
def extract_metadata(self, request: HttpRequest, variant: str = "default") -> bytes:
"""Extract metadata from HTTP GET request"""
transaction = self.profile.http_get_transactions.get(variant)
if not transaction or not transaction.client.metadata:
raise ValueError(f"No metadata chain for variant {variant}")
if not any(request.path.startswith(uri) for uri in transaction.uris):
raise ProfileMismatchError(
"uri",
f"request path '{request.path}' does not match any http-get uri"
)
chain = transaction.client.metadata
data = self._extract_from_request(request, chain.termination, transaction.uris)
return self.engine.reverse_chain(data, chain)
def extract_session_id(self, request: HttpRequest, variant: str = "default") -> bytes:
"""Extract session ID from HTTP POST request"""
transaction = self.profile.http_post_transactions.get(variant)
if not transaction or not transaction.client.id:
raise ValueError(f"No id chain for variant {variant}")
if not any(request.path.startswith(uri) for uri in transaction.uris):
raise ProfileMismatchError(
"uri",
f"request path '{request.path}' does not match any http-post uri"
)
chain = transaction.client.id
data = self._extract_from_request(request, chain.termination, transaction.uris)
return self.engine.reverse_chain(data, chain)
def extract_output(self, request: HttpRequest, variant: str = "default") -> bytes:
"""Extract output from HTTP POST request"""
transaction = self.profile.http_post_transactions.get(variant)
if not transaction or not transaction.client.output:
raise ValueError(f"No output chain for variant {variant}")
if not any(request.path.startswith(uri) for uri in transaction.uris):
raise ProfileMismatchError(
"uri",
f"request path '{request.path}' does not match any http-post uri"
)
chain = transaction.client.output
data = self._extract_from_request(request, chain.termination, transaction.uris)
return self.engine.reverse_chain(data, chain)
def build_get_response(self, data: bytes, variant: str = "default") -> HttpResponse:
"""Build HTTP response with tasks (response to GET)"""
transaction = self.profile.http_get_transactions.get(variant)
if not transaction:
raise ValueError(f"No transaction for variant {variant}")
headers = dict(transaction.server.headers)
body = b''
if transaction.server.output:
transformed = self.engine.apply_chain(data, transaction.server.output)
if transaction.server.output.termination.type == TerminationType.PRINT:
body = transformed
elif transaction.server.output.termination.type == TerminationType.HEADER:
headers[transaction.server.output.termination.target] = transformed.decode('utf-8', errors='ignore')
return HttpResponse(status_code=200, headers=headers, body=body)
def build_post_response(self, data: bytes = b'', variant: str = "default") -> HttpResponse:
"""Build HTTP response for POST"""
transaction = self.profile.http_post_transactions.get(variant)
if not transaction:
raise ValueError(f"No transaction for variant {variant}")
headers = dict(transaction.server.headers)
body = b''
if transaction.server.output and data:
transformed = self.engine.apply_chain(data, transaction.server.output)
if transaction.server.output.termination.type == TerminationType.PRINT:
body = transformed
return HttpResponse(status_code=200, headers=headers, body=body)
def _extract_from_request(
self,
request: HttpRequest,
termination: Termination,
uris: Optional[List[str]] = None
) -> bytes:
"""Extract data from HTTP request based on termination type"""
if termination.type == TerminationType.HEADER:
value = request.headers.get(termination.target, '')
return value.encode('utf-8')
elif termination.type == TerminationType.PARAMETER:
value = request.query.get(termination.target, '')
return value.encode('utf-8')
elif termination.type == TerminationType.PRINT:
return request.body
elif termination.type == TerminationType.URI_APPEND:
if not uris:
raise ProfileMismatchError("uri", "uri-append used but no uris configured")
for base in uris:
if request.path.startswith(base):
suffix = request.path[len(base):]
return suffix.encode('utf-8')
raise ProfileMismatchError(
"uri",
f"request path '{request.path}' does not match any uri-append base"
)
return b''
# =========================================================================
# HELPER FUNCTIONS
# =========================================================================
def parse_http_request(raw_request: str) -> HttpRequest:
"""Parse raw HTTP request string"""
lines = raw_request.split('\r\n')
if not lines:
raise ValueError("Empty request")
# Parse request line
request_line = lines[0].split(' ')
if len(request_line) < 3:
raise ValueError("Invalid request line")
method = request_line[0]
full_uri = request_line[1]
# Parse URI and query string
if '?' in full_uri:
path, query_string = full_uri.split('?', 1)
query = dict(param.split('=', 1) for param in query_string.split('&') if '=' in param)
else:
path = full_uri
query = {}
# Parse headers
headers = {}
i = 1
while i < len(lines) and lines[i]:
if ':' in lines[i]:
key, value = lines[i].split(':', 1)
headers[key.strip()] = value.strip()
i += 1
# Body is everything after blank line
body = b''
if i < len(lines):
body = '\r\n'.join(lines[i+1:]).encode('utf-8')
return HttpRequest(
method=method,
uri=full_uri,
path=path,
query=query,
headers=headers,
body=body
)
if __name__ == '__main__':
# Example usage
import sys
if len(sys.argv) < 2:
print("Usage: python openmalleable.py <profile_path>")
sys.exit(1)
# User handles file I/O, library only accepts string
with open(sys.argv[1], 'r') as f:
profile_content = f.read()
profile = MalleableProfile(profile_content)
print(f"Loaded profile: {profile.profile_name}")
print(f"HTTP-GET transactions: {list(profile.http_get_transactions.keys())}")
print(f"HTTP-POST transactions: {list(profile.http_post_transactions.keys())}")