Moved mqtt based updater out of the pico_p1 project

updater/Cargo.toml

@@ -0,0 +1,59 @@
+[package]
+name = "updater"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+cortex-m = { version = "0.7", features = ["inline-asm"] }
+cortex-m-rt = "0.7"
+defmt = "0.3"
+defmt-rtt = "0.4"
+embassy-net = { version = "0.1", features = [
+	"tcp",
+	"dhcpv4",
+	"nightly",
+	"medium-ethernet",
+	"defmt",
+	"dns",
+] }
+embassy-boot = { version = "0.1", features = [
+	"nightly",
+	"defmt",
+	"ed25519-salty",
+] }
+embassy-time = { version = "0.1", features = [
+	"defmt",
+	"unstable-traits",
+	"defmt-timestamp-uptime",
+	"nightly",
+] }
+rand_core = "0.6.4"
+embedded-io-async = { version = "0.5", features = ["defmt-03"] }
+embedded-storage = "0.3.0"
+rust-mqtt = { version = "0.1.5", features = [
+	"defmt",
+	"no_std",
+	"tls",
+], default-features = false }
+nourl = { version = "0.1.1", features = ["defmt"] }
+heapless = { version = "0.7.16", features = ["defmt", "serde"] }
+serde = { version = "1.0.188", default-features = false, features = ["derive"] }
+serde-json-core = "0.5.1"
+embedded-tls = { version = "0.15.0", default-features = false, features = [
+	"async",
+	"defmt",
+] }
+reqwless = { version = "0.5.0", features = ["defmt"] }
+static_cell = { version = "1.2.0", features = ["nightly"] }
+
+[patch.crates-io]
+embassy-net = { git = "https://github.com/embassy-rs/embassy" }
+embassy-boot = { git = "https://github.com/embassy-rs/embassy" }
+embassy-time = { git = "https://github.com/embassy-rs/embassy" }
+
+# Updated to embedded-io 0.5.0
+rust-mqtt = { git = "https://git.huizinga.dev/Dreaded_X/rust-mqtt" }
+# Make mqtt:// and mqtts:// actually work
+nourl = { git = "https://git.huizinga.dev/Dreaded_X/nourl" }
+# Waiting for this to get updated to embedded-io 0.5 properly
+reqwless = { path = "../../reqwless" }

updater/src/lib.rs

@@ -0,0 +1,211 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::*;
+use embassy_boot::{AlignedBuffer, BlockingFirmwareUpdater};
+use embassy_net::{dns::DnsQueryType, driver::Driver, tcp::TcpSocket, Stack};
+use embassy_time::{Duration, Timer};
+use embedded_io_async::{Read, Write};
+use embedded_storage::nor_flash::NorFlash;
+use embedded_tls::{Aes128GcmSha256, NoVerify, TlsConfig, TlsConnection, TlsContext};
+use heapless::Vec;
+use nourl::Url;
+use rand_core::{CryptoRng, RngCore};
+use reqwless::{
+    request::{Method, Request, RequestBuilder},
+    response::Response,
+};
+use rust_mqtt::{
+    client::{client::MqttClient, client_config::ClientConfig},
+    packet::v5::publish_packet::QualityOfService,
+};
+use serde::Serialize;
+use static_cell::make_static;
+
+#[derive(Serialize)]
+#[serde(rename_all = "snake_case", tag = "status")]
+enum Status<'a> {
+    Connected { version: &'a str },
+    Disconnected,
+    PreparingUpdate,
+    Erasing,
+    Writing { progress: u32 },
+    Verifying,
+    UpdateComplete,
+}
+
+impl Status<'_> {
+    fn vec(&self) -> Vec<u8, 1024> {
+        serde_json_core::to_vec(self)
+            .expect("The buffer should be large enough to contain all the data")
+    }
+}
+
+// TODO: Make this the owner of the blocking firmware updater
+// TODO: When fixed, use the async firmware updater
+pub struct Updater {
+    topic_status: &'static str,
+    topic_update: &'static str,
+    version: &'static str,
+    public_key: &'static [u8],
+}
+
+impl Updater {
+    pub fn new(
+        topic_status: &'static str,
+        topic_update: &'static str,
+        version: &'static str,
+        public_key: &'static [u8],
+    ) -> Self {
+        Self {
+            topic_status,
+            topic_update,
+            version,
+            public_key,
+        }
+    }
+
+    pub fn add_will<const MAX_PROPERTIES: usize>(
+        &self,
+        config: &mut ClientConfig<'_, MAX_PROPERTIES, impl RngCore>,
+    ) {
+        let msg = make_static!(Status::Disconnected.vec());
+        config.add_will(self.topic_status, msg, true);
+    }
+
+    pub async fn ready<const MAX_PROPERTIES: usize>(
+        &self,
+        client: &mut MqttClient<'_, impl Write + Read, MAX_PROPERTIES, impl RngCore>,
+    ) {
+        let status = Status::Connected {
+            version: self.version,
+        }
+        .vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, true)
+            .await
+            .unwrap();
+
+        client.subscribe_to_topic(self.topic_update).await.unwrap();
+    }
+
+    pub async fn update<const MAX_PROPERTIES: usize>(
+        &self,
+        stack: &'static Stack<impl Driver>,
+        updater: &mut BlockingFirmwareUpdater<'_, impl NorFlash, impl NorFlash>,
+        rng: &mut (impl RngCore + CryptoRng),
+        client: &mut MqttClient<'_, impl Write + Read, MAX_PROPERTIES, impl RngCore>,
+        url: Url<'_>,
+    ) {
+        info!("Preparing for OTA...");
+        let status = Status::PreparingUpdate.vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+            .await
+            .unwrap();
+
+        let ip = stack.dns_query(url.host(), DnsQueryType::A).await.unwrap()[0];
+
+        let mut rx_buffer = [0; 1024];
+        let mut tx_buffer = [0; 1024];
+
+        let mut socket = TcpSocket::new(stack, &mut rx_buffer, &mut tx_buffer);
+
+        let addr = (ip, url.port_or_default());
+        debug!("Addr: {}", addr);
+        socket.connect(addr).await.unwrap();
+
+        let mut read_record_buffer = [0; 16384 * 2];
+        let mut write_record_buffer = [0; 16384];
+        let mut tls: TlsConnection<TcpSocket, Aes128GcmSha256> =
+            TlsConnection::new(socket, &mut read_record_buffer, &mut write_record_buffer);
+        tls.open::<_, NoVerify>(TlsContext::new(&TlsConfig::new(), rng))
+            .await
+            .unwrap();
+
+        debug!("Path: {}", url.path());
+        Request::get(url.path())
+            .host(url.host())
+            .build()
+            .write(&mut tls)
+            .await
+            .unwrap();
+
+        let mut headers = [0; 1024];
+        let resp = Response::read(&mut tls, Method::GET, &mut headers)
+            .await
+            .unwrap();
+
+        let mut body = resp.body().reader();
+
+        debug!("Erasing flash...");
+        let status = Status::Erasing.vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+            .await
+            .unwrap();
+
+        let writer = updater
+            .prepare_update()
+            .map_err(|e| warn!("E: {:?}", Debug2Format(&e)))
+            .unwrap();
+
+        debug!("Writing...");
+        let status = Status::Writing { progress: 0 }.vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+            .await
+            .unwrap();
+
+        // The first 64 bytes of the file contain the signature
+        let mut signature = [0; 64];
+        body.read_exact(&mut signature).await.unwrap();
+
+        trace!("Signature: {:?}", signature);
+
+        let mut buffer = AlignedBuffer([0; 4096]);
+        let mut size = 0;
+        while let Ok(read) = body.read(&mut buffer.0).await {
+            if read == 0 {
+                break;
+            }
+            debug!("Writing chunk: {}", read);
+            writer.write(size, &buffer.0[..read]).unwrap();
+            size += read as u32;
+
+            let status = Status::Writing { progress: size }.vec();
+            client
+                .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+                .await
+                .unwrap();
+        }
+        debug!("Total size: {}", size);
+
+        let status = Status::Verifying.vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+            .await
+            .unwrap();
+
+        updater
+            .verify_and_mark_updated(self.public_key, &signature, size)
+            .unwrap();
+
+        // Update mqtt message should be send using retain
+        // TODO: Clear the message
+
+        let status = Status::UpdateComplete.vec();
+        client
+            .send_message(self.topic_status, &status, QualityOfService::QoS1, false)
+            .await
+            .unwrap();
+
+        client.disconnect().await.unwrap();
+
+        info!("Restarting in 5 seconds...");
+        Timer::after(Duration::from_secs(5)).await;
+
+        cortex_m::peripheral::SCB::sys_reset();
+    }
+}