1"""
  2A temporary file for quick development. Some of these should be broken out into test cases later on.
  3
  4# (c) 2020-2024 TurnAround Factor, Inc.
  5#
  6# CUI DISTRIBUTION CONTROL
  7# Controlled by: DLA J68 R&D SBIP
  8# CUI Category: Small Business Research and Technology
  9# Distribution/Dissemination Controls: PROTECTED BY SBIR DATA RIGHTS
 10# POC: GOV SBIP Program Manager Denise Price, 571-767-0111
 11# Distribution authorized to U.S. Government Agencies only, to protect information not owned by the
 12# U.S. Government and protected by a contractor’s ‘limited rights’ statement, or received with the understanding that
 13# it is not routinely transmitted outside the U.S. Government (determination made September 14, 2024). Other requests
 14# for this document shall be referred to ACOR DLA Logistics Operations (J-68), 8725 John J. Kingman Rd., Suite 4317,
 15# Fort Belvoir, VA 22060-6221
 16#
 17# SBIR DATA RIGHTS
 18# Contract No.:SP4701-23-C-0083
 19# Contractor Name: TurnAround Factor, Inc.
 20# Contractor Address: 10365 Wood Park Ct. Suite 313 / Ashland, VA 23005
 21# Expiration of SBIR Data Rights Period: September 24, 2029
 22# The Government's rights to use, modify, reproduce, release, perform, display, or disclose technical data or computer
 23# software marked with this legend are restricted during the period shown as provided in paragraph (b)(4) of the Rights
 24# in Noncommercial Technical Data and Computer Software--Small Business Innovative Research (SBIR) Program clause
 25# contained in the above identified contract. No restrictions apply after the expiration date shown above. Any
 26# reproduction of technical data, computer software, or portions thereof marked with this legend must also reproduce
 27# the markings.
 28"""
 29
 30import itertools
 31import time
 32from contextlib import suppress
 33from enum import IntEnum
 34from typing import Literal
 35
 36import pytest
 37
 38from hitl_tester.modules.cyber_6t import spn_types
 39from hitl_tester.modules.cyber_6t.canbus import CANBus, CANFrame
 40from hitl_tester.modules.cyber_6t.j1939da import PGN
 41
 42from hitl_tester.modules import properties
 43from hitl_tester.modules.logger import logger
 44
 45BATTERY_CHANNEL = "can0"
 46"""Channel identification. Expected type is backend dependent."""
 47
 48HARD_RESET = False
 49"""Whether to use a soft or hard reset before each test."""
 50
 51properties.apply()  # Allow modifying the above globals
 52
 53
 54class ManufacturerID(IntEnum):
 55    """Enum to hold Manufacturer IDs."""
 56
 57    SAFT = 269  # NOTE: unused
 58    BRENTRONICS = 822
 59
 60
 61BATTERY_ADDRESS = 0
 62
 63
 64class BatteryInfo:
 65    """class to store Battery Information storage"""
 66
 67    def __init__(self):
 68        self.data = {}
 69
 70    def get_item(self, key):
 71        """Gets a specific item from data"""
 72        return self.data[key]
 73
 74    def set_item(self, key, value):
 75        """Sets a specific item from data"""
 76        self.data[key] = value
 77
 78
 79BATTERY_INFO = BatteryInfo()
 80
 81
 82def cmp(
 83    a: float | str,
 84    sign: Literal["<", "<=", ">", ">=", "=="],
 85    b: float | str,
 86    unit_a: str = "",
 87    unit_b: str = "",
 88    form: str = "",
 89) -> str:
 90    """Generate a formatted string based on a comparison."""
 91    if not unit_b:
 92        unit_b = unit_a
 93
 94    if isinstance(a, str) or isinstance(b, str):
 95        return f"{a:{form}}{unit_a} {('≠', '=')[a == b]} {b:{form}}{unit_b}"
 96
 97    sign_str = {
 98        "<": ("≮", "<")[a < b],
 99        "<=": ("≰", "≤")[a <= b],
100        ">": ("≯", ">")[a > b],
101        ">=": ("≱", "≥")[a >= b],
102        "==": ("≠", "=")[a == b],
103    }
104
105    return f"{a:{form}}{unit_a} {sign_str[sign]} {b:{form}}{unit_b}"
106
107
108@pytest.fixture(scope="class", autouse=True)
109def reset_test_environment():
110    """Before each test class, reset the 6T."""
111
112    if BATTERY_ADDRESS:
113        power_cycle_frame = CANFrame(
114            destination_address=BATTERY_ADDRESS,
115            pgn=PGN["PropA"],
116            data=[0, 0, 1, 1, 1, not HARD_RESET, 1, 3, 0, -1],
117        )
118        maintenance_mode_frame = CANFrame(
119            destination_address=BATTERY_ADDRESS,
120            pgn=PGN["PropA"],
121            data=[0, 0, 1, 1, 1, 3, 1, 3, 0, -1],
122        )
123        factory_reset_frame = CANFrame(
124            destination_address=BATTERY_ADDRESS,
125            pgn=PGN["PropA"],
126            data=[1, 0, 0, -1, 3, 0xF, 3, 0, 0x1F, -1],
127        )
128        with CANBus(BATTERY_CHANNEL) as bus:
129            logger.write_info_to_report("Power-Cycling 6T")
130            bus.process_call(power_cycle_frame)
131            time.sleep(10)
132            logger.write_info_to_report("Entering maintenance mode")
133            bus.process_call(maintenance_mode_frame)
134            logger.write_info_to_report("Factory resetting 6T")
135            bus.process_call(factory_reset_frame)
136            time.sleep(10)
137    else:  # Battery has not yet been found
138        return
139
140
141class TestLocate6T:
142    """Scan for 6T battery."""
143
144    MAX_ATTEMPTS = 12  # 2 minutes worth of attempts
145
146    def __init__(self):
147        self.id = 0
148        self.manufacturer_code = 0
149        self.manufacturer_name = ""
150
151    def test_profile(self) -> None:
152        """
153        | Description          | Scan the bus for devices                                         |
154        | :------------------- | :--------------------------------------------------------------- |
155        | Instructions         | 1. Request the names of everyone on the bus                 </br>\
156                                 2. Use the response data for all communication                   |
157        | Estimated Duration   | 1 second                                                         |
158        """
159        global BATTERY_ADDRESS
160
161        name_request = CANFrame(pgn=PGN["Request"], data=[PGN["Address Claimed"].id])
162        with CANBus(BATTERY_CHANNEL) as bus:
163            for attempt in range(self.MAX_ATTEMPTS):
164                with suppress(IndexError):
165                    if name_frame := bus.process_call(name_request):
166                        # Save responses to profile
167                        self.id = int(name_frame.data[0])
168                        self.manufacturer_code = name_frame.data[1]
169                        BATTERY_INFO.set_item("manufacturer_code", name_frame.data[1])
170                        if (mfg_name := spn_types.manufacturer_id(name_frame.data[1])) == "Reserved":
171                            mfg_name = f"Unknown ({name_frame.data[1]})"
172                        self.manufacturer_name = mfg_name
173                        BATTERY_ADDRESS = name_frame.source_address
174                        break
175                logger.write_warning_to_html_report(f"Failed to locate 6T. Retrying, attempt {attempt + 1}")
176                time.sleep(10)
177            else:
178                message = "Could not locate 6T"
179                logger.write_warning_to_html_report(message)
180                pytest.exit(message)
181
182        # Log results
183        printable_id = "".join(chr(i) if chr(i).isprintable() else "." for i in self.id.to_bytes(3, byteorder="big"))
184        logger.write_result_to_html_report(
185            f"Found {self.manufacturer_name} 6T at address {BATTERY_ADDRESS} "
186            f"(ID: {self.id:06X}, ASCII ID: {printable_id})"
187        )
188
189
190class TestScanForProprietary:
191    """Scan all 512 proprietary commands for any responses."""
192
193    def test_scan_for_proprietary(self):
194        """
195        | Description          | Scan all 512 proprietary commands for any responses              |
196        | :------------------- | :--------------------------------------------------------------- |
197        | Instructions         | 1. Scan addresses $EFxx and $FFxx                           </br>\
198                                 2. Log any successful responses                             </br>\
199                                 3. Check if responded commands are not documented                |
200        | Estimated Duration   | 85 seconds                                                       |
201        """
202        TestLocate6T().test_profile()
203        proprietary_request = CANFrame(destination_address=BATTERY_ADDRESS, pgn=PGN["RQST"], data=[0])
204        ack_responses: dict[int, CANFrame] = {}  # type: ignore # address: response
205        with CANBus(BATTERY_CHANNEL) as bus:
206            for address in itertools.chain(range(0xEF00, 0xF000), range(0xFF00, 0x10000)):
207                proprietary_request.data = [address]
208                if response_frame := bus.process_call(proprietary_request):  # Log if response is not NACK
209                    if not (response_frame.pgn.id == PGN["Acknowledgement"].id and response_frame.data[0] == 1):
210                        ack_responses[address] = response_frame
211                        logger.write_info_to_report(f"Got response at address {address:04X}:\n{response_frame}")
212
213            milprf_commands = [0xFF00, 0xFF01, 0xFF02, 0xFF03, 0xFF04, 0xFF05, 0xFF06, 0xFF07, 0xFF08]
214            j1939_transmit_commands = [0xE800, 0xEE00, 0xFEE6, 0xFCB6, 0xFECA, 0xFECB, 0xFE50, 0xFDC5, 0xFEDA, 0xD800]
215
216            if BATTERY_INFO.get_item("manufacturer_code") == ManufacturerID.SAFT:
217                saft_commands = [
218                    0xFFD2,
219                    0xFFD4,
220                    0xFFD5,
221                    0xFFD6,
222                    0xFFD7,
223                    0xFFD8,
224                    0xFFDC,
225                    0xFFDD,
226                    0xFFDE,
227                    0xFFE0,
228                    0xFFE1,
229                    0xFFE1,
230                    0xFFE4,
231                    0xFFE9,
232                    0x1EF00,
233                ]
234                documented_commands = list(dict.fromkeys(milprf_commands + j1939_transmit_commands + saft_commands))
235            else:
236                documented_commands = milprf_commands + j1939_transmit_commands
237
238            if unknown_proprietary_commands := set(ack_responses) - set(documented_commands):
239                message = (
240                    f"Unknown Proprietary Command{'s' if len(unknown_proprietary_commands) > 1 else ''}: "
241                    f"{', '.join(map(str, unknown_proprietary_commands))}"
242                )
243                logger.write_warning_to_html_report(message)
244                message = (
245                    f"{len(unknown_proprietary_commands)} command"
246                    f"{'s' if len(unknown_proprietary_commands) > 1 else ''} missing known documentation"
247                )
248                logger.write_failure_to_html_report(message)
249                pytest.fail(message)
250
251            logger.write_result_to_html_report("All responded commands are known and documented")
252
253
254def request_data(end_of_message: bool = True) -> bool:
255    """Send a RTS."""
256    cts_pgn = PGN["ECUID"].id
257    end_acknowledge_frame = CANFrame(  # Memory access only works in maintenance mode
258        destination_address=BATTERY_ADDRESS,
259        pgn=PGN["TP.CM"],
260        data=[17, 0x0203, 0xFF, 0xFF, cts_pgn],  # End of Message Acknowledge, bytes, packets, reserved, PGN
261    )
262    cts_r = CANFrame(destination_address=BATTERY_ADDRESS, pgn=PGN["RQST"], data=[cts_pgn])
263
264    with CANBus(BATTERY_CHANNEL) as bus:
265        # Get request to send
266        logger.write_info_to_report("Sending RTS")
267        rts_frame = bus.process_call(cts_r)
268        logger.write_info_to_report(f"Got response:\n{rts_frame}")
269        logger.write_info_to_report("Not sending CTS. Purposefully waiting")
270
271        # Prep for End of Message
272        expected_bytes = int(rts_frame.data[1])
273        expected_packets = int(rts_frame.data[2])
274        rts_pgn_id = int(rts_frame.data[-1])
275
276        # Acknowledge packets
277        if end_of_message:
278            end_acknowledge_frame.data[1] = expected_bytes
279            end_acknowledge_frame.data[2] = expected_packets
280            end_acknowledge_frame.data[-1] = rts_pgn_id
281            op_frame = bus.process_call(end_acknowledge_frame)
282            logger.write_info_to_report(f"End of message response:\n{op_frame}")
283            return True
284
285        time.sleep(10)
286        if response_frame := bus.read_frame():
287            if response_frame.data[0] == 255:
288                if response_frame.data[1] == 3:
289                    logger.write_info_to_report(f"Got timeout response:\n{response_frame}")
290                else:
291                    logger.write_info_to_report(f"Received different connection abort message:\n{response_frame}")
292                return True
293
294            logger.write_info_to_report(f"May not have received a timeout response:\n{response_frame}")
295
296        logger.write_info_to_report("No response after waiting...")
297        return False
298
299
300def request_data_cts_test(size: int | None = None, sequence_id: int | None = None, end_of_message: bool = True) -> bool:
301    """Send a CTS."""
302    test_passed = True
303    cts_pgn = PGN["ECUID"].id
304    cts_frame = CANFrame(
305        destination_address=BATTERY_ADDRESS,
306        pgn=PGN[0xEC00],
307        data=[17, 0x0203, 0xFF, 0xFF, cts_pgn],  # Packets that can be sent, next packet#, reserved, PGN
308    )
309    end_acknowledge_frame = CANFrame(  # Memory access only works in maintenance mode
310        destination_address=BATTERY_ADDRESS,
311        pgn=PGN["TP.CM"],
312        data=[19, 0, 0, 0xFF, 0],  # End of Message Acknowledge, bytes, packets, reserved, PGN
313    )
314    cts_r = CANFrame(destination_address=BATTERY_ADDRESS, pgn=PGN["Request"], data=[cts_pgn])
315
316    with CANBus(BATTERY_CHANNEL) as bus:
317        # Get request to send
318        logger.write_info_to_report("Sending RTS")
319        rts_frame = bus.process_call(cts_r)
320        logger.write_info_to_report(f"Got response:\n{rts_frame}")
321
322        # Send malicious CTS
323        expected_bytes = int(rts_frame.data[1])
324        expected_packets = int(rts_frame.data[2])
325        rts_pgn_id = int(rts_frame.data[-1])
326        cts_frame.data[1] = expected_packets if size is None else size
327        cts_frame.data[2] = 1 if sequence_id is None else sequence_id
328        cts_frame.data[-1] = rts_pgn_id
329
330        logger.write_info_to_report("Sending CTS")
331        bus.send_frame(cts_frame)
332        packet_count = 0
333        while response_frame := bus.read_frame():
334
335            logger.write_info_to_report(f"Got response:\n{response_frame}")
336
337            if response_frame.pgn.id == PGN["TP.CM", [32]].id:
338                continue
339
340            packet_count += 1
341
342            if sequence_id is not None:
343                if expected_packets >= sequence_id > response_frame.data[0]:
344
345                    cmp_text = cmp(response_frame.data[0], ">=", sequence_id)
346
347                    logger.write_warning_to_html_report(
348                        f"Received sequence ID {response_frame.data[0]}, "
349                        f"which is before the requested sequence ID: {cmp_text}"
350                    )
351                    # Pass BT for now with warning, special case
352                    if BATTERY_INFO.get_item("manufacturer_code") == ManufacturerID.BRENTRONICS:
353                        continue
354                    test_passed = False
355
356        # Acknowledge packets
357        if end_of_message:
358            end_acknowledge_frame.data[1] = expected_bytes
359            end_acknowledge_frame.data[2] = expected_packets
360            end_acknowledge_frame.data[-1] = rts_pgn_id
361            op_frame = bus.process_call(end_acknowledge_frame)
362            logger.write_info_to_report(f"End of message response:\n{op_frame}")
363        else:
364            time.sleep(10)
365            if response_frame := bus.read_frame():
366                logger.write_info_to_report(f"Got possible timeout response:\n{response_frame}")
367            else:
368                logger.write_info_to_report("No response after waiting...")
369
370        if size is not None:
371            cmp_text = cmp(packet_count, "==", expected_packets)
372            if size == 0:
373                if packet_count != 0:
374                    logger.write_warning_to_html_report(
375                        f"Did not receive expected packets with invalid size: {cmp(packet_count, '==', 0)}"
376                    )
377                    test_passed = False
378                else:
379                    logger.write_result_to_html_report(f"Received expected packets: {cmp(packet_count, '==', 0)}")
380            elif size > expected_packets:
381                if packet_count != expected_packets:
382                    logger.write_warning_to_html_report(f"Did not receive expected packets: {cmp_text}")
383                    test_passed = False
384                else:
385                    logger.write_result_to_html_report(f"Received expected packets: {cmp_text}")
386            elif size != packet_count:
387                logger.write_warning_to_html_report(f"Did not receive expected packets: {cmp_text}")
388                test_passed = False
389            else:
390                logger.write_result_to_html_report(f"Received expected packets: {cmp(packet_count, '==', size)}")
391
392        if sequence_id is not None:
393            if sequence_id == 0 or sequence_id > expected_packets:
394                cmp_text = cmp(packet_count, "==", 0)
395                if packet_count != 0:
396                    logger.write_warning_to_html_report(
397                        f"Did not receive expected number of packets with invalid request: {cmp_text}"
398                    )
399                    test_passed = False
400                else:
401                    logger.write_result_to_html_report(
402                        f"Received expected packets with invalid sequence request: {cmp_text}"
403                    )
404            else:
405                overall_expected_packets = expected_packets - sequence_id + 1
406                cmp_text = cmp(packet_count, "==", overall_expected_packets)
407                if packet_count != abs(expected_packets - sequence_id + 1):
408                    logger.write_warning_to_html_report(f"Did not receive expected number of packets: {cmp_text}")
409                else:
410                    logger.write_result_to_html_report(f"Received expected number of packets: {cmp_text}")
411
412        if sequence_id is None and size is None:
413            if expected_packets == packet_count:
414                logger.write_result_to_html_report(
415                    f"Received expected number of packets: {cmp(packet_count, '==', expected_packets)}"
416                )
417            else:
418                logger.write_warning_to_html_report(
419                    f"Did not receive expected number of packets: {cmp(packet_count, '==', expected_packets)}"
420                )
421
422        if BATTERY_INFO.get_item("manufacturer_code") == ManufacturerID.BRENTRONICS and test_passed is False:
423            return True
424
425        return test_passed
426
427
428class TestMaliciousCTS:
429    """Send a CTS with bad sequence ID or size."""
430
431    def test_malicious_cts(self):
432        """
433        | Description          | Send a CTS with bad sequence ID or size                          |
434        | :------------------- | :--------------------------------------------------------------- |
435        | Instructions         | 1. Send a normal CTS                                        </br>\
436                                 2. Send a CTS with sequence ID of zero                      </br>\
437                                 3. Send a CTS with sequence ID too high                     </br>\
438                                 4. Send a CTS with zero size                                </br>\
439                                 5. Send a CTS with small size                               </br>\
440                                 6. Send a CTS with size too big                                  |
441        | Estimated Duration   | 85 seconds                                                       |
442        """
443        tests_failed = []
444        TestLocate6T().test_profile()
445
446        cts_tests = [
447            {"sequence_id": None, "size": None, "log_text": "Sending normal CTS"},
448            {"sequence_id": 0, "size": None, "log_text": "Sending CTS with sequence ID of zero"},
449            {"sequence_id": 3, "size": None, "log_text": "Sending CTS with sequence ID of three"},
450            {"sequence_id": 200, "size": None, "log_text": "Sending CTS with sequence ID too high"},
451            {"sequence_id": None, "size": 0, "log_text": "Sending CTS with zero size"},
452            {"sequence_id": None, "size": 1, "log_text": "Sending CTS with small size"},
453            {"sequence_id": None, "size": 200, "log_text": "Sending CTS with size too big"},
454        ]
455
456        for cts_test in cts_tests:
457            if cts_test["sequence_id"] is None:
458                sequence_id = None
459            else:
460                sequence_id = cts_test["sequence_id"]
461
462            if cts_test["size"] is None:
463                size = None
464            else:
465                size = cts_test["size"]
466
467            logger.write_result_to_html_report(cts_test["log_text"])
468            did_test_pass = request_data_cts_test(size, sequence_id)
469
470            if did_test_pass is False:
471                logger.write_failure_to_html_report(f"Test failed when {cts_test['log_text']}.")
472                tests_failed.append(cts_test["log_text"])
473
474        if len(tests_failed) > 0:
475            pytest.fail("One or more tests failed Malicious CTS vulnerability scan")
476        elif BATTERY_INFO.get_item("manufacturer_code") == ManufacturerID.BRENTRONICS:
477            logger.write_warning_to_html_report("See warnings for possible discrepancies with J1939 expectations")
478        else:
479            logger.write_result_to_html_report("Malicious CTS vulnerability scans met expectations")
480
481
482class TestConnectionExhaustion:
483    """Start a connection, but don't close it."""
484
485    def test_connection_exhaustion(self):
486        """
487        | Description          | Start a connection, but don't close it                           |
488        | :------------------- | :--------------------------------------------------------------- |
489        | Instructions         | 1. Open a connection                                        </br>\
490                                 2. Refuse to close it and check for timeouts                     |
491        | Estimated Duration   | 85 seconds                                                       |
492        """
493        TestLocate6T().test_profile()
494
495        logger.write_info_to_report("Establishing connection...")
496        test_passed = request_data(end_of_message=False)
497
498        if test_passed is False:
499            logger.write_failure_to_html_report("No Connection Abort message was received after waiting 10 seconds")
500            pytest.fail("No Connection Abort message was received after waiting 10 seconds")
501        else:
502            logger.write_result_to_html_report("Connection Abort message was received after waiting 10 seconds")