From b84d229326a1eab2aabe3bb4504c55526578e352 Mon Sep 17 00:00:00 2001 From: MCKero Date: Fri, 28 Aug 2026 16:25:37 +0100 Subject: [PATCH] Implement serial receive, unlocking the UV-K5 programming protocol Nothing could be sent to the firmware. Three separate pieces were missing. USART1 was a register stub with no chardev, so there was no source of incoming bytes. It now takes a chardev property, defaulting to serial0, so -serial works. The DMA model never serviced USART. App/driver/uart.c receives over a circular peripheral-to-memory channel and never reads DR; it finds new data with write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(DMA1, CHANNEL_2) so leaving CNDTR at its programmed value made the buffer look permanently empty however many bytes arrived. DMA now drains USART1's queue byte by byte, decrementing CNDTR and reloading it in circular mode. Channels also remember the length they were given, since CNDTR counts down and the offset into the buffer has to be derived from the difference. Servicing happens on a CNDTR read rather than from a timer: that read is precisely how the driver looks for data, so no polling is needed and no byte can be delivered before the guest asks. And transmit was invisible to the far end. DR writes went to stderr only, so a host tool would send a command, the firmware would answer, and the answer went nowhere the tool could see -- indistinguishable from being ignored. This cost a debugging round: the first test run reported "no reply at all" with 0 bytes of boot output, which looked like receive failing when the boot banner was in fact being written to stderr as always. DR now also writes the raw byte to the chardev when one is connected. SR reports RXNE when bytes are queued and DR consumes one, so a polling firmware would work too, even though this one uses DMA. tools/test_serial_rx.py speaks the real wire protocol -- AB CD framing, the fixed XOR obfuscation, CRC-16/XMODEM -- and checks two exchanges end to end: 0x0514 hello -> 0x0515 ack 0x051B EEPROM read -> 0x051C with the requested 8 bytes at 0x0E70 Both pass. CPS/CHIRP-style tools can now talk to the emulator. keypad_test.py, test_flash_persist.py, test_freq_entry.py and the 143 unit tests still pass. --- qemu/py32f071.c | 215 ++++++++++++++++++++++++++++++++++++++- tools/test_serial_rx.py | 220 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 430 insertions(+), 5 deletions(-) create mode 100755 tools/test_serial_rx.py diff --git a/qemu/py32f071.c b/qemu/py32f071.c index eac95fb..1bcfcd8 100644 --- a/qemu/py32f071.c +++ b/qemu/py32f071.c @@ -28,9 +28,12 @@ #include "hw/arm/armv7m.h" #include "hw/boards.h" #include "hw/qdev-properties.h" +/* Serial receive: USART1 takes a chardev so a host tool can drive the firmware. */ +#include "hw/qdev-properties-system.h" +#include "chardev/char-fe.h" +#include "sysemu/sysemu.h" #include "hw/sysbus.h" #include "exec/address-spaces.h" -#include "sysemu/sysemu.h" #include "qom/object.h" /* ---------------------------------------------------------------- memory map */ @@ -820,8 +823,19 @@ OBJECT_DECLARE_SIMPLE_TYPE(PY32DmaState, PY32_DMA) typedef struct { uint32_t ccr, cndtr, cpar, cmar; + /* + * The length the guest programmed, kept separately because cndtr counts down. + * Needed to derive how far into the buffer a transfer has got, and to reload + * the count in circular mode. + */ + uint32_t total; } PY32DmaChannel; +/* Defined further down; DMA drains its receive queue. */ +typedef struct PY32StubState PY32StubState; +static bool py32_stub_rx_empty(PY32StubState *s); +static bool py32_stub_rx_pop(PY32StubState *s, uint8_t *out); + struct PY32DmaState { SysBusDevice parent_obj; MemoryRegion iomem; @@ -836,6 +850,12 @@ struct PY32DmaState { /* Set by the SoC: lets the DMA clock bytes through an SPI controller. */ PY32SpiState *spi[2]; + /* + * Set by the SoC. USART1's receive queue is drained from here because the + * firmware's UART driver never reads DR -- it watches the DMA count instead. + */ + PY32StubState *usart1; + /* * The address space DMA transfers move bytes through. * @@ -975,6 +995,73 @@ static void py32_dma_kick(void *dma, PY32SpiState *spi) py32_dma_run_for_spi((PY32DmaState *)dma, spi); } +/* + * Move queued USART bytes into the guest buffer, one at a time, decrementing the + * channel's remaining count. + * + * The count is the whole point. App/driver/uart.c configures a circular + * peripheral-to-memory channel and never reads DR; it locates new data with + * + * write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(...) + * + * so a model that leaves CNDTR at its initial value reports an empty buffer + * forever, no matter how many bytes arrived. Serial receive was dead for exactly + * that reason, and with it the whole UV-K5 programming protocol. + * + * Circular mode reloads the count and wraps the address on completion rather than + * stopping, which is what makes the firmware's pointer arithmetic work across the + * end of the buffer. + */ +static void py32_dma_service_usart_rx(PY32DmaState *s) +{ + AddressSpace *as = s->as; + + if (!as || !s->usart1) { + return; + } + + for (int ch = 0; ch < PY32_DMA_CHANNELS; ch++) { + PY32DmaChannel *c = &s->ch[ch]; + + if (!(c->ccr & DMA_CCR_EN) || (c->ccr & DMA_CCR_DIR)) { + continue; /* disabled, or memory-to-peripheral */ + } + if ((c->cpar & ~0x3ffu) != PY32_USART1_BASE) { + continue; + } + if (c->total == 0) { + continue; /* never configured with a length */ + } + + while (!py32_stub_rx_empty(s->usart1)) { + uint8_t byte; + if (!py32_stub_rx_pop(s->usart1, &byte)) { + break; + } + + const uint32_t done = c->total - c->cndtr; + const uint32_t dest = c->cmar + ((c->ccr & DMA_CCR_MINC) ? done : 0); + address_space_write(as, dest, MEMTXATTRS_UNSPECIFIED, &byte, 1); + + if (c->cndtr > 0) { + c->cndtr--; + } + + if (c->cndtr == 0) { + if (c->ccr & DMA_CCR_CIRC) { + c->cndtr = c->total; /* wrap, keep running */ + } else { + c->ccr &= ~DMA_CCR_EN; + break; + } + } + } + + s->isr |= DMA_FLAG_GIF(ch); + py32_dma_update_irq(s); + } +} + static uint64_t py32_dma_read(void *opaque, hwaddr addr, unsigned size) { PY32DmaState *s = opaque; @@ -991,7 +1078,16 @@ static uint64_t py32_dma_read(void *opaque, hwaddr addr, unsigned size) if (ch < PY32_DMA_CHANNELS) { switch (reg) { case DMA_CCR: return s->ch[ch].ccr; - case DMA_CNDTR: return s->ch[ch].cndtr; + case DMA_CNDTR: + /* + * Deliver any pending serial bytes before answering. This read is + * precisely how App/driver/uart.c discovers new data -- it computes + * a write pointer from the remaining count -- so servicing here + * needs no timer and cannot deliver bytes the guest has not asked + * about yet. + */ + py32_dma_service_usart_rx(s); + return s->ch[ch].cndtr; case DMA_CPAR: return s->ch[ch].cpar; case DMA_CMAR: return s->ch[ch].cmar; default: break; @@ -1021,7 +1117,10 @@ static void py32_dma_write(void *opaque, hwaddr addr, uint64_t value, unsigned s } switch (reg) { - case DMA_CNDTR: s->ch[ch].cndtr = value; break; + case DMA_CNDTR: + s->ch[ch].cndtr = value; + s->ch[ch].total = value; /* remember it; cndtr counts down */ + break; case DMA_CPAR: s->ch[ch].cpar = value; break; case DMA_CMAR: s->ch[ch].cmar = value; break; case DMA_CCR: { @@ -1479,12 +1578,62 @@ struct PY32StubState { char *stub_name; uint32_t size; uint32_t regs[0x100]; + + /* + * Receive path, USART1 only. + * + * A chardev supplies bytes; DR hands them to the guest. The DMA model drains + * this queue on behalf of the circular receive channel, because + * App/driver/uart.c never reads DR directly -- it derives a write pointer from + * the channel's remaining count. + */ + CharBackend chr; + uint8_t rx_fifo[256]; + unsigned rx_head, rx_tail; }; -/* USART_SR transmit flags, from the vendor header: TXE is bit 7, TC bit 6. */ +/* USART_SR flags, from the vendor header. */ +#define PY32_USART_SR_RXNE (1u << 5) #define PY32_USART_SR_TC (1u << 6) #define PY32_USART_SR_TXE (1u << 7) +static bool py32_stub_rx_empty(PY32StubState *s) +{ + return s->rx_head == s->rx_tail; +} + +/* Pull one received byte, or return false when nothing is queued. */ +static bool py32_stub_rx_pop(PY32StubState *s, uint8_t *out) +{ + if (py32_stub_rx_empty(s)) { + return false; + } + *out = s->rx_fifo[s->rx_tail]; + s->rx_tail = (s->rx_tail + 1) % sizeof(s->rx_fifo); + return true; +} + +static int py32_stub_can_receive(void *opaque) +{ + PY32StubState *s = opaque; + const unsigned used = (s->rx_head - s->rx_tail) % sizeof(s->rx_fifo); + return sizeof(s->rx_fifo) - 1 - used; +} + +static void py32_stub_receive(void *opaque, const uint8_t *buf, int size) +{ + PY32StubState *s = opaque; + + for (int i = 0; i < size; i++) { + const unsigned next = (s->rx_head + 1) % sizeof(s->rx_fifo); + if (next == s->rx_tail) { + break; /* full; drop rather than overwrite */ + } + s->rx_fifo[s->rx_head] = buf[i]; + s->rx_head = next; + } +} + static uint64_t py32_stub_read(void *opaque, hwaddr addr, unsigned size) { PY32StubState *s = opaque; @@ -1504,6 +1653,19 @@ static uint64_t py32_stub_read(void *opaque, hwaddr addr, unsigned size) */ if (addr == 0x00 && s->stub_name && !strcmp(s->stub_name, "usart1")) { value |= PY32_USART_SR_TXE | PY32_USART_SR_TC; + /* RXNE so a firmware that polls instead of using DMA also works. */ + if (!py32_stub_rx_empty(s)) { + value |= PY32_USART_SR_RXNE; + } + } + + /* Reading DR consumes a received byte, as on hardware. */ + if (addr == 0x04 && s->stub_name && !strcmp(s->stub_name, "usart1")) { + uint8_t byte; + if (py32_stub_rx_pop(s, &byte)) { + return byte; + } + return 0; } qemu_log_mask(LOG_UNIMP, "py32-%s: read 0x%03" HWADDR_PRIx " -> 0x%08x\n", @@ -1559,7 +1721,20 @@ static void py32_stub_write(void *opaque, hwaddr addr, uint64_t value, unsigned s->regs[idx] = value; } if (addr == 0x04 && s->stub_name && !strcmp(s->stub_name, "usart1")) { - py32_stub_serial_byte((char)(value & 0xff)); + const uint8_t byte = value & 0xff; + + /* Human-readable copy on stderr, which is what the web UI log reads. */ + py32_stub_serial_byte((char)byte); + + /* + * And the raw byte to the chardev, if one is attached. Without this the + * transmit side is invisible to anything on the other end of the port: a + * host tool sends a command, the firmware answers, and the answer only ever + * reaches stderr -- which looks exactly like the firmware ignoring it. + */ + if (qemu_chr_fe_backend_connected(&s->chr)) { + qemu_chr_fe_write_all(&s->chr, &byte, 1); + } } qemu_log_mask(LOG_UNIMP, "py32-%s: write 0x%03" HWADDR_PRIx " = 0x%08" PRIx64 "\n", s->stub_name ?: "stub", addr, value); @@ -1581,11 +1756,22 @@ static void py32_stub_realize(DeviceState *dev, Error **errp) s->stub_name ?: TYPE_PY32_STUB, s->size ? s->size : 0x400); sysbus_init_mmio(SYS_BUS_DEVICE(dev), &s->iomem); + + /* + * Only USART1 takes a chardev: it is the firmware's console and the port the + * UV-K5 programming protocol speaks over. Harmless when unset -- without a + * backend the receive queue simply stays empty, which is the old behaviour. + */ + if (s->stub_name && !strcmp(s->stub_name, "usart1")) { + qemu_chr_fe_set_handlers(&s->chr, py32_stub_can_receive, + py32_stub_receive, NULL, NULL, s, NULL, true); + } } static Property py32_stub_properties[] = { DEFINE_PROP_STRING("stub-name", PY32StubState, stub_name), DEFINE_PROP_UINT32("size", PY32StubState, size, 0x400), + DEFINE_PROP_CHR("chardev", PY32StubState, chr), DEFINE_PROP_END_OF_LIST(), }; @@ -1797,11 +1983,30 @@ static void py32f071_soc_realize(DeviceState *dev_soc, Error **errp) for (int i = 0; i < PY32_NUM_STUB; i++) { qdev_prop_set_string(DEVICE(&s->stub[i]), "stub-name", py32_stubs[i].name); qdev_prop_set_uint32(DEVICE(&s->stub[i]), "size", py32_stubs[i].size); + + /* + * Give USART1 a chardev so something can talk *to* the firmware. This is + * the port the UV-K5 programming protocol runs over (App/app/uart.c: + * 0x0514 handshake, 0x051B/0x051D EEPROM read and write, 0x05DD reset). + * Defaults to "serial0", so -serial on the command line just works. + */ + if (!strcmp(py32_stubs[i].name, "usart1")) { + Chardev *chr = serial_hd(0); + if (chr) { + qdev_prop_set_chr(DEVICE(&s->stub[i]), "chardev", chr); + } + } + if (!sysbus_realize(SYS_BUS_DEVICE(&s->stub[i]), errp)) { return; } memory_region_add_subregion(&s->container, py32_stubs[i].base, sysbus_mmio_get_region(SYS_BUS_DEVICE(&s->stub[i]), 0)); + + /* DMA drains USART1's receive queue; see py32_dma_service_usart_rx. */ + if (!strcmp(py32_stubs[i].name, "usart1")) { + s->dma.usart1 = &s->stub[i]; + } } /* diff --git a/tools/test_serial_rx.py b/tools/test_serial_rx.py new file mode 100755 index 0000000..987b8d6 --- /dev/null +++ b/tools/test_serial_rx.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""The firmware must receive serial bytes and answer a programming command. + +Serial receive used to be impossible. USART1 was a register stub with no chardev, so +nothing could be sent in; and App/driver/uart.c locates incoming data with + + write_ptr = sizeof(UART_DMA_Buffer) - LL_DMA_GetDataLength(DMA1, CHANNEL_2) + +over a circular DMA channel, while the DMA model only ever serviced SPI and never +decremented CNDTR for USART. That expression was therefore always 0 and the buffer +always looked empty, which made the whole UV-K5 programming protocol in App/app/uart.c +unreachable: 0x0514 handshake, 0x051B/0x051D EEPROM read and write, 0x05DD reset. + +This sends a real 0x0514 handshake and checks for a reply. + +Wire format, from App/app/uart.c: + AB CD DC BA +The payload is obfuscated with a fixed XOR key, and the CRC is CRC-16/XMODEM over the +payload. Replies use the same framing. +""" +import gzip +import os +import shutil +import socket +import struct +import subprocess +import sys +import tempfile +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.dirname(HERE) +QEMU = os.path.expanduser("~/qemu-build/qemu-7.2+dfsg/build/qemu-system-arm") +ELF = os.path.expanduser("~/uvk5-port/uvk5-sat/build/CW/nr7y.cw.elf") +PRISTINE = os.path.join(ROOT, "assets", "pristine", "flash-pristine.img.gz") + +BOOT_SECONDS = 20 + +# App/app/uart.c: Obfuscation[] applied to the payload of every frame. +XOR_KEY = bytes([ + 0x16, 0x6C, 0x14, 0xE6, 0x2E, 0x91, 0x0D, 0x40, + 0x21, 0x35, 0xD5, 0x40, 0x13, 0x03, 0xE9, 0x80, +]) + + +def crc16_xmodem(data: bytes) -> int: + crc = 0 + for byte in data: + crc ^= byte << 8 + for _ in range(8): + crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF + return crc + + +def obfuscate(payload: bytes) -> bytes: + return bytes(b ^ XOR_KEY[i % len(XOR_KEY)] for i, b in enumerate(payload)) + + +def build_frame(payload: bytes) -> bytes: + body = payload + struct.pack(" len(buf): + break + length = struct.unpack_from(" len(buf) or length > 512: + i = start + 2 + continue + body = obfuscate(buf[start + 4:end]) + out.append(body[:length]) + i = end + 2 + return out + + +def main(): + for path, what in ((QEMU, "QEMU"), (ELF, "firmware"), (PRISTINE, "pristine image")): + if not os.path.exists(path): + sys.exit(f"missing {what}: {path}") + + workdir = tempfile.mkdtemp(prefix="uvk5-serial-") + image = os.path.join(workdir, "flash.img") + sock_path = os.path.join(workdir, "serial.sock") + with gzip.open(PRISTINE, "rb") as src, open(image, "wb") as dst: + shutil.copyfileobj(src, dst) + + # A listening socket for QEMU's serial chardev to connect back to. + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(sock_path) + srv.listen(1) + srv.settimeout(40) + + proc = subprocess.Popen( + [QEMU, "-M", f"uv-k5-v3,flash-image={image}", "-nographic", "-monitor", "none", + "-serial", f"unix:{sock_path}", "-kernel", ELF], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + failures = [] + try: + conn, _ = srv.accept() + conn.settimeout(15) + print("serial connected") + time.sleep(BOOT_SECONDS) + + # Drain the boot banner the firmware transmits, so it is not mistaken for a + # reply. Transmit already worked; this test is about the other direction. + conn.setblocking(False) + banner = b"" + deadline = time.time() + 2 + while time.time() < deadline: + try: + chunk = conn.recv(4096) + if not chunk: + break + banner += chunk + except BlockingIOError: + time.sleep(0.1) + print(f"boot output: {len(banner)} bytes" + f"{' (' + banner[:40].decode(errors='replace').strip() + ')' if banner else ''}") + conn.setblocking(True) + + # 0x0514: hello. Payload is the command id plus a 4-byte timestamp. + payload = struct.pack("