diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 2a3110a8..5b3efdb3 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -21,6 +21,9 @@ jobs: with: node-version: "22" + - name: Install Wrangler + run: npm install --global wrangler@4.80.0 + - name: Install wasm-pack run: cargo install wasm-pack@0.13.1 --locked @@ -39,7 +42,7 @@ jobs: # committed to the repository. - name: Start wrangler dev working-directory: crates/ct_worker - run: npx wrangler@4.80.0 -e=dev dev --port 8787 --persist-to .wrangler/state & + run: wrangler -e=dev dev --port 8787 --persist-to .wrangler/state & - name: Wait for wrangler dev to be ready run: | @@ -72,6 +75,9 @@ jobs: with: node-version: "22" + - name: Install Wrangler + run: npm install --global wrangler@4.80.0 + - name: Install wasm-pack run: cargo install wasm-pack@0.13.1 --locked @@ -89,7 +95,7 @@ jobs: # committed to the repository. - name: Start wrangler dev working-directory: crates/bootstrap_mtc_worker - run: npx wrangler@4.80.0 -e=dev dev --port 8787 --persist-to .wrangler/state & + run: wrangler -e=dev dev --port 8787 --persist-to .wrangler/state & - name: Wait for wrangler dev to be ready run: | diff --git a/crates/bootstrap_mtc_worker/config.schema.json b/crates/bootstrap_mtc_worker/config.schema.json index a452e227..2bf708bd 100644 --- a/crates/bootstrap_mtc_worker/config.schema.json +++ b/crates/bootstrap_mtc_worker/config.schema.json @@ -51,8 +51,7 @@ }, "monitoring_url": { "type": "string", - "default": "", - "description": "URL for log monitoring. If unspecified, use the submission URL and the Worker will proxy requests to the R2 bucket." + "description": "URL for log monitoring. Omit when no monitoring service is configured." }, "location_hint": { "type": "string", @@ -117,4 +116,4 @@ "required": [ "logs" ] -} \ No newline at end of file +} diff --git a/crates/bootstrap_mtc_worker/src/frontend_worker.rs b/crates/bootstrap_mtc_worker/src/frontend_worker.rs index b4877ec7..6612fe89 100644 --- a/crates/bootstrap_mtc_worker/src/frontend_worker.rs +++ b/crates/bootstrap_mtc_worker/src/frontend_worker.rs @@ -23,7 +23,7 @@ use generic_log_worker::{ log_ops::{CHECKPOINT_KEY, ProofError, prove_subtree_inclusion, read_leaf}, obs::{Wshim, metrics}, serialize, - util::{WorkerByteStream, now_millis}, + util::now_millis, }; use serde::{Deserialize, Serialize}; use serde_with::{base64::Base64, serde_as}; @@ -39,7 +39,7 @@ use axum::{ Json, Router, body::Bytes, extract::{Path, State}, - http::{HeaderMap, StatusCode, header}, + http::{StatusCode, header}, middleware, response::{AppendHeaders, IntoResponse}, routing::{get, post}, @@ -61,7 +61,8 @@ struct MetadataResponse<'a> { #[serde_as(as = "Base64")] cosigner_public_key: &'a [u8], submission_url: &'a str, - monitoring_url: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + monitoring_url: Option<&'a str>, } // POST body structure for the `/get-certificate` endpoint @@ -112,7 +113,6 @@ async fn main( .route("/logs/{log}/get-landmark-bundle", get(get_landmark_bundle)) .route("/logs/{log}/metadata", get(metadata)) .route("/logs/{log}/sequencer_id", get(sequencer_id)) - .route("/logs/{log}/{*key}", get(get_object)) .layer(middleware::from_fn_with_state( (env.clone(), metrics::FrontendWorkerMetrics::new(®istry)), request_metrics, @@ -137,28 +137,18 @@ async fn main( } #[derive(serde::Deserialize)] -struct PathParams { +struct PathParams { log: String, - #[serde(flatten)] - rest: Rest, -} - -#[derive(serde::Deserialize)] -struct Key { - key: String, } -impl axum::extract::FromRequestParts for PathParams -where - T: serde::de::DeserializeOwned + Send, -{ +impl axum::extract::FromRequestParts for PathParams { type Rejection = AppError; async fn from_request_parts( parts: &mut axum::http::request::Parts, state: &Env, ) -> Result { - let Path(params) = Path::>::from_request_parts(parts, state) + let Path(params) = Path::::from_request_parts(parts, state) .await .map_err(|_| { AppError::InternalServerError("path param does not have log field".into()) @@ -178,14 +168,12 @@ type ApiResult = std::result::Result; enum AppError { InternalServerError(String), - NotFound, BadRequest(String), UnknownLog, FailedToSerializeSignaturelessCert(bootstrap_mtc_api::MtcError), SubtreeInclusionProofFailed(tlog_core::TlogError), LeafIndexBeforeFirstActiveLandmark, LeafIndexNotInLog, - RedirectToMonitorApi(&'static str), LeafIndexPendingLandmark { retry_after: u64 }, } @@ -206,7 +194,6 @@ impl IntoResponse for AppError { ) .into_response() } - AppError::NotFound => (StatusCode::NOT_FOUND, "Not Found").into_response(), Self::BadRequest(e) => ( StatusCode::BAD_REQUEST, format!("Bad request{}{e}", if e.is_empty() { "" } else { ": " }), @@ -237,11 +224,6 @@ impl IntoResponse for AppError { "Leaf index will be covered by next landmark", ) .into_response(), - Self::RedirectToMonitorApi(url) => ( - StatusCode::NOT_FOUND, - format!("Use {url} for monitoring API"), - ) - .into_response(), } } } @@ -250,7 +232,7 @@ impl IntoResponse for AppError { #[worker::send] async fn get_roots( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { Ok(( StatusCode::OK, @@ -264,7 +246,7 @@ async fn get_roots( #[worker::send] async fn add_entry( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, body: Bytes, ) -> ApiResult { let params = &CONFIG.logs[&log]; @@ -402,7 +384,7 @@ async fn add_entry( #[worker::send] async fn get_certificate( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, body: Bytes, ) -> ApiResult { let params = &CONFIG.logs[&log]; @@ -484,7 +466,7 @@ async fn get_certificate( #[worker::send] async fn get_landmark_bundle( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { let object_backend = ObjectBucket::new(load_public_bucket(&env, &log)?); @@ -506,7 +488,7 @@ async fn get_landmark_bundle( #[worker::send] async fn metadata( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { let params = &CONFIG.logs[&log]; let cosigner = load_checkpoint_cosigner(&env, &log); @@ -519,11 +501,8 @@ async fn metadata( cosigner_id: cosigner.cosigner_id().to_string(), cosigner_public_key: cosigner.verifying_key(), submission_url: ¶ms.submission_url, - monitoring_url: if params.monitoring_url.is_empty() { - ¶ms.submission_url - } else { - ¶ms.monitoring_url - }, + monitoring_url: (!params.monitoring_url.is_empty()) + .then_some(params.monitoring_url.as_str()), }) .unwrap(), )) @@ -533,7 +512,7 @@ async fn metadata( #[worker::send] async fn sequencer_id( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { // Print out the Durable Object ID of the sequencer to allow looking it up // in internal Cloudflare dashboards. This value does not need to be secret. @@ -542,41 +521,6 @@ async fn sequencer_id( Ok((StatusCode::OK, object_id.to_string())) } -/// `GET /logs/{log}/{*key}` — direct read-through to the public R2 bucket when -/// the log's `monitoring_url` is unspecified. -#[worker::send] -async fn get_object( - State(env): State, - PathParams { - log, - rest: Key { key }, - }: PathParams, -) -> ApiResult { - // Enable direct access to the bucket via the Worker if monitoring_url is - // unspecified. - if CONFIG.logs[&log].monitoring_url.is_empty() { - let bucket = load_public_bucket(&env, &log)?; - if let Some(obj) = bucket.get(key).execute().await? { - let body = obj - .body() - .ok_or_else(|| AppError::InternalServerError("R2 object missing body".into()))? - .stream()?; - Ok(( - StatusCode::OK, - headers_from_http_metadata(obj.http_metadata()), - axum::body::Body::from_stream(WorkerByteStream::new(body)), - )) - } else { - Err(AppError::NotFound) - } - } else { - // TODO: should this be an HTTP redirect instead of a 404? - Err(AppError::RedirectToMonitorApi( - &CONFIG.logs[&log].monitoring_url, - )) - } -} - /// Builds the issuer RDN with the trust anchor ID. fn build_issuer_rdn(log_id: &str) -> std::result::Result { let utf8_value = Utf8StringRef::new(log_id).map_err(|e| e.to_string())?; @@ -676,20 +620,6 @@ async fn get_landmark_sequence( Ok(landmark_sequence) } -fn headers_from_http_metadata(meta: HttpMetadata) -> HeaderMap { - let mut h = HeaderMap::new(); - if let Some(hdr) = meta.cache_control { - h.append("Cache-Control", hdr.try_into().unwrap()); - } - if let Some(hdr) = meta.content_encoding { - h.append("Content-Encoding", hdr.try_into().unwrap()); - } - if let Some(hdr) = meta.content_type { - h.append("Content-Type", hdr.try_into().unwrap()); - } - h -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/ct_worker/config.schema.json b/crates/ct_worker/config.schema.json index 6626f307..6ee07a39 100644 --- a/crates/ct_worker/config.schema.json +++ b/crates/ct_worker/config.schema.json @@ -41,8 +41,7 @@ }, "monitoring_url": { "type": "string", - "default": "", - "description": "URL for log monitoring. If unspecified, use the submission URL and the Worker will proxy requests to the R2 bucket." + "description": "URL for log monitoring. Omit when no monitoring service is configured." }, "temporal_interval": { "type": "object", diff --git a/crates/ct_worker/src/frontend_worker.rs b/crates/ct_worker/src/frontend_worker.rs index d1ca065a..88c9cb67 100644 --- a/crates/ct_worker/src/frontend_worker.rs +++ b/crates/ct_worker/src/frontend_worker.rs @@ -11,7 +11,6 @@ use generic_log_worker::{ get_cached_metadata, get_durable_object_stub, init_logging, load_cache_kv, load_public_bucket, obs::{Wshim, metrics}, put_cache_entry_metadata, serialize, - util::WorkerByteStream, }; use p256::pkcs8::EncodePublicKey; use serde::Serialize; @@ -26,7 +25,7 @@ use axum::{ Json, Router, body::Bytes, extract::{Path, State}, - http::{HeaderMap, StatusCode, header}, + http::{StatusCode, header}, middleware, response::IntoResponse, routing::{get, post}, @@ -54,7 +53,8 @@ struct LogV3JsonResponse<'a> { key: &'a [u8], mmd: usize, submission_url: &'a str, - monitoring_url: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + monitoring_url: Option<&'a str>, temporal_interval: &'a TemporalInterval, } @@ -86,7 +86,6 @@ async fn main( .route("/logs/{log}/ct/v1/add-pre-chain", post(add_pre_chain)) .route("/logs/{log}/log.v3.json", get(log_v3_json)) .route("/logs/{log}/sequencer_id", get(sequencer_id)) - .route("/logs/{log}/{*key}", get(get_object)) .layer(middleware::from_fn_with_state( (env.clone(), metrics::FrontendWorkerMetrics::new(®istry)), request_metrics, @@ -111,28 +110,18 @@ async fn main( } #[derive(serde::Deserialize)] -struct PathParams { +struct PathParams { log: String, - #[serde(flatten)] - rest: Rest, } -#[derive(serde::Deserialize)] -struct Key { - key: String, -} - -impl axum::extract::FromRequestParts for PathParams -where - T: serde::de::DeserializeOwned + Send, -{ +impl axum::extract::FromRequestParts for PathParams { type Rejection = AppError; async fn from_request_parts( parts: &mut axum::http::request::Parts, state: &Env, ) -> Result { - let Path(params) = Path::>::from_request_parts(parts, state) + let Path(params) = Path::::from_request_parts(parts, state) .await .map_err(|_| { AppError::InternalServerError("path param does not have log field".into()) @@ -152,11 +141,9 @@ type ApiResult = std::result::Result; enum AppError { InternalServerError(String), - NotFound, BadRequest(String), UnknownLog, ReadonlyLog, - RedirectToMonitorApi(&'static str), } impl From for AppError { @@ -188,7 +175,6 @@ impl IntoResponse for AppError { ) .into_response() } - AppError::NotFound => (StatusCode::NOT_FOUND, "Not Found").into_response(), Self::BadRequest(e) => { ( StatusCode::BAD_REQUEST, @@ -205,11 +191,6 @@ impl IntoResponse for AppError { "The log is temporarily in read-only mode during maintenance. Please try again after 5 minutes." ).into_response() } - Self::RedirectToMonitorApi(url) => ( - StatusCode::NOT_FOUND, - format!("Use {url} for monitoring API"), - ) - .into_response(), } } } @@ -218,7 +199,7 @@ impl IntoResponse for AppError { #[worker::send] async fn get_roots( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { Ok(( StatusCode::OK, @@ -232,7 +213,7 @@ async fn get_roots( #[worker::send] async fn add_chain( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, body: Bytes, ) -> ApiResult { add_chain_or_pre_chain(body, &env, &log, false).await @@ -242,7 +223,7 @@ async fn add_chain( #[worker::send] async fn add_pre_chain( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, body: Bytes, ) -> ApiResult { add_chain_or_pre_chain(body, &env, &log, true).await @@ -252,7 +233,7 @@ async fn add_pre_chain( #[worker::send] async fn log_v3_json( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { let params = &CONFIG.logs[&log]; let verifying_key = load_signing_key(&env, &log)?.verifying_key(); @@ -269,11 +250,8 @@ async fn log_v3_json( log_id, key: key.as_bytes(), submission_url: ¶ms.submission_url, - monitoring_url: if params.monitoring_url.is_empty() { - ¶ms.submission_url - } else { - ¶ms.monitoring_url - }, + monitoring_url: (!params.monitoring_url.is_empty()) + .then_some(params.monitoring_url.as_str()), mmd: MAX_MERGE_DELAY_SECS, temporal_interval: ¶ms.temporal_interval, }) @@ -286,7 +264,7 @@ async fn log_v3_json( #[worker::send] async fn sequencer_id( State(env): State, - PathParams { log, .. }: PathParams<()>, + PathParams { log }: PathParams, ) -> ApiResult { // Print out the Durable Object ID of the sequencer to allow looking it up // in internal Cloudflare dashboards. This value does not need to be secret. @@ -295,41 +273,6 @@ async fn sequencer_id( Ok((StatusCode::OK, object_id.to_string())) } -/// `GET /logs/{log}/{*key}` — direct read-through to the public R2 bucket when -/// the log's `monitoring_url` is unspecified. -#[worker::send] -async fn get_object( - State(env): State, - PathParams { - log, - rest: Key { key }, - }: PathParams, -) -> ApiResult { - // Enable direct access to the bucket via the Worker if monitoring_url is - // unspecified. - if CONFIG.logs[&log].monitoring_url.is_empty() { - let bucket = load_public_bucket(&env, &log)?; - if let Some(obj) = bucket.get(key).execute().await? { - let body = obj - .body() - .ok_or_else(|| AppError::InternalServerError("R2 object missing body".into()))? - .stream()?; - Ok(( - StatusCode::OK, - headers_from_http_metadata(obj.http_metadata()), - axum::body::Body::from_stream(WorkerByteStream::new(body)), - )) - } else { - Err(AppError::NotFound) - } - } else { - // TODO: should this be an HTTP redirect instead of a 404? - Err(AppError::RedirectToMonitorApi( - &CONFIG.logs[&log].monitoring_url, - )) - } -} - #[allow(clippy::too_many_lines)] async fn add_chain_or_pre_chain( body: Bytes, @@ -469,17 +412,3 @@ async fn add_chain_or_pre_chain( .map_err(|e| e.to_string())?; Ok((StatusCode::OK, Json(sct)).into_response()) } - -fn headers_from_http_metadata(meta: HttpMetadata) -> HeaderMap { - let mut h = HeaderMap::new(); - if let Some(hdr) = meta.cache_control { - h.append("Cache-Control", hdr.try_into().unwrap()); - } - if let Some(hdr) = meta.content_encoding { - h.append("Content-Encoding", hdr.try_into().unwrap()); - } - if let Some(hdr) = meta.content_type { - h.append("Content-Type", hdr.try_into().unwrap()); - } - h -} diff --git a/crates/integration_tests/src/assertions.rs b/crates/integration_tests/src/assertions.rs index 06739368..3ea1b73b 100644 --- a/crates/integration_tests/src/assertions.rs +++ b/crates/integration_tests/src/assertions.rs @@ -203,13 +203,19 @@ pub async fn fetch_and_verify_checkpoint( client: &CtClient, log_meta: &LogV3JsonResponse, witness_key_der: Option<&[u8]>, - now_millis: u64, ) -> Result { let checkpoint_bytes = client .get_checkpoint() .await .context("fetching checkpoint")?; + let now_millis = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .context("system time is before Unix epoch")? + .as_millis() + .try_into() + .context("system time does not fit in u64 milliseconds")?; + verify_checkpoint_bytes( &checkpoint_bytes, &client.log, @@ -436,14 +442,13 @@ pub async fn fetch_checkpoint_until_size( client: &CtClient, log_meta: &LogV3JsonResponse, min_size: u64, - now_millis: u64, ) -> Result { const MAX_RETRIES: u32 = 12; const RETRY_DELAY_MS: u64 = 500; let mut last_err = None; for attempt in 0..MAX_RETRIES { - match fetch_and_verify_checkpoint(client, log_meta, None, now_millis).await { + match fetch_and_verify_checkpoint(client, log_meta, None).await { Ok(cp) if cp.text.size() >= min_size => return Ok(cp), Ok(cp) => { last_err = Some(anyhow::anyhow!( diff --git a/crates/integration_tests/src/client.rs b/crates/integration_tests/src/client.rs index 9a509254..eed9816a 100644 --- a/crates/integration_tests/src/client.rs +++ b/crates/integration_tests/src/client.rs @@ -11,6 +11,8 @@ use base64::prelude::*; use serde::{Deserialize, Serialize}; use serde_with::{base64::Base64, serde_as}; +use crate::local_r2; + // --------------------------------------------------------------------------- // Configuration // --------------------------------------------------------------------------- @@ -57,7 +59,7 @@ pub struct LogV3JsonResponse { pub key: Vec, pub mmd: u64, pub submission_url: String, - pub monitoring_url: String, + pub monitoring_url: Option, pub temporal_interval: TemporalInterval, } @@ -192,22 +194,16 @@ impl CtClient { self.get_raw("checkpoint").await } - /// `GET /logs/:log/{path}` — raw bytes (tiles, checkpoint, etc.) + /// Reads raw log data (tiles, checkpoint, etc.). pub async fn get_raw(&self, path: &str) -> Result> { - let resp = self - .client - .get(self.url(path)) - .send() - .await - .with_context(|| format!("GET {path}"))?; - let status = resp.status(); - if !status.is_success() { - bail!("GET {path} returned {status}"); + if local_r2::is_loopback_base_url(&base_url()) { + return local_r2::get("ct_worker", &format!("static-ct-public-{}", self.log), path) + .await? + .with_context(|| format!("R2 object missing: {path}")); } - resp.bytes() - .await - .map(|b| b.to_vec()) - .with_context(|| format!("reading body for {path}")) + + let metadata = self.get_log_v3_json().await?; + get_raw_http(&self.client, metadata.monitoring_url.as_deref(), path).await } /// `GET /logs/:log/{path}` — returns the HTTP status code (does not fail on 4xx/5xx). @@ -243,7 +239,7 @@ pub struct BootstrapMtcMetadataResponse { #[serde_as(as = "Base64")] pub cosigner_public_key: Vec, pub submission_url: String, - pub monitoring_url: String, + pub monitoring_url: Option, } /// Response body from `POST /logs/:log/add-entry`. @@ -378,27 +374,25 @@ impl BootstrapMtcClient { } } - /// `GET /logs/:log/checkpoint` — raw bytes. + /// Reads the checkpoint from local R2 state. pub async fn get_checkpoint(&self) -> Result> { self.get_raw("checkpoint").await } - /// `GET /logs/:log/{path}` — raw bytes. + /// Reads raw log data. pub async fn get_raw(&self, path: &str) -> Result> { - let resp = self - .client - .get(self.url(path)) - .send() - .await - .with_context(|| format!("GET {path}"))?; - let status = resp.status(); - if !status.is_success() { - bail!("GET {path} returned {status}"); + if local_r2::is_loopback_base_url(&base_url()) { + return local_r2::get( + "bootstrap_mtc_worker", + &format!("mtc-public-{}", self.log), + path, + ) + .await? + .with_context(|| format!("R2 object missing: {path}")); } - resp.bytes() - .await - .map(|b| b.to_vec()) - .with_context(|| format!("reading body for {path}")) + + let metadata = self.get_metadata().await?; + get_raw_http(&self.client, metadata.monitoring_url.as_deref(), path).await } /// `GET /logs/:log/{path}` — returns the HTTP status code without failing on 4xx/5xx. @@ -413,6 +407,28 @@ impl BootstrapMtcClient { } } +async fn get_raw_http( + client: &reqwest::Client, + monitoring_url: Option<&str>, + path: &str, +) -> Result> { + let monitoring_url = monitoring_url.context("log does not advertise a monitoring URL")?; + let url = format!("{}/{path}", monitoring_url.trim_end_matches('/')); + let resp = client + .get(&url) + .send() + .await + .with_context(|| format!("GET {url}"))?; + let status = resp.status(); + if !status.is_success() { + bail!("GET {url} returned {status}"); + } + resp.bytes() + .await + .map(|bytes| bytes.to_vec()) + .with_context(|| format!("reading body for {url}")) +} + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- diff --git a/crates/integration_tests/src/lib.rs b/crates/integration_tests/src/lib.rs index 09082708..1137cff1 100644 --- a/crates/integration_tests/src/lib.rs +++ b/crates/integration_tests/src/lib.rs @@ -12,3 +12,4 @@ pub mod assertions; pub mod client; pub mod fixtures; +pub mod local_r2; diff --git a/crates/integration_tests/src/local_r2.rs b/crates/integration_tests/src/local_r2.rs new file mode 100644 index 00000000..ac6a0bdc --- /dev/null +++ b/crates/integration_tests/src/local_r2.rs @@ -0,0 +1,51 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. +// Licensed under the BSD-3-Clause license found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause + +use anyhow::{Context, Result, bail}; +use std::{path::PathBuf, time::Duration}; + +#[must_use] +pub fn is_loopback_base_url(base_url: &str) -> bool { + base_url.starts_with("http://localhost:") || base_url.starts_with("http://127.0.0.1:") +} + +/// Reads an object from a Worker's persisted local R2 state. +/// +/// # Errors +/// +/// Returns an error for non-loopback test URLs or if Wrangler fails. +pub async fn get(worker: &str, bucket: &str, key: &str) -> Result>> { + let base_url = + std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:8787".to_string()); + if !is_loopback_base_url(&base_url) { + bail!("local R2 inspection requires a loopback BASE_URL"); + } + + let worker_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(format!("../{worker}")); + let object = format!("{bucket}/{key}"); + let mut command = tokio::process::Command::new("wrangler"); + command.current_dir(worker_dir).args([ + "r2", + "object", + "get", + &object, + "--local", + "--persist-to", + ".wrangler/state", + "--pipe", + ]); + command.kill_on_drop(true); + let output = tokio::time::timeout(Duration::from_secs(30), command.output()) + .await + .context("wrangler r2 object get timed out")? + .context("running wrangler r2 object get")?; + if output.status.success() { + return Ok(Some(output.stdout)); + } + + let stderr = String::from_utf8_lossy(&output.stderr); + if stderr.contains("specified key does not exist") { + return Ok(None); + } + bail!("wrangler r2 object get failed for {key}: {stderr}"); +} diff --git a/crates/integration_tests/tests/bootstrap_mtc_api.rs b/crates/integration_tests/tests/bootstrap_mtc_api.rs index be8f55f1..a20ba8bd 100644 --- a/crates/integration_tests/tests/bootstrap_mtc_api.rs +++ b/crates/integration_tests/tests/bootstrap_mtc_api.rs @@ -14,8 +14,10 @@ //! # Running //! //! ```text +//! npm install --global wrangler@4.80.0 +//! //! # From crates/bootstrap_mtc_worker/: -//! npx wrangler -e=dev dev & +//! wrangler -e=dev dev --persist-to .wrangler/state & //! //! # From workspace root: //! cargo test -p integration_tests --test bootstrap_mtc_api @@ -24,8 +26,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use integration_tests::{ - client::BootstrapMtcClient, + client::{BootstrapMtcClient, base_url}, fixtures::{garbage_chain, make_bootstrap_mtc_chain}, + local_r2, }; use tokio::sync::OnceCell; use x509_cert::{Certificate, der::Decode}; @@ -168,6 +171,9 @@ async fn metadata_returns_valid_fields() { !meta.submission_url.is_empty(), "submission_url must be non-empty" ); + if local_r2::is_loopback_base_url(&base_url()) { + assert!(meta.monitoring_url.is_none()); + } } /// `POST /logs/:log/add-entry` with a valid bootstrap chain returns 200 with @@ -221,6 +227,14 @@ async fn unknown_log_returns_400() { assert_eq!(status, 400, "expected 400 for unknown log"); } +/// Persisted log objects are not exposed by the submission Worker. +#[tokio::test] +async fn checkpoint_is_not_served_by_frontend() { + ensure_initialized().await; + let client = BootstrapMtcClient::default_log(); + assert_eq!(client.get_status("checkpoint").await.unwrap(), 404); +} + /// After `add-entry`, the checkpoint tree size covers the returned `leaf_index`. #[tokio::test] async fn add_entry_appears_in_checkpoint() { diff --git a/crates/integration_tests/tests/static_ct_api.rs b/crates/integration_tests/tests/static_ct_api.rs index 92dce365..54ec7c0b 100644 --- a/crates/integration_tests/tests/static_ct_api.rs +++ b/crates/integration_tests/tests/static_ct_api.rs @@ -10,8 +10,10 @@ //! # Running //! //! ```text +//! npm install --global wrangler@4.80.0 +//! //! # From crates/ct_worker/: -//! npx wrangler -e=dev dev & +//! wrangler -e=dev dev --persist-to .wrangler/state & //! //! # From workspace root: //! cargo test -p integration_tests --test static_ct_api @@ -22,31 +24,19 @@ //! BASE_URL=http://localhost:8787 LOG_NAME=dev2026h1a cargo test -p integration_tests --test static_ct_api //! ``` -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use integration_tests::{ assertions::{ assert_leaf_in_checkpoint, assert_sct_signature, assert_sct_structure, fetch_and_verify_checkpoint, fetch_checkpoint_until_size, leaf_index_from_sct, }, - client::CtClient, + client::{CtClient, base_url}, fixtures::{empty_chain, garbage_chain, make_chains}, + local_r2, }; use tokio::sync::OnceCell; -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -fn now_millis() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() - .try_into() - .unwrap() -} - // --------------------------------------------------------------------------- // Initialization guard // --------------------------------------------------------------------------- @@ -96,13 +86,7 @@ async fn ensure_initialized() { // covers our warmup entry before releasing the gate. let leaf_index = leaf_index_from_sct(&sct).expect("leaf_index from warmup SCT"); - let _ = fetch_checkpoint_until_size( - &client, - &meta, - leaf_index + 1, - now_millis(), - ) - .await; + let _ = fetch_checkpoint_until_size(&client, &meta, leaf_index + 1).await; return; } Ok((status, _)) => { @@ -168,6 +152,9 @@ async fn log_v3_json_returns_valid_metadata() { !meta.submission_url.is_empty(), "submission_url must be set" ); + if local_r2::is_loopback_base_url(&base_url()) { + assert!(meta.monitoring_url.is_none()); + } // Key must be a valid P-256 SPKI. p256::ecdsa::VerifyingKey::from_public_key_der(&meta.key) @@ -195,6 +182,14 @@ async fn unknown_log_returns_400() { assert_eq!(status, 400, "expected 400 for unknown log"); } +/// Persisted log objects are not exposed by the submission Worker. +#[tokio::test] +async fn checkpoint_is_not_served_by_frontend() { + ensure_initialized().await; + let client = CtClient::default_log(); + assert_eq!(client.get_status("checkpoint").await.unwrap(), 404); +} + /// `POST` with a JSON body that is not a valid DER certificate returns 400. #[tokio::test] async fn add_chain_with_garbage_cert_returns_400() { @@ -307,7 +302,7 @@ async fn add_chain_sct_appears_in_checkpoint() { let leaf_index = leaf_index_from_sct(&sct).expect("extracting leaf_index"); let min_size = leaf_index + 1; - let checkpoint = fetch_checkpoint_until_size(&client, &meta, min_size, now_millis()) + let checkpoint = fetch_checkpoint_until_size(&client, &meta, min_size) .await .expect("waiting for checkpoint"); @@ -337,7 +332,7 @@ async fn add_chain_leaf_verifiable_in_tree() { let leaf_index = leaf_index_from_sct(&sct).expect("extracting leaf_index"); - let checkpoint = fetch_checkpoint_until_size(&client, &meta, leaf_index + 1, now_millis()) + let checkpoint = fetch_checkpoint_until_size(&client, &meta, leaf_index + 1) .await .expect("waiting for checkpoint"); @@ -364,7 +359,7 @@ async fn add_pre_chain_leaf_verifiable_in_tree() { let leaf_index = leaf_index_from_sct(&sct).expect("extracting leaf_index"); - let checkpoint = fetch_checkpoint_until_size(&client, &meta, leaf_index + 1, now_millis()) + let checkpoint = fetch_checkpoint_until_size(&client, &meta, leaf_index + 1) .await .expect("waiting for checkpoint"); @@ -383,7 +378,7 @@ async fn checkpoint_signature_is_valid() { let client = CtClient::default_log(); let meta = client.get_log_v3_json().await.expect("log.v3.json"); - fetch_and_verify_checkpoint(&client, &meta, None, now_millis()) + fetch_and_verify_checkpoint(&client, &meta, None) .await .expect("checkpoint signature verification"); } diff --git a/crates/integration_tests/tests/tlog_mirror.rs b/crates/integration_tests/tests/tlog_mirror.rs index 9f55d169..34184442 100644 --- a/crates/integration_tests/tests/tlog_mirror.rs +++ b/crates/integration_tests/tests/tlog_mirror.rs @@ -31,7 +31,6 @@ use serde_with::{base64::Base64, serde_as}; use sha2::{Digest as _, Sha256}; use signed_note::{KeyName, Note, NoteSignature, VerifierList}; use std::collections::HashMap; -use std::path::PathBuf; use std::time::Duration; use tlog_checkpoint::{CheckpointSigner, TreeWithTimestamp}; use tlog_core::{ @@ -209,46 +208,15 @@ fn http_client() -> reqwest::Client { .expect("build HTTP client") } -fn mirror_worker_dir() -> PathBuf { - PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../mirror_worker") -} - fn r2_key(path: &str) -> String { let origin_hash = hex::encode(Sha256::digest(LOG_ORIGIN.as_bytes())); - format!("{R2_BUCKET}/{origin_hash}/{path}") + format!("{origin_hash}/{path}") } async fn local_r2_object(path: &str) -> Option> { - let url = base_url(); - assert!( - url.starts_with("http://localhost:") || url.starts_with("http://127.0.0.1:"), - "local R2 inspection requires a loopback BASE_URL", - ); - let command = tokio::process::Command::new("wrangler") - .current_dir(mirror_worker_dir()) - .args([ - "r2", - "object", - "get", - &r2_key(path), - "--local", - "--persist-to", - ".wrangler/state", - "--pipe", - ]) - .output(); - let output = tokio::time::timeout(Duration::from_secs(30), command) + integration_tests::local_r2::get("mirror_worker", R2_BUCKET, &r2_key(path)) .await - .expect("wrangler r2 object get timed out") - .expect("run wrangler r2 object get"); - if output.status.success() { - return Some(output.stdout); - } - let stderr = String::from_utf8_lossy(&output.stderr); - if stderr.contains("specified key does not exist") { - return None; - } - panic!("wrangler r2 object get failed for {path}: {stderr}"); + .unwrap_or_else(|e| panic!("local R2 read failed for {path}: {e}")) } async fn require_local_r2_object(path: &str) -> Vec {