big progress

flashloader/image-loader.py

@@ -5,6 +5,7 @@ import struct
 import logging
 import argparse
 import time
+import enum
 from tmtccmd.com.serial_base import SerialCfg
 from tmtccmd.com.serial_cobs import SerialCobsComIF
 from tmtccmd.com.ser_utils import prompt_com_port
@@ -34,9 +35,17 @@ APP_IMG_SZ = 0x1E000
 CHUNK_SIZE = 896
 
 MEMORY_SERVICE = 6
+ACTION_SERVICE = 8
+
 RAW_MEMORY_WRITE_SUBSERVICE = 2
 BOOT_NVM_MEMORY_ID = 1
 
+
+class ActionId(enum.IntEnum):
+    CORRUPT_APP_A = 128
+    CORRUPT_APP_B = 129
+
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -57,11 +66,12 @@ def main() -> int:
         prog="image-loader", description="Python VA416XX Image Loader Application"
     )
     parser.add_argument("-p", "--ping", action="store_true", help="Send ping command")
+    parser.add_argument("-c", "--corrupt", action="store_true", help="Corrupt a target")
     parser.add_argument(
-        "-f",
-        "--flash",
+        "-t",
+        "--target",
         choices=["bl", "a", "b"],
-        help="Flash target (Bootloader or slot A or B)",
+        help="Target (Bootloader or slot A or B)",
     )
     parser.add_argument(
         "path", nargs="?", default=None, help="Path to the App to flash"
@@ -84,19 +94,40 @@ def main() -> int:
     com_if = SerialCobsComIF(serial_cfg)
     com_if.open()
     file_path = None
-    if args.flash:
-        if not args.path:
-            _LOGGER.error("App Path needs to be specified for the flash process")
-            return -1
-        file_path = Path(args.path)
-        if not file_path.exists():
-            _LOGGER.error("File does not exist")
-            return -1
-    ping_tc = PusTc(apid=0x00, service=PusService.S17_TEST, subservice=1)
+    if args.target:
+        if not args.corrupt:
+            if not args.path:
+                _LOGGER.error("App Path needs to be specified for the flash process")
+                return -1
+            file_path = Path(args.path)
+            if not file_path.exists():
+                _LOGGER.error("File does not exist")
+                return -1
     if args.ping:
         _LOGGER.info("Sending ping command")
+        ping_tc = PusTc(apid=0x00, service=PusService.S17_TEST, subservice=1)
         com_if.send(ping_tc.pack())
-    if args.flash:
+    if args.corrupt:
+        if not args.target:
+            _LOGGER.error("target for corruption command required")
+            return -1
+        if args.target == "bl":
+            _LOGGER.error("can not corrupt bootloader")
+        if args.target == "a":
+            packet = PusTc(
+                apid=0,
+                service=ACTION_SERVICE,
+                subservice=ActionId.CORRUPT_APP_A,
+            )
+            com_if.send(packet.pack())
+        if args.target == "b":
+            packet = PusTc(
+                apid=0,
+                service=ACTION_SERVICE,
+                subservice=ActionId.CORRUPT_APP_B,
+            )
+            com_if.send(packet.pack())
+    else:
         assert file_path is not None
         loadable_segments = []
         _LOGGER.info("Parsing ELF file for loadable sections")
@@ -110,19 +141,19 @@ def main() -> int:
                 # Basic validity checks of the base addresses.
                 if idx == 0:
                     if (
-                        args.flash == "bl"
+                        args.target == "bl"
                         and segment.header.p_paddr != BOOTLOADER_START_ADDR
                     ):
                         raise ValueError(
                             f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
                             f"bootloader, expected {BOOTLOADER_START_ADDR}"
                         )
-                    if args.flash == "a" and segment.header.p_paddr != APP_A_START_ADDR:
+                    if args.target == "a" and segment.header.p_paddr != APP_A_START_ADDR:
                         raise ValueError(
                             f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
                             f"App A, expected {APP_A_START_ADDR}"
                         )
-                    if args.flash == "b" and segment.header.p_paddr != APP_B_START_ADDR:
+                    if args.target == "b" and segment.header.p_paddr != APP_B_START_ADDR:
                         raise ValueError(
                             f"detected possibly invalid start address {segment.header.p_paddr:#08x} for "
                             f"App B, expected {APP_B_START_ADDR}"
@@ -150,11 +181,11 @@ def main() -> int:
                 )
                 total_size += segment.header.p_filesz
             context_str = None
-            if args.flash == "bl":
+            if args.target == "bl":
                 context_str = "Bootloader"
-            elif args.flash == "a":
+            elif args.target == "a":
                 context_str = "App Slot A"
-            elif args.flash == "b":
+            elif args.target == "b":
                 context_str = "App Slot B"
             _LOGGER.info(
                 f"Flashing {context_str} with image {file_path} (size {total_size})"
@@ -181,7 +212,7 @@ def main() -> int:
                     current_addr += next_chunk_size
                     pos_in_segment += next_chunk_size
                     time.sleep(0.2)
-            if args.flash == "bl":
+            if args.target == "bl":
                 _LOGGER.info("Blanking the bootloader checksum")
                 # Blank the checksum. For the bootloader, the bootloader will calculate the
                 # checksum itself on the initial run.
@@ -192,10 +223,10 @@ def main() -> int:
             else:
                 crc_addr = None
                 size_addr = None
-                if args.flash == "a":
+                if args.target == "a":
                     crc_addr = APP_A_CRC_ADDR
                     size_addr = APP_A_SIZE_ADDR
-                elif args.flash == "b":
+                elif args.target == "b":
                     crc_addr = APP_B_CRC_ADDR
                     size_addr = APP_B_SIZE_ADDR
                 assert crc_addr is not None

flashloader/src/main.rs

@@ -33,6 +33,10 @@ const COBS_RX_DEBUGGING: bool = false;
 
 const BOOT_NVM_MEMORY_ID: u8 = 1;
 
+pub enum ActionId {
+    CorruptImageA = 128,
+    CorruptImageB = 129,
+}
 pub trait WdtInterface {
     fn feed(&self);
 }
@@ -49,6 +53,11 @@ impl WdtInterface for OptWdt {
 
 static CLOCKS: OnceCell<Clocks> = OnceCell::new();
 
+pub const APP_A_START_ADDR: u32 = 0x4000;
+pub const APP_A_END_ADDR: u32 = 0x22000;
+pub const APP_B_START_ADDR: u32 = 0x22000;
+pub const APP_B_END_ADDR: u32 = 0x40000;
+
 #[rtic::app(device = pac, dispatchers = [U1, U2, U3])]
 mod app {
     use super::*;
@@ -303,6 +312,31 @@ mod app {
             cx.shared.decode_buffer_busy.lock(|busy| *busy = false);
             match PusTcReader::new(cx.local.read_buf) {
                 Ok((pus_tc, _)) => {
+                    if pus_tc.service() == PusServiceId::Action as u8 {
+                        let mut corrupt_image = |base_addr: u32| {
+                            // Safety: We only use this for NVM handling and we only do NVM
+                            // handling here.
+                            let mut sys_cfg = unsafe { pac::Sysconfig::steal() };
+                            let nvm = Nvm::new(
+                                &mut sys_cfg,
+                                cx.local.rom_spi.take().unwrap(),
+                                CLOCKS.get().as_ref().unwrap(),
+                            );
+                            let mut buf = [0u8; 4];
+                            nvm.read_data(base_addr + 32, &mut buf);
+                            buf[0] += 1;
+                            nvm.write_data(base_addr + 32, &buf);
+                            *cx.local.rom_spi = Some(nvm.release(&mut sys_cfg));
+                        };
+                        if pus_tc.subservice() == ActionId::CorruptImageA as u8 {
+                            rprintln!("corrupting App Image A");
+                            corrupt_image(APP_A_START_ADDR);
+                        }
+                        if pus_tc.subservice() == ActionId::CorruptImageB as u8 {
+                            rprintln!("corrupting App Image B");
+                            corrupt_image(APP_B_START_ADDR);
+                        }
+                    }
                     if pus_tc.service() == PusServiceId::Test as u8 && pus_tc.subservice() == 1 {
                         log::info!(target: "TC Handler", "received ping TC");
                     } else if pus_tc.service() == PusServiceId::MemoryManagement as u8 {