diff --git a/Cargo.lock b/Cargo.lock index f2bc6e2..319ef2d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -472,6 +472,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "powerfmt", +] + [[package]] name = "digest" version = "0.10.7" @@ -782,9 +791,11 @@ dependencies = [ "esp-idf-svc", "p256", "p384", + "rsa", "serde", "serde_json", "sha2", + "time", "tracing", "tracing-subscriber", "x509-cert", @@ -1274,6 +1285,9 @@ name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +dependencies = [ + "spin", +] [[package]] name = "leb128fmt" @@ -1297,6 +1311,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "libm" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" + [[package]] name = "libredox" version = "0.1.16" @@ -1401,6 +1421,48 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint-dig" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7" +dependencies = [ + "lazy_static", + "libm", + "num-integer", + "num-iter", + "num-traits", + "rand", + "smallvec", + "zeroize", +] + +[[package]] +name = "num-conv" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" + +[[package]] +name = "num-integer" +version = "0.1.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f" +dependencies = [ + "num-traits", +] + +[[package]] +name = "num-iter" +version = "0.1.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1429034a0490724d0075ebb2bc9e875d6503c3cf69e235a8941aa757d83ef5bf" +dependencies = [ + "autocfg", + "num-integer", + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1408,6 +1470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", + "libm", ] [[package]] @@ -1483,6 +1546,17 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkcs1" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f" +dependencies = [ + "der", + "pkcs8", + "spki", +] + [[package]] name = "pkcs8" version = "0.10.2" @@ -1508,6 +1582,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -1582,6 +1671,26 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" +dependencies = [ + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -1667,6 +1776,27 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rsa" +version = "0.9.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" +dependencies = [ + "const-oid", + "digest", + "num-bigint-dig", + "num-integer", + "num-traits", + "pkcs1", + "pkcs8", + "rand_core", + "sha2", + "signature", + "spki", + "subtle", + "zeroize", +] + [[package]] name = "rustc-hash" version = "2.1.2" @@ -1879,6 +2009,12 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" + [[package]] name = "spki" version = "0.7.3" @@ -2040,6 +2176,37 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +dependencies = [ + "deranged", + "itoa", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" + +[[package]] +name = "time-macros" +version = "0.2.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -2712,6 +2879,26 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zerocopy" +version = "0.8.48" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.48" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "zerofrom" version = "0.1.7" diff --git a/Cargo.toml b/Cargo.toml index 6019f0c..4e04cb2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,10 @@ p256 = { version = "0.13", features = ["ecdsa"] } # leaf signs DSSE p384 = { version = "0.13", features = ["ecdsa"] } # intermediate signs leaf x509-cert = { version = "0.2", features = ["pem"] } base64 = "0.22" +# Cloud Logging: RSA-PKCS#1v1.5 SHA-256 signing for service-account JWTs. +rsa = { version = "0.9", default-features = false, features = ["std", "pem", "sha2"] } +# RFC 3339 timestamp formatting for Cloud Logging entries. +time = { version = "0.3", default-features = false, features = ["std", "formatting"] } [build-dependencies] embuild = "0.33" diff --git a/src/cloud_log.rs b/src/cloud_log.rs index 369682f..3357c61 100644 --- a/src/cloud_log.rs +++ b/src/cloud_log.rs @@ -11,8 +11,18 @@ //! missing required keys (`project_id`, `sa_email`, `sa_key_id`, //! `sa_key_pem`), the firmware boots normally with serial-only logs. -use anyhow::{anyhow, Result}; +use anyhow::{anyhow, bail, Context, Result}; +use base64::Engine; +use embedded_svc::http::client::Client; +use embedded_svc::io::Write as _; +use esp_idf_svc::http::client::{Configuration as HttpConfig, EspHttpConnection, FollowRedirectsPolicy}; +use esp_idf_svc::http::Method; use esp_idf_svc::nvs::{EspDefaultNvsPartition, EspNvs, NvsDefault}; +use rsa::pkcs1v15::SigningKey; +use rsa::pkcs8::DecodePrivateKey; +use rsa::signature::{SignatureEncoding, Signer}; +use rsa::RsaPrivateKey; +use serde::{Deserialize, Serialize}; use std::collections::VecDeque; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -187,6 +197,13 @@ impl CloudLogLayer { impl Layer for CloudLogLayer { fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { + // Skip events emitted by this module itself, otherwise the + // sender's "post failed" / "token refreshed" tracing calls + // get pushed onto the queue it's draining, creating a tight + // feedback loop. Cloud_log's own messages stay serial-only. + if event.metadata().target() == module_path!() { + return; + } let level = *event.metadata().level(); if level > self.min_level { // tracing's Level ordering: TRACE > DEBUG > INFO > WARN > ERROR. @@ -276,58 +293,374 @@ impl FieldCapture { } } -/// Background task. Periodically drains the queue and (in this stub) -/// logs each entry to serial showing what it WOULD send. The real POST -/// to Cloud Logging is in a follow-up commit. +/// Sender thread main loop. Drains the queue every `FLUSH_INTERVAL`, +/// mints/refreshes a service-account access token as needed, and +/// POSTs batches to Cloud Logging. +/// +/// Uses tracing internally — the `module_path!()` filter in +/// `CloudLogLayer::on_event` keeps cloud_log's own messages out of +/// the queue (no feedback loop). pub fn run(cfg: GcpConfig, queue: LogQueue) -> ! { - // Use eprintln rather than tracing here to avoid feedback loops - // (this thread emitting tracing events that the layer would push - // back onto the queue). - eprintln!( - "cloud_log: stub sender starting (project={}, sa={}, key_id={}, key_pem_bytes={}, min_severity={:?})", - cfg.project_id, - cfg.sa_email, - cfg.sa_key_id, - cfg.sa_key_pem.len(), - cfg.min_severity + tracing::info!( + project = %cfg.project_id, + sa = %cfg.sa_email, + min_severity = ?cfg.min_severity, + "cloud_log: sender starting", ); + + // Parse the SA private key once at startup; if it's malformed + // we can't do anything useful and log loudly forever. + let signing_key = match parse_signing_key(&cfg.sa_key_pem) { + Ok(k) => k, + Err(e) => { + loop { + tracing::error!( + error = %format!("{:#}", e), + "cloud_log: SA key parse failed; sender disabled", + ); + std::thread::sleep(Duration::from_secs(300)); + } + } + }; + + let mac = device_mac(); + let log_name = format!("projects/{}/logs/esp32-firmware", cfg.project_id); + + let mut token: Option = None; + let mut consecutive_failures: u32 = 0; + loop { - std::thread::sleep(FLUSH_INTERVAL); + let sleep_for = if consecutive_failures > 0 { + // Exponential backoff capped at 5 min for cloud-logging + // failures (separate budget from the OTA loop). + let exp = consecutive_failures.min(5); + Duration::from_secs(FLUSH_INTERVAL.as_secs() << exp).min(Duration::from_secs(300)) + } else { + FLUSH_INTERVAL + }; + std::thread::sleep(sleep_for); + let batch = queue.drain(BATCH_MAX_ENTRIES); if batch.is_empty() { + consecutive_failures = 0; continue; } - eprintln!( - "cloud_log: would POST batch of {} entries to projects/{}/logs/esp32-firmware", - batch.len(), - cfg.project_id - ); - for entry in &batch { - // One-line summary per entry. - let ts = entry - .timestamp_unix_secs - .map(|s| s.to_string()) - .unwrap_or_else(|| "".into()); - eprintln!( - " [{}] {:?} {} {}{}{}", - ts, - entry.severity, - entry.target, - entry.message, - if entry.fields.is_empty() { - String::new() - } else { - format!( - " {}", - serde_json::to_string(&entry.fields).unwrap_or_default() - ) - }, - if entry.dropped_before > 0 { - format!(" (dropped {} before)", entry.dropped_before) - } else { - String::new() - }, - ); + + if token.as_ref().map_or(true, |t| t.expired_or_close()) { + match mint_access_token(&cfg, &signing_key) { + Ok(t) => { + tracing::info!( + expires_in_secs = t.expires_at_unix.saturating_sub(now_unix_secs().unwrap_or(0)), + "cloud_log: minted new access token", + ); + token = Some(t); + } + Err(e) => { + consecutive_failures = consecutive_failures.saturating_add(1); + tracing::warn!( + failures = consecutive_failures, + error = %format!("{:#}", e), + "cloud_log: token mint failed", + ); + continue; + } + } + } + + let bearer = &token.as_ref().unwrap().token; + match post_batch(&log_name, &cfg.project_id, &mac, bearer, &batch) { + Ok(()) => { + tracing::info!( + entries = batch.len(), + "cloud_log: posted batch", + ); + consecutive_failures = 0; + } + Err(e) => { + consecutive_failures = consecutive_failures.saturating_add(1); + tracing::warn!( + entries = batch.len(), + failures = consecutive_failures, + error = %format!("{:#}", e), + "cloud_log: post failed; dropping batch", + ); + // Drop the batch on failure rather than re-enqueue + // (avoids unbounded growth on a long outage). Loss is + // already surfaced via dropped_before on subsequent + // entries. + } } } } + +struct CachedToken { + token: String, + expires_at_unix: u64, +} + +impl CachedToken { + fn expired_or_close(&self) -> bool { + // Refresh 5 minutes before expiry to avoid races. + match now_unix_secs() { + Some(now) => now + 300 >= self.expires_at_unix, + None => true, + } + } +} + +fn parse_signing_key(pem_bytes: &[u8]) -> Result> { + let pem_str = std::str::from_utf8(pem_bytes).context("SA key PEM is not UTF-8")?; + let key = RsaPrivateKey::from_pkcs8_pem(pem_str).context("parse SA PKCS#8 private key")?; + Ok(SigningKey::::new(key)) +} + +#[derive(Serialize)] +struct JwtHeader<'a> { + alg: &'static str, + typ: &'static str, + kid: &'a str, +} + +#[derive(Serialize)] +struct JwtClaims<'a> { + iss: &'a str, + scope: &'static str, + aud: &'static str, + iat: u64, + exp: u64, +} + +#[derive(Deserialize)] +struct TokenResp { + access_token: String, + expires_in: u64, +} + +fn mint_access_token( + cfg: &GcpConfig, + signing_key: &SigningKey, +) -> Result { + let now = now_unix_secs().ok_or_else(|| anyhow!("NTP not synced; cannot mint JWT"))?; + + let header = JwtHeader { + alg: "RS256", + typ: "JWT", + kid: &cfg.sa_key_id, + }; + let claims = JwtClaims { + iss: &cfg.sa_email, + scope: "https://www.googleapis.com/auth/logging.write", + aud: "https://oauth2.googleapis.com/token", + iat: now, + exp: now + 3600, + }; + + let header_b64 = b64url(&serde_json::to_vec(&header)?); + let claims_b64 = b64url(&serde_json::to_vec(&claims)?); + let signing_input = format!("{}.{}", header_b64, claims_b64); + let sig = signing_key.sign(signing_input.as_bytes()); + let sig_b64 = b64url(sig.to_bytes().as_ref()); + let jwt = format!("{}.{}", signing_input, sig_b64); + + let body = format!( + "grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer&assertion={}", + jwt + ); + let resp_bytes = http_post( + "https://oauth2.googleapis.com/token", + "application/x-www-form-urlencoded", + body.as_bytes(), + None, + ) + .context("POST oauth2/token")?; + let resp: TokenResp = serde_json::from_slice(&resp_bytes) + .context("parse token response JSON")?; + + Ok(CachedToken { + token: resp.access_token, + expires_at_unix: now + resp.expires_in, + }) +} + +#[derive(Serialize)] +struct WriteEntriesRequest<'a> { + #[serde(rename = "logName")] + log_name: &'a str, + resource: MonitoredResource<'a>, + entries: Vec, +} + +#[derive(Serialize)] +struct MonitoredResource<'a> { + #[serde(rename = "type")] + type_: &'static str, + labels: ResourceLabels<'a>, +} + +#[derive(Serialize)] +struct ResourceLabels<'a> { + project_id: &'a str, + location: &'static str, + namespace: &'static str, + node_id: &'a str, +} + +#[derive(Serialize)] +struct Entry { + severity: &'static str, + #[serde(rename = "jsonPayload")] + json_payload: serde_json::Value, + #[serde(rename = "timestamp", skip_serializing_if = "Option::is_none")] + timestamp: Option, +} + +fn post_batch( + log_name: &str, + project_id: &str, + mac: &str, + bearer: &str, + batch: &[LogEntry], +) -> Result<()> { + let entries: Vec = batch + .iter() + .map(|e| Entry { + severity: severity_str(e.severity), + json_payload: build_payload(e), + timestamp: e.timestamp_unix_secs.and_then(unix_to_rfc3339), + }) + .collect(); + let _ = (project_id, mac); // used inline in MonitoredResource below + + let req = WriteEntriesRequest { + log_name, + resource: MonitoredResource { + type_: "generic_node", + labels: ResourceLabels { + project_id, + location: "global", + namespace: "esp32", + node_id: mac, + }, + }, + entries, + }; + let body = serde_json::to_vec(&req)?; + let auth = format!("Bearer {}", bearer); + http_post( + "https://logging.googleapis.com/v2/entries:write", + "application/json", + &body, + Some(&auth), + ) + .map(|_| ()) +} + +fn severity_str(level: Level) -> &'static str { + // Cloud Logging severities (LogSeverity enum): + // https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#logseverity + match level { + Level::TRACE => "DEBUG", + Level::DEBUG => "DEBUG", + Level::INFO => "INFO", + Level::WARN => "WARNING", + Level::ERROR => "ERROR", + } +} + +fn build_payload(e: &LogEntry) -> serde_json::Value { + let mut map = e.fields.clone(); + map.insert( + "message".to_string(), + serde_json::Value::String(e.message.clone()), + ); + map.insert( + "module".to_string(), + serde_json::Value::String(e.target.clone()), + ); + if e.dropped_before > 0 { + map.insert( + "dropped_before".to_string(), + serde_json::Value::Number(e.dropped_before.into()), + ); + } + serde_json::Value::Object(map) +} + +fn unix_to_rfc3339(secs: u64) -> Option { + use time::format_description::well_known::Rfc3339; + use time::OffsetDateTime; + OffsetDateTime::from_unix_timestamp(secs as i64) + .ok() + .and_then(|dt| dt.format(&Rfc3339).ok()) +} + +fn b64url(input: &[u8]) -> String { + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(input) +} + +/// Single helper for HTTPS POST. Returns the response body bytes. +/// Errors on any non-2xx status with the body included for diagnosis. +fn http_post( + url: &str, + content_type: &str, + body: &[u8], + bearer: Option<&str>, +) -> Result> { + let conn = EspHttpConnection::new(&HttpConfig { + crt_bundle_attach: Some(esp_idf_svc::sys::esp_crt_bundle_attach), + follow_redirects_policy: FollowRedirectsPolicy::FollowAll, + timeout: Some(Duration::from_secs(30)), + buffer_size: Some(2048), + buffer_size_tx: Some(4096), + ..Default::default() + })?; + let mut client = Client::wrap(conn); + let body_len = body.len().to_string(); + let mut headers: Vec<(&str, &str)> = vec![ + ("content-type", content_type), + ("content-length", body_len.as_str()), + ("accept", "application/json"), + ]; + if let Some(b) = bearer { + headers.push(("authorization", b)); + } + let mut req = client.request(Method::Post, url, &headers)?; + req.write_all(body).context("write request body")?; + req.flush().ok(); + let mut resp = req.submit()?; + let status = resp.status(); + let mut buf = Vec::with_capacity(1024); + let mut chunk = [0u8; 1024]; + loop { + let n = resp.read(&mut chunk).context("read response chunk")?; + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + if !(200..300).contains(&status) { + bail!( + "POST {} -> HTTP {} body={}", + url, + status, + String::from_utf8_lossy(&buf) + ); + } + Ok(buf) +} + +/// MAC address as a `aabbccddeeff` hex string. Used as the `node_id` +/// label on Cloud Logging entries to identify which device the log +/// came from. +fn device_mac() -> String { + let mut mac = [0u8; 8]; + unsafe { + esp_idf_svc::sys::esp_efuse_mac_get_default(mac.as_mut_ptr()); + } + let mut s = String::with_capacity(12); + for b in &mac[..6] { + use std::fmt::Write as _; + let _ = write!(s, "{:02x}", b); + } + s +} diff --git a/src/main.rs b/src/main.rs index e64fc98..3959ff5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -93,6 +93,36 @@ fn main() -> Result<()> { "wifi connected", ); + // SNTP is only needed for cloud logging — the service-account JWT + // auth requires real wall-clock time for `iat`/`exp` (Google rejects + // tokens minted from a 1970 clock). Devices without `[gcp]` skip + // the wait and boot faster. Cloud Logging entry timestamps + // themselves are assigned server-side by GCP if we omit them. + let _sntp = if gcp.is_some() { + let sntp = esp_idf_svc::sntp::EspSntp::new_default()?; + let started = std::time::Instant::now(); + loop { + use esp_idf_svc::sntp::SyncStatus; + if sntp.get_sync_status() == SyncStatus::Completed { + tracing::info!( + elapsed_ms = started.elapsed().as_millis() as u64, + "ntp: synced", + ); + break; + } + if started.elapsed() > Duration::from_secs(15) { + tracing::warn!( + "ntp: not synced after 15s, cloud logging will fail until clock catches up", + ); + break; + } + std::thread::sleep(Duration::from_millis(200)); + } + Some(sntp) + } else { + None + }; + fetch("https://api.ipify.org?format=json")?; fetch("https://wttr.in/?format=3")?;