mirror of
https://github.com/imjasonh/esp32
synced 2026-07-06 23:52:24 +00:00
cloud_log: real sender — JWT, RSA-SHA256, OAuth2, POST to logging.googleapis.com
Replaces the stub eprintln sender with the real implementation: - mint_access_token: builds + signs a service-account JWT (RS256 over base64url(header).base64url(claims)), POSTs to oauth2.googleapis.com with the standard grant_type=...jwt-bearer form body, parses the access_token + expires_in. - CachedToken: held in the sender thread's local state. Refreshes 300 seconds before expiry (Google issues 1h tokens; we re-mint at T+55min). - post_batch: serializes the WriteEntriesRequest with logName, resource (generic_node + project + MAC node_id), and entries (severity + jsonPayload + optional timestamp), POSTs to logging.googleapis.com/v2/entries:write with the bearer. - Backoff on failure: sleep doubles on each consecutive failure (capped at 5 min). Batches dropped on POST failure rather than re-enqueued; loss is surfaced via dropped_before on the next entry. Crates added: rsa 0.9 (with sha2 feature for AssociatedOid; PKCS#1 v1.5 needs the DigestInfo prefix), time 0.3 (RFC3339 formatting). main.rs: SNTP startup is now gated on gcp.is_some(). The JWT auth needs a real wall-clock for / (Google rejects ~5min skew), so devices with cloud logging pay a one-time ~few-second sync at boot. Devices without [gcp] skip it. Log entry timestamps themselves are still optional — when omitted, GCP server-side assigns them. Cloud Logging severity mapping: tracing TRACE/DEBUG -> DEBUG, INFO -> INFO, WARN -> WARNING, ERROR -> ERROR (matches GCP's LogSeverity enum). device_mac() reads via esp_efuse_mac_get_default and uses the lower 6 bytes as the node_id resource label, so multi-device logs can be filtered by MAC. Firmware size: 1.60 MB -> 1.71 MB (+rsa, +time, +sender code). Still under the 1.94 MB slot. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
7cb4e8c75c
commit
1264ca7da1
4 changed files with 600 additions and 46 deletions
187
Cargo.lock
generated
187
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
425
src/cloud_log.rs
425
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<S: Subscriber> Layer<S> 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<CachedToken> = 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(|| "<no-ntp>".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<SigningKey<sha2::Sha256>> {
|
||||
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::<sha2::Sha256>::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<sha2::Sha256>,
|
||||
) -> Result<CachedToken> {
|
||||
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<Entry>,
|
||||
}
|
||||
|
||||
#[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<String>,
|
||||
}
|
||||
|
||||
fn post_batch(
|
||||
log_name: &str,
|
||||
project_id: &str,
|
||||
mac: &str,
|
||||
bearer: &str,
|
||||
batch: &[LogEntry],
|
||||
) -> Result<()> {
|
||||
let entries: Vec<Entry> = 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<String> {
|
||||
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<Vec<u8>> {
|
||||
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
|
||||
}
|
||||
|
|
|
|||
30
src/main.rs
30
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")?;
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue