Fuzz CANBUS

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
CANBUS Fuzzer - Fuzzes CAN bus with various message patterns
Author: KittySploit Team
Version: 1.0.0
"""

from kittysploit import *
from core.output_handler import print_info, print_success, print_error, print_warning
import time
import random

class Module(Post):
    """Fuzz CAN bus with various message patterns"""
    
    __info__ = {
        "name": "Fuzz CANBUS",
        "description": "Fuzzes CAN bus with various message patterns to discover vulnerabilities",
        "author": "KittySploit Team",
        "version": "1.0.0",
        "session_type": SessionType.CANBUS,
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': ['rce', 'shell', 'file_read'],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [],
         'consumes_capabilities': ['shell'],
         'option_bindings': {},
         'suggested_followups': []},
    },
    }
    
    can_id = OptString("", "CAN ID to fuzz (hex format, e.g., 0x123). Empty = random IDs", required=False)
    fuzz_type = OptChoice("data", "Fuzzing type", required=True, choices=["data", "id", "both", "timing"])
    iterations = OptInteger(100, "Number of fuzzing iterations", required=True)
    delay = OptFloat(0.01, "Delay between messages in seconds", required=True)
    data_length = OptInteger(8, "Data length in bytes (for data fuzzing)", required=True)
    
    def check(self):
        """Check if session is a CANBUS session and if python-can is available"""
        try:
            import can
        except ImportError:
            print_error("python-can is required but not installed")
            print_info("Install it with: pip install python-can")
            return False
        
        session_id_value = str(self.session_id)
        if not session_id_value:
            print_error("Session ID not set")
            return False
        
        if self.framework and hasattr(self.framework, 'session_manager'):
            session = self.framework.session_manager.get_session(session_id_value)
            if session:
                if session.session_type == 'canbus':
                    return True
                else:
                    print_error(f"Session is not a CANBUS session (type: {session.session_type})")
                    return False
            else:
                print_error("Session not found")
                return False
        else:
            print_warning("Session manager not available - assuming valid session")
            return True
    
    def run(self):
        """Fuzz CAN bus"""
        try:
            import can
            
            session_id_value = str(self.session_id)
            
            if not self.framework or not hasattr(self.framework, 'session_manager'):
                print_error("Framework or session manager not available")
                return False
            
            session = self.framework.session_manager.get_session(session_id_value)
            if not session:
                print_error("Session not found")
                return False
            
            # Get CAN bus configuration from session
            interface = session.data.get('interface') if session.data else 'socketcan'
            channel = session.data.get('channel') if session.data else 'can0'
            bitrate = session.data.get('bitrate') if session.data else 500000
            
            # Parse CAN ID if specified
            can_id = None
            if self.can_id:
                can_id_str = str(self.can_id)
                if can_id_str.startswith('0x') or can_id_str.startswith('0X'):
                    can_id = int(can_id_str, 16)
                else:
                    can_id = int(can_id_str, 16) if all(c in '0123456789ABCDEFabcdef' for c in can_id_str) else int(can_id_str)
            
            print_info("Starting CANBUS fuzzing...")
            print_info("=" * 80)
            print_info(f"Fuzzing type: {self.fuzz_type}")
            print_info(f"Iterations: {self.iterations}")
            print_info(f"Delay: {self.delay} seconds")
            if can_id:
                print_info(f"Target CAN ID: 0x{can_id:03X}")
            else:
                print_info("Target CAN ID: Random")
            print_warning("⚠️  Fuzzing may affect vehicle/device behavior!")
            print_info("")
            
            # Connect to CAN bus
            try:
                if interface == "socketcan":
                    bus = can.interface.Bus(channel=channel, bustype='socketcan')
                elif interface == "virtual":
                    bus = can.interface.Bus(channel=channel, bustype='virtual')
                else:
                    bus = can.interface.Bus(channel=channel, bustype=interface, bitrate=bitrate)
            except Exception as e:
                print_error(f"Failed to connect to CAN bus: {e}")
                return False
            
            # Fuzz based on type
            fuzz_type = str(self.fuzz_type)
            iterations = int(self.iterations)
            delay = float(self.delay)
            data_length = int(self.data_length)
            
            sent = 0
            for i in range(iterations):
                try:
                    # Generate fuzzed message
                    message = self._generate_fuzzed_message(can_id, fuzz_type, data_length)
                    
                    bus.send(message)
                    sent += 1
                    
                    if (i + 1) % 10 == 0:
                        print_info(f"Sent {i+1}/{iterations} messages...")
                    
                    time.sleep(delay)
                except Exception as e:
                    print_error(f"Error sending fuzzed message {i+1}: {e}")
            
            bus.shutdown()
            
            print_success(f"\nFuzzing completed: {sent}/{iterations} messages sent")
            return True
            
        except Exception as e:
            print_error(f"Error fuzzing CAN bus: {e}")
            import traceback
            traceback.print_exc()
            return False
    
    def _generate_fuzzed_message(self, can_id, fuzz_type, data_length):
        """Generate a fuzzed CAN message"""
        import can
        
        # Determine CAN ID
        if fuzz_type in ["id", "both"]:
            if can_id is None:
                # Random CAN ID
                fuzzed_id = random.randint(0, 0x7FF)  # Standard 11-bit
            else:
                # Fuzz around the base ID
                fuzzed_id = can_id + random.randint(-10, 10)
                fuzzed_id = max(0, min(0x7FF, fuzzed_id))
        else:
            fuzzed_id = can_id if can_id is not None else random.randint(0, 0x7FF)
        
        # Determine data
        if fuzz_type in ["data", "both"]:
            # Generate random data
            data = bytes([random.randint(0, 255) for _ in range(data_length)])
        else:
            # Use pattern data
            data = bytes([0xAA] * data_length)
        
        # Create message
        message = can.Message(
            arbitration_id=fuzzed_id,
            data=data,
            is_extended_id=False,
            is_remote_frame=False
        )
        
        return message