Skip to content

Halberd Technique Development Guide

Jump to bottom
Arpan Sarkar edited this page Apr 15, 2025 · 2 revisions

This comprehensive reference will guide you through creating effective and well-structured attack techniques for the Halberd multi-cloud security testing tool.

Table of Contents

  1. Introduction to Halberd Techniques
  2. Technique Structure and Architecture
  3. Writing Effective Technique Code
  4. Documentation and Metadata
  5. Testing and Validation
  6. Contributing Guidelines
  7. Templates and Examples

Introduction to Halberd Techniques

Halberd is an open-source multi-cloud security testing tool designed to emulate real-world attacks across various cloud platforms (Entra ID, M365, Azure, AWS, and GCP). Each technique in Halberd represents a specific attack method that security professionals can use to test their defenses.

What Makes a Good Technique?

A good Halberd technique should be:

  • Realistic: Emulates actual attacker techniques seen in the wild
  • Educational: Teaches about a specific security vulnerability or attack vector
  • Usable: Easy to configure and execute with clear parameters and outputs
  • Well-documented: Provides context on what the technique does and why it matters
  • Safe: Contains appropriate safeguards to prevent unintended damage

Supported Cloud Platforms

Halberd currently supports techniques for:

  • Microsoft Entra ID
  • Microsoft 365
  • Microsoft Azure
  • Amazon Web Services (AWS)
  • Google Cloud Platform (GCP)

Technique Structure and Architecture

All Halberd techniques inherit from the BaseTechnique class defined in base_technique.py and are registered with the TechniqueRegistry to make them available in the application.

Basic Structure

from ..base_technique import BaseTechnique, ExecutionStatus, MitreTechnique
from ..technique_registry import TechniqueRegistry
from typing import Dict, Any, Tuple

@TechniqueRegistry.register
class MyNewTechnique(BaseTechnique):
    def __init__(self):
        mitre_techniques = [
            MitreTechnique(
                technique_id="T1234",
                technique_name="Example Technique",
                tactics=["Initial Access", "Persistence"],
                sub_technique_name="Example Sub-Technique"
            )
        ]
        super().__init__(
            "My Technique Name", 
            "Description of what this technique does...",
            mitre_techniques
        )

    def execute(self, **kwargs: Any) -> Tuple[ExecutionStatus, Dict[str, Any]]:
        # Validate input parameters
        self.validate_parameters(kwargs)
        
        try:
            # Implementation of technique
            # ...
            
            return ExecutionStatus.SUCCESS, {
                "message": "Successfully executed technique",
                "value": { /* Results */ }
            }
        except Exception as e:
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Failed to execute technique"
            }

    def get_parameters(self) -> Dict[str, Dict[str, Any]]:
        return {
            "param_name": {
                "type": "str", 
                "required": True, 
                "default": None, 
                "name": "Human Readable Name", 
                "input_field_type": "text"
            }
        }

Key Components

  1. Class Definition: Create a class that inherits from BaseTechnique
  2. Registration: Use @TechniqueRegistry.register decorator to register your technique
  3. MITRE ATT&CK Mapping: Link to relevant MITRE ATT&CK techniques
  4. Initialization: Set up technique metadata in __init__ method
  5. Parameters: Define input parameters in get_parameters method
  6. Execution Logic: Implement the technique in the execute method

Writing Effective Technique Code

The execute Method

The core of your technique is the execute method. This is where the actual attack happens.

def execute(self, **kwargs: Any) -> Tuple[ExecutionStatus, Dict[str, Any]]:
    # Always validate input parameters first
    self.validate_parameters(kwargs)
    
    try:
        # Extract parameters from kwargs
        param1 = kwargs.get("param1")
        param2 = kwargs.get("param2", "default_value")  # Optional param with default
        
        # Implement technique logic here
        # ...

        # Return success with results
        return ExecutionStatus.SUCCESS, {
            "message": "Successfully executed technique",
            "value": {
                "key1": "value1",
                "key2": "value2"
            }
        }
    except Exception as e:
        # Always handle exceptions and return error information
        return ExecutionStatus.FAILURE, {
            "error": str(e),
            "message": "Failed to execute technique: specific reason"
        }

Parameter Definition

Input parameters are defined in the get_parameters method:

def get_parameters(self) -> Dict[str, Dict[str, Any]]:
    return {
        "param_name": {
            "type": "str",  # Python type (str, int, bool, etc.)
            "required": True,  # Is this parameter required?
            "default": None,   # Default value if not provided
            "name": "Human Readable Parameter Name",  # Displayed to user
            "input_field_type": "text"  # UI input type
        }
    }

Available input_field_type options:

  • "text": Basic text input
  • "email": Email input
  • "password": Password input (masked)
  • "number": Numeric input
  • "bool": Boolean toggle
  • "upload": File upload
  • "select": Dropdown input

Error Handling

Proper error handling is crucial for technique reliability:

try:
    # Technique logic
except ClientError as e:
    # Handle specific API errors
    return ExecutionStatus.FAILURE, {
        "error": str(e),
        "message": "API error occurred: " + str(e)
    }
except ValueError as e:
    # Handle validation errors
    return ExecutionStatus.FAILURE, {
        "error": str(e),
        "message": "Invalid input: " + str(e)
    }
except Exception as e:
    # Catch any other errors
    return ExecutionStatus.FAILURE, {
        "error": str(e),
        "message": "Failed to execute technique"
    }

Documentation and Metadata

Good documentation is essential for useful techniques. Halberd supports several forms of metadata to document your technique.

Basic Metadata

Set in the __init__ method:

super().__init__(
    "My Technique Name",  # Displayed name
    "Detailed description of what this technique does...",  # Technique description
    mitre_techniques,  # MITRE ATT&CK mappings
    azure_trm_techniques,  # Azure Threat Research Matrix mappings (optional)
    references,  # List of TechniqueReference objects (optional)
    notes  # List of TechniqueNote objects (optional)
)

MITRE ATT&CK Mapping

mitre_techniques = [
    MitreTechnique(
        technique_id="T1078.004",
        technique_name="Valid Accounts",
        tactics=["Defense Evasion", "Persistence", "Privilege Escalation", "Initial Access"],
        sub_technique_name="Cloud Accounts"
    )
]

Azure Threat Research Matrix (ATRM) Mapping

For Azure techniques, you can add ATRM mappings:

azure_trm_technique = [
    AzureTRMTechnique(
        technique_id="AZT301.2",
        technique_name="Virtual Machine Scripting",
        tactics=["Execution"],
        sub_technique_name="CustomScriptExtension"
    )
]

References and Notes

References and notes provide additional context for users:

technique_refs = [
    TechniqueReference("Grant limited access to Azure Storage resources using shared access signatures (SAS)", 
                     "https://learn.microsoft.com/en-us/azure/storage/common/storage-sas-overview"),
    TechniqueReference("Azure Disk | Exfiltrate VM Disk", 
                     "https://zigmax.net/azure-disk-exfiltrate-vm-disk/")
]

technique_notes = [
    TechniqueNote("Combine with AzureShareVmDisk to obtain SAS URLs"),
    TechniqueNote("Use smaller block sizes on memory-constrained systems"),
    TechniqueNote("Increase timeout for slower network connections")
]

Testing and Validation

Before submitting your technique, test it thoroughly:

  1. Parameter Validation: Ensure all parameters are correctly validated
  2. Edge Cases: Test with both valid and invalid inputs
  3. Error Handling: Verify appropriate error messages are returned
  4. Output Format: Confirm the output follows the expected structure
  5. Cloud Integration: Test against actual cloud resources when possible

Contributing Guidelines

When contributing techniques to Halberd:

  1. Follow the standard Python PEP 8 style guidelines
  2. Use meaningful variable names that reflect their purpose
  3. Add comments for complex logic
  4. Include proper error handling
  5. Provide all necessary metadata (MITRE mappings, descriptions, etc.)
  6. Test the technique before submission

Templates and Examples

Standard Technique Template

from ..base_technique import BaseTechnique, ExecutionStatus, MitreTechnique, TechniqueReference, TechniqueNote
from ..technique_registry import TechniqueRegistry
from typing import Dict, Any, Tuple

@TechniqueRegistry.register
class ExampleTechnique(BaseTechnique):
    def __init__(self):
        mitre_techniques = [
            MitreTechnique(
                technique_id="T1234",
                technique_name="Example Technique",
                tactics=["Initial Access"],
                sub_technique_name=None
            )
        ]
        
        technique_refs = [
            TechniqueReference("Technique Documentation", 
                             "https://example.com/docs"),
            TechniqueReference("Related Research", 
                             "https://example.com/research")
        ]
        
        technique_notes = [
            TechniqueNote("Important note about usage"),
            TechniqueNote("Configuration requirements")
        ]
        
        super().__init__(
            "Example Technique", 
            "Detailed description of what this technique does and why it's useful...",
            mitre_techniques,
            references=technique_refs,
            notes=technique_notes
        )

    def execute(self, **kwargs: Any) -> Tuple[ExecutionStatus, Dict[str, Any]]:
        # Validate parameters
        self.validate_parameters(kwargs)
        
        try:
            # Extract parameters
            param1 = kwargs.get("param1")
            param2 = kwargs.get("param2", "default")
            
            # Implementation
            # ...
            
            # Return success with results
            return ExecutionStatus.SUCCESS, {
                "message": "Successfully executed technique",
                "value": {
                    "result1": "value1",
                    "result2": "value2"
                }
            }
            
        except ValueError as e:
            # Input validation error
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Invalid input: " + str(e)
            }
        except Exception as e:
            # General error
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Failed to execute technique"
            }

    def get_parameters(self) -> Dict[str, Dict[str, Any]]:
        return {
            "param1": {
                "type": "str", 
                "required": True, 
                "default": None, 
                "name": "Parameter 1", 
                "input_field_type": "text"
            },
            "param2": {
                "type": "int", 
                "required": False, 
                "default": 123, 
                "name": "Parameter 2", 
                "input_field_type": "number"
            },
            "param3": {
                "type": "bool", 
                "required": False, 
                "default": False, 
                "name": "Toggle Feature", 
                "input_field_type": "bool"
            }
        }

Entra ID Technique Example

from ..base_technique import BaseTechnique, ExecutionStatus, MitreTechnique
from ..technique_registry import TechniqueRegistry
from typing import Dict, Any, Tuple
from core.entra.graph_request import GraphRequest

@TechniqueRegistry.register
class EntraEnumerateUsers(BaseTechnique):
    def __init__(self):
        mitre_techniques = [
            MitreTechnique(
                technique_id="T1087.004",
                technique_name="Account Discovery",
                tactics=["Discovery"],
                sub_technique_name="Cloud Account"
            )
        ]
        super().__init__(
            "Enumerate Users", 
            "Enumerates users in Entra ID",
            mitre_techniques
        )

    def execute(self, **kwargs: Any) -> Tuple[ExecutionStatus, Dict[str, Any]]:
        self.validate_parameters(kwargs)
        
        try:
            endpoint_url = "https://graph.microsoft.com/v1.0/users/"
            
            raw_response = GraphRequest().get(url=endpoint_url)

            if 'error' in raw_response:
                return ExecutionStatus.FAILURE, {
                    "error": {"error_code": raw_response.get('error').get('code'),
                              "error_detail": raw_response.get('error').get('message')},
                    "message": "Failed to enumerate users in tenant"
                }

            output = []
            if raw_response:
                output = [({
                    'display_name': user_info.get('displayName', 'N/A'),
                    'upn': user_info.get('userPrincipalName', 'N/A'),
                    'mail': user_info.get('mail', 'N/A'),
                    'job_title': user_info.get('jobTitle', 'N/A'),
                    'mobile_phone': user_info.get('mobilePhone', 'N/A'),
                    'office_location': user_info.get('officeLocation', 'N/A'),
                    'id': user_info.get('id', 'N/A'),
                }) for user_info in raw_response]

                return ExecutionStatus.SUCCESS, {
                    "message": f"Successfully enumerated {len(output)} users",
                    "value": output
                }
            else:
                return ExecutionStatus.SUCCESS, {
                    "message": f"No users found",
                    "value": output
                }

        except Exception as e:
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Failed to enumerate users in tenant"
            }

    def get_parameters(self) -> Dict[str, Dict[str, Any]]:
        return {}  # No parameters required

AWS Technique Example

from ..base_technique import BaseTechnique, ExecutionStatus, MitreTechnique
from ..technique_registry import TechniqueRegistry
from typing import Dict, Any, Tuple
import boto3
from botocore.exceptions import ClientError

@TechniqueRegistry.register
class AWSEnumerateS3Buckets(BaseTechnique):
    def __init__(self):
        mitre_techniques = [
            MitreTechnique(
                technique_id="T1619",
                technique_name="Cloud Storage Object Discovery",
                tactics=["Discovery"],
                sub_technique_name=None
            )
        ]
        super().__init__(
            "Enumerate S3 Buckets", 
            "Enumerates S3 buckets in the target AWS account",
            mitre_techniques
        )

    def execute(self, **kwargs: Any) -> Tuple[ExecutionStatus, Dict[str, Any]]:
        self.validate_parameters(kwargs)
        try:
            # Initialize boto3 client
            my_client = boto3.client("s3")

            # Enumerate S3 buckets
            raw_response = my_client.list_buckets()

            if 200 <= raw_response['ResponseMetadata']['HTTPStatusCode'] < 300:
                # Create output
                buckets = [bucket['Name'] for bucket in raw_response['Buckets']]
                
                if buckets:
                    return ExecutionStatus.SUCCESS, {
                        "message": f"Successfully enumerated {len(buckets)} S3 buckets",
                        "value": buckets
                    }
            
            return ExecutionStatus.FAILURE, {
                "error": raw_response.get('ResponseMetadata', 'N/A'),
                "message": "Failed to enumerate S3 buckets"
            }
        except ClientError as e:
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Failed to enumerate S3 buckets"
            }
        except Exception as e:
            return ExecutionStatus.FAILURE, {
                "error": str(e),
                "message": "Failed to enumerate S3 buckets"
            }

    def get_parameters(self) -> Dict[str, Dict[str, Any]]:
        return {}  # No parameters required

By following these templates and guidelines, you can create effective Halberd techniques that help security professionals test and improve their cloud security posture.

Halberd Wiki

Clone this wiki locally