diff --git a/docs/uds/scan_modes.md b/docs/uds/scan_modes.md index eba1bf3a8..cbfd5473b 100644 --- a/docs/uds/scan_modes.md +++ b/docs/uds/scan_modes.md @@ -122,14 +122,12 @@ The scan is finished, if no new session transition is found. The service scan operates at the UDS protocol level. UDS provides several endpoints called *services*. -Each service has an identifier and a specific list of arguments. -The scanning procedure is requesting all possible services with a fixed payload. -The payload does not have to be specific; it can be empty, it can be all zeroes, … -A few ECUs behave unstable when they receive an invalid payload. +Each service has an identifier and a specific list of arguments or sub-functions. In order to identify available services, a reverse matching is applied. According to the UDS standard, ECUs reply with the error codes `serviceNotSupported` or `serviceNotSupportedInActiveSession` when an unimplemented service is requested. Therefore, each service which responds with a different error code is considered available. +To address the different services and their varying length of arguments and sub-functions the scanner automatically appends `\x00` bytes if the received response was `incorrectMessageLengthOrInvalidFormat`. ## Identifier Scan diff --git a/src/gallia/commands/scan/uds/services.py b/src/gallia/commands/scan/uds/services.py index d91019883..b015284fd 100644 --- a/src/gallia/commands/scan/uds/services.py +++ b/src/gallia/commands/scan/uds/services.py @@ -5,19 +5,18 @@ import asyncio import reprlib from argparse import BooleanOptionalAction, Namespace -from binascii import unhexlify from typing import Any from gallia.command import UDSScanner from gallia.services.uds import ( NegativeResponse, + UDSErrorCodes, UDSIsoServices, UDSRequestConfig, UDSResponse, ) -from gallia.services.uds.core.exception import UDSException +from gallia.services.uds.core.exception import MalformedResponse, UDSException from gallia.services.uds.core.utils import g_repr -from gallia.services.uds.helpers import suggests_service_not_supported from gallia.utils import ParseSkips, auto_int @@ -48,7 +47,7 @@ def configure_parser(self) -> None: "--scan-response-ids", default=False, action=BooleanOptionalAction, - help="Do not scan reply flag in SID", + help="Include IDs in scan with reply flag set", ) self.parser.add_argument( "--auto-reset", @@ -56,12 +55,6 @@ def configure_parser(self) -> None: default=False, help="Reset ECU with UDS ECU Reset before every request", ) - self.parser.add_argument( - "--payload", - default=None, - type=unhexlify, - help="Payload which will be appended for each request as hex string", - ) self.parser.add_argument( "--skip", nargs="+", @@ -113,7 +106,7 @@ async def main(self, args: Namespace) -> None: resp: UDSResponse = await self.ecu.set_session( session, UDSRequestConfig(tags=["preparation"]) ) - except (UDSException, RuntimeError) as e: + except (UDSException, RuntimeError) as e: # FIXME why catch RuntimeError? self.logger.warning( f"session change: {g_repr(session)} reason: {g_repr(e)}" ) @@ -143,28 +136,42 @@ async def main(self, args: Namespace) -> None: ) break - pdu = bytes([sid]) + args.payload if args.payload else bytes([sid]) - - try: - resp = await self.ecu.send_raw( - pdu, config=UDSRequestConfig(tags=["ANALYZE"]) - ) - except asyncio.TimeoutError: - self.logger.info(f"{g_repr(sid)}: timeout") - continue - except Exception as e: - self.logger.info(f"{g_repr(sid)}: {e!r} occurred") - await self.ecu.reconnect() - continue + for length_payload in [1, 2, 3, 5]: + pdu = bytes([sid]) + bytes(length_payload) + try: + resp = await self.ecu.send_raw( + pdu, config=UDSRequestConfig(tags=["ANALYZE"]) + ) + except asyncio.TimeoutError: + self.logger.info(f"{g_repr(sid)}: timeout") + continue + except MalformedResponse as e: + self.logger.warning( + f"{g_repr(sid)}: {e!r} occurred, this needs to be investigated!" + ) + continue + except Exception as e: + self.logger.info(f"{g_repr(sid)}: {e!r} occurred") + await self.ecu.reconnect() + continue + + if isinstance(resp, NegativeResponse) and resp.response_code in [ + UDSErrorCodes.serviceNotSupported, + UDSErrorCodes.serviceNotSupportedInActiveSession, + ]: + self.logger.info(f"{g_repr(sid)}: not supported [{resp}]") + break - if suggests_service_not_supported(resp): - self.logger.info(f"{g_repr(sid)}: not supported [{resp}]") - continue + if isinstance(resp, NegativeResponse) and resp.response_code in [ + UDSErrorCodes.incorrectMessageLengthOrInvalidFormat, + ]: + continue - self.logger.result( - f"{g_repr(sid)}: available in session {g_repr(session)}: {resp}" - ) - found[session][sid] = resp + self.logger.result( + f"{g_repr(sid)}: available in session {g_repr(session)}: {resp}" + ) + found[session][sid] = resp + break await self.ecu.leave_session(session)