CANBUS OBD-II Query

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

"""
CANBUS OBD-II Query - Sends read-only OBD-II diagnostic queries over a CANBUS session
Author: KittySploit Team
Version: 1.0.0
"""

from kittysploit import *
from core.output_handler import print_info, print_success, print_error, print_warning, print_status
from datetime import datetime
import json
import time


PID_NAMES = {
    0x00: "Supported PIDs 01-20",
    0x01: "Monitor status",
    0x04: "Calculated engine load",
    0x05: "Engine coolant temperature",
    0x0C: "Engine RPM",
    0x0D: "Vehicle speed",
    0x0F: "Intake air temperature",
    0x10: "MAF air flow rate",
    0x11: "Throttle position",
    0x1C: "OBD standards",
    0x20: "Supported PIDs 21-40",
    0x40: "Supported PIDs 41-60",
    0x60: "Supported PIDs 61-80",
    0x80: "Supported PIDs 81-A0",
    0xA0: "Supported PIDs A1-C0",
}


class Module(Post):
    """Send read-only OBD-II queries via an active CANBUS session"""

    __info__ = {
        "name": "CANBUS OBD-II Query",
        "description": "Queries OBD-II data such as supported PIDs, VIN, custom PID values, and DTCs",
        "author": "KittySploit Team",
        "version": "1.0.0",
        "session_type": SessionType.CANBUS,
        "tags": ["canbus", "automotive", "obd2", "diagnostic"],
        "references": [
            "https://en.wikipedia.org/wiki/OBD-II_PIDs",
            "https://en.wikipedia.org/wiki/ISO_15765-2",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["recon", "active_exploitation"],
            "expected_requests": 4,
            "reversible": True,
            "approval_required": True,
            "produces": ["vehicle_diagnostics", "risk_signals"],
            "cost": 0.8,
            "noise": 0.2,
            "value": 1.3,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": [],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "vehicle_diagnostics", "from_detail": "obd2_results"},
                ],
                "consumes_capabilities": ["canbus_session"],
                "option_bindings": {},
                "suggested_followups": [
                    "post/canbus/gather/ecu_discovery",
                    "post/canbus/gather/dump_messages",
                ],
            },
        },
    }

    query = OptChoice(
        "supported_pids",
        "OBD-II query to run",
        required=True,
        choices=["supported_pids", "vin", "dtc", "pid"],
    )
    mode = OptString("01", "OBD-II service/mode for custom PID queries (hex, e.g., 01)", required=False)
    pid = OptString("0C", "OBD-II PID for custom PID queries (hex, e.g., 0C)", required=False)
    request_id = OptString("0x7DF", "OBD-II request CAN ID (functional 0x7DF or physical 0x7E0-0x7E7)", required=True)
    response_ids = OptString("0x7E8-0x7EF", "Accepted response CAN IDs or ranges", required=True)
    timeout = OptFloat(1.0, "Response timeout per request in seconds", required=True)
    retries = OptInteger(1, "Number of send attempts per query", required=True)
    output_file = OptString("", "Optional JSON output file", required=False)
    store_results = OptBool(True, "Store results in session.data['obd2_results']", required=False)

    def check(self):
        """Validate CANBUS session and python-can availability."""
        try:
            import can  # noqa: F401
        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
                print_error(f"Session is not a CANBUS session (type: {session.session_type})")
                return False
            print_error("Session not found")
            return False

        print_warning("Session manager not available - assuming valid session")
        return True

    def run(self):
        """Run the selected OBD-II query."""
        try:
            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
            request_id = self._parse_can_id(self.request_id)
            response_ids = self._parse_id_set(str(self.response_ids))
            timeout = max(0.05, float(self.timeout))
            retries = max(1, min(5, int(self.retries)))

            if request_id is None:
                print_error(f"Invalid request CAN ID: {self.request_id}")
                return False
            if not response_ids:
                print_error(f"Invalid response ID filter: {self.response_ids}")
                return False

            print_info("=" * 80)
            print_success("CANBUS OBD-II Query")
            print_info(f"Query       : {self.query}")
            print_info(f"Bus         : {interface}:{channel} @ {bitrate} bps")
            print_info(f"Request ID  : {self._fmt_can_id(request_id)}")
            print_info(f"Response IDs: {self._format_id_set(response_ids)}")
            print_info("=" * 80)

            bus = None
            own_bus = False
            try:
                bus = self._get_session_bus(data)
                if bus is None:
                    bus = self._open_bus(interface, channel, bitrate)
                    own_bus = True

                query_type = str(self.query)
                if query_type == "supported_pids":
                    results = self._query_supported_pids(bus, request_id, response_ids, timeout, retries)
                elif query_type == "vin":
                    results = self._query_vin(bus, request_id, response_ids, timeout, retries)
                elif query_type == "dtc":
                    results = self._query_dtcs(bus, request_id, response_ids, timeout, retries)
                else:
                    mode = self._parse_byte(self.mode)
                    pid = self._parse_byte(self.pid)
                    if mode is None or pid is None:
                        print_error("MODE and PID must be hex bytes for custom PID queries")
                        return False
                    results = self._query_pid(bus, request_id, response_ids, timeout, retries, mode, pid)
            finally:
                if own_bus and bus is not None:
                    try:
                        bus.shutdown()
                    except Exception:
                        pass

            report = {
                "query": str(self.query),
                "request_id": request_id,
                "request_id_hex": self._fmt_can_id(request_id),
                "response_ids": sorted(response_ids),
                "interface": interface,
                "channel": channel,
                "bitrate": bitrate,
                "queried_at": datetime.now().isoformat(),
                "results": results,
            }

            self._display_results(report)

            if self.output_file:
                with open(str(self.output_file), "w") as f:
                    json.dump(report, f, indent=2, default=str)
                print_success(f"\nOBD-II results saved to: {self.output_file}")

            if bool(self.store_results):
                try:
                    session.data.setdefault("obd2_results", []).append(report)
                    if hasattr(self.framework.session_manager, "update_session_data"):
                        self.framework.session_manager.update_session_data(
                            session_id_value,
                            {"obd2_results": session.data.get("obd2_results", [])},
                        )
                except Exception:
                    pass

            print_info("=" * 80)
            print_success("OBD-II query completed")
            return True

        except Exception as e:
            print_error(f"Error running OBD-II query: {e}")
            import traceback
            traceback.print_exc()
            return False

    def _query_supported_pids(self, bus, request_id, response_ids, timeout, retries):
        groups = [0x00, 0x20, 0x40, 0x60, 0x80, 0xA0]
        all_supported = []
        raw = []
        for base_pid in groups:
            responses = self._send_obd_request(bus, request_id, response_ids, 0x01, base_pid, timeout, retries)
            if not responses:
                if base_pid == 0x00:
                    break
                continue
            raw.extend(responses)
            group_supported = []
            for response in responses:
                data = response["payload"]
                if len(data) >= 6 and data[0] == 0x41 and data[1] == base_pid:
                    group_supported.extend(self._decode_supported_bitmap(base_pid, data[2:6]))
            for pid in group_supported:
                if pid not in all_supported:
                    all_supported.append(pid)
            if base_pid != 0x00 and not group_supported:
                break
            if base_pid + 0x20 not in group_supported:
                break
        return {
            "supported_pids": [{"pid": pid, "pid_hex": f"0x{pid:02X}", "name": PID_NAMES.get(pid, "Unknown")} for pid in all_supported],
            "raw_responses": raw,
        }

    def _query_vin(self, bus, request_id, response_ids, timeout, retries):
        responses = self._send_obd_request(bus, request_id, response_ids, 0x09, 0x02, timeout, retries)
        vins = []
        for response in responses:
            data = response["payload"]
            if len(data) >= 3 and data[0] == 0x49 and data[1] == 0x02:
                vin_bytes = bytes(b for b in data[3:] if 32 <= b <= 126)
                vin = vin_bytes.decode("ascii", errors="ignore").strip()
                if vin:
                    vins.append({"response_id": response["response_id"], "response_id_hex": response["response_id_hex"], "vin": vin})
        return {"vins": vins, "raw_responses": responses}

    def _query_dtcs(self, bus, request_id, response_ids, timeout, retries):
        responses = self._send_obd_request(bus, request_id, response_ids, 0x03, None, timeout, retries)
        dtcs = []
        for response in responses:
            data = response["payload"]
            if not data or data[0] != 0x43:
                continue
            for i in range(1, len(data) - 1, 2):
                if data[i] == 0 and data[i + 1] == 0:
                    continue
                dtcs.append({
                    "response_id": response["response_id"],
                    "response_id_hex": response["response_id_hex"],
                    "dtc": self._decode_dtc(data[i], data[i + 1]),
                })
        return {"dtcs": dtcs, "raw_responses": responses}

    def _query_pid(self, bus, request_id, response_ids, timeout, retries, mode, pid):
        responses = self._send_obd_request(bus, request_id, response_ids, mode, pid, timeout, retries)
        decoded = []
        for response in responses:
            payload = response["payload"]
            if len(payload) >= 2 and payload[0] == (mode + 0x40) and payload[1] == pid:
                decoded.append({
                    "response_id": response["response_id"],
                    "response_id_hex": response["response_id_hex"],
                    "mode": f"0x{mode:02X}",
                    "pid": f"0x{pid:02X}",
                    "name": PID_NAMES.get(pid, "Custom PID"),
                    "data": bytes(payload[2:]).hex().upper(),
                    "interpreted": self._interpret_pid(pid, payload[2:]),
                })
        return {"pid_values": decoded, "raw_responses": responses}

    def _send_obd_request(self, bus, request_id, response_ids, mode, pid, timeout, retries):
        import can

        request_data = [mode] if pid is None else [mode, pid]
        frame = bytes([len(request_data)] + request_data + [0x00] * (7 - len(request_data)))
        responses = []
        seen = set()

        for attempt in range(retries):
            bus.send(can.Message(arbitration_id=request_id, data=frame, is_extended_id=False, is_remote_frame=False))
            deadline = time.time() + timeout
            while time.time() < deadline:
                msg = bus.recv(timeout=0.05)
                if not msg or msg.arbitration_id not in response_ids:
                    continue
                payload = self._decode_isotp_response(bus, msg, request_id, response_ids, deadline)
                if not payload:
                    continue
                key = (msg.arbitration_id, bytes(payload))
                if key in seen:
                    continue
                seen.add(key)
                responses.append({
                    "response_id": msg.arbitration_id,
                    "response_id_hex": self._fmt_can_id(msg.arbitration_id),
                    "payload": list(payload),
                    "payload_hex": bytes(payload).hex().upper(),
                    "attempt": attempt + 1,
                })
        return responses

    def _decode_isotp_response(self, bus, first_msg, request_id, response_ids, deadline):
        import can

        data = bytes(first_msg.data)
        if not data:
            return b""
        pci = data[0]
        frame_type = pci >> 4

        if frame_type == 0x0:
            length = pci & 0x0F
            return data[1:1 + length]

        if frame_type != 0x1:
            return b""

        total_len = ((pci & 0x0F) << 8) | data[1]
        payload = bytearray(data[2:])

        # Flow Control: Continue To Send, block size 0, separation time 0.
        try:
            bus.send(can.Message(
                arbitration_id=request_id,
                data=bytes([0x30, 0x00, 0x00, 0, 0, 0, 0, 0]),
                is_extended_id=False,
                is_remote_frame=False,
            ))
        except Exception as e:
            print_warning(f"Failed to send ISO-TP flow control: {e}")

        expected_sn = 1
        while len(payload) < total_len and time.time() < deadline:
            msg = bus.recv(timeout=0.05)
            if not msg or msg.arbitration_id not in response_ids:
                continue
            chunk = bytes(msg.data)
            if not chunk or (chunk[0] >> 4) != 0x2:
                continue
            sn = chunk[0] & 0x0F
            if sn != expected_sn:
                print_warning(f"Unexpected ISO-TP sequence number: got {sn}, expected {expected_sn}")
            payload.extend(chunk[1:])
            expected_sn = (expected_sn + 1) & 0x0F

        return bytes(payload[:total_len])

    def _decode_supported_bitmap(self, base_pid, bitmap):
        supported = []
        bits = int.from_bytes(bytes(bitmap), "big")
        for offset in range(1, 33):
            if bits & (1 << (32 - offset)):
                supported.append(base_pid + offset)
        return supported

    def _decode_dtc(self, high, low):
        systems = ["P", "C", "B", "U"]
        system = systems[(high & 0xC0) >> 6]
        digit1 = (high & 0x30) >> 4
        digit2 = high & 0x0F
        digit3 = (low & 0xF0) >> 4
        digit4 = low & 0x0F
        return f"{system}{digit1}{digit2:X}{digit3:X}{digit4:X}"

    def _interpret_pid(self, pid, data):
        values = list(data)
        if pid == 0x04 and len(values) >= 1:
            return {"engine_load_percent": round(values[0] * 100 / 255, 2)}
        if pid == 0x05 and len(values) >= 1:
            return {"coolant_temp_c": values[0] - 40}
        if pid == 0x0C and len(values) >= 2:
            return {"rpm": ((values[0] * 256) + values[1]) / 4}
        if pid == 0x0D and len(values) >= 1:
            return {"speed_kmh": values[0]}
        if pid == 0x0F and len(values) >= 1:
            return {"intake_air_temp_c": values[0] - 40}
        if pid == 0x10 and len(values) >= 2:
            return {"maf_gps": ((values[0] * 256) + values[1]) / 100}
        if pid == 0x11 and len(values) >= 1:
            return {"throttle_percent": round(values[0] * 100 / 255, 2)}
        return None

    def _display_results(self, report):
        results = report["results"]
        query = report["query"]
        print_info("")
        if query == "supported_pids":
            pids = results.get("supported_pids", [])
            print_status(f"Supported PIDs ({len(pids)})")
            for item in pids:
                print_info(f"  {item['pid_hex']:<6} {item['name']}")
        elif query == "vin":
            vins = results.get("vins", [])
            print_status("VIN")
            if not vins:
                print_warning("  No VIN response decoded")
            for item in vins:
                print_success(f"  {item['response_id_hex']}: {item['vin']}")
        elif query == "dtc":
            dtcs = results.get("dtcs", [])
            print_status(f"Diagnostic Trouble Codes ({len(dtcs)})")
            if not dtcs:
                print_success("  No DTCs returned")
            for item in dtcs:
                print_info(f"  {item['response_id_hex']}: {item['dtc']}")
        else:
            values = results.get("pid_values", [])
            print_status("PID response")
            if not values:
                print_warning("  No matching PID response decoded")
            for item in values:
                print_info(f"  {item['response_id_hex']} {item['mode']}:{item['pid']} {item['data']}")
                if item.get("interpreted") is not None:
                    print_success(f"    {item['interpreted']}")

        raw = results.get("raw_responses", [])
        print_info("")
        print_status(f"Raw responses ({len(raw)})")
        for response in raw[:20]:
            print_info(f"  {response['response_id_hex']}: {response['payload_hex']}")
        if len(raw) > 20:
            print_info(f"  ... {len(raw) - 20} more")

    def _get_session_bus(self, data):
        for key in ("bus", "can_bus", "connection"):
            bus = data.get(key)
            if bus is not None and hasattr(bus, "send") and hasattr(bus, "recv"):
                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_byte(self, value):
        try:
            text = str(value).strip()
            parsed = int(text[2:], 16) if text.lower().startswith("0x") else int(text, 16)
            if 0 <= parsed <= 0xFF:
                return parsed
        except Exception:
            pass
        return None

    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 _parse_id_set(self, value):
        ids = set()
        for part in str(value or "").split(","):
            part = part.strip()
            if not part:
                continue
            if "-" in part:
                left, right = part.split("-", 1)
                start = self._parse_can_id(left)
                end = self._parse_can_id(right)
                if start is None or end is None or start > end or end - start > 4096:
                    continue
                ids.update(range(start, end + 1))
            else:
                can_id = self._parse_can_id(part)
                if can_id is not None:
                    ids.add(can_id)
        return ids

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

    def _format_id_set(self, ids):
        ordered = sorted(ids)
        if len(ordered) <= 8:
            return ", ".join(self._fmt_can_id(i) for i in ordered)
        return f"{self._fmt_can_id(ordered[0])}-{self._fmt_can_id(ordered[-1])} ({len(ordered)} IDs)"