witness.rs 79.8%

170 of 213 lines run, 18 of 40 functions · run #338 on proposal/125 · run missed

//! `oxy witness serve` (Oxygen spec §14.8, oxygenhub docs/transparency.md):
//! a small transparency-log witness anyone can run. It cosigns the
//! checkpoints of the logs it is told about (C2SP tlog-witness), each only
//! when it extends the last one this witness cosigned, with its own key
//! and clock, so a log cannot show two histories without the witness
//! seeing it.
//!
//! State is one file per log in the state directory: the size and root of
//! the last checkpoint cosigned, replaced atomically.

use axum::extract::State;
use axum::http::{header, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::post;
use axum::Router;
use oxy_core::identity::{Identity, SecretKey};
use oxy_core::tlog::{self, AddCheckpoint, Checkpoint, Note, WitnessError, WitnessKey, Witnessed};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

/// What a witness serves.
pub struct WitnessConfig {
    /// Its name in cosignatures (a host name, no spaces).
    pub name: String,
    pub identity: Identity,
    pub sk: SecretKey,
    /// The logs it witnesses: their origins and keys.
    pub logs: Vec<(String, Identity)>,
    /// Where it keeps what it cosigned.
    pub state: PathBuf,
    pub tls: Option<crate::TlsFiles>,
}

struct Witness {
    cfg: WitnessConfig,
    /// One add-checkpoint at a time: each reads and moves its log's state.
    lock: Mutex<()>,
}

fn state_file(w: &Witness, origin: &str) -> PathBuf {
    let safe: String = origin
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '-' {
                c
            } else {
                '_'
            }
        })
        .collect();
    w.cfg.state.join(safe)
}

fn read_state(w: &Witness, origin: &str) -> Option<(u64, tlog::Hash)> {
    let text = std::fs::read_to_string(state_file(w, origin)).ok()?;
    let (size, root) = text.trim().split_once(' ')?;
    let root = oxy_core::hash::ObjectRef::from_hex(root)?;
    Some((size.parse().ok()?, root.0))
}

fn plain(status: StatusCode, why: &str) -> Response {
    (
        status,
        [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
        format!("{why}\n"),
    )
        .into_response()
}

async fn add_checkpoint(State(w): State<Arc<Witness>>, body: String) -> Response {
    let Ok(req) = AddCheckpoint::parse(&body) else {
        return plain(StatusCode::BAD_REQUEST, "not an add-checkpoint request");
    };
    let Some(origin) = Note::parse(&req.note)
        .ok()
        .and_then(|n| Checkpoint::parse(&n.text).ok())
        .map(|c| c.origin)
    else {
        return plain(StatusCode::BAD_REQUEST, "the checkpoint does not parse");
    };
    let Some(key) = w
        .cfg
        .logs
        .iter()
        .find(|(o, _)| *o == origin)
        .map(|(_, k)| k.clone())
    else {
        return plain(
            StatusCode::NOT_FOUND,
            &format!("this witness does not witness {origin}"),
        );
    };
    let _held = w.lock.lock().unwrap_or_else(|p| p.into_inner());
    let latest = read_state(&w, &origin);
    let me = WitnessKey {
        name: &w.cfg.name,
        identity: &w.cfg.identity,
        sk: &w.cfg.sk,
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let conflict = |n: u64| {
        (
            StatusCode::CONFLICT,
            [(header::CONTENT_TYPE, "text/x.tlog.size")],
            format!("{n}\n"),
        )
            .into_response()
    };
    match tlog::witness(&req, &origin, &key, latest, &me, now) {
        Ok(Witnessed::Cosigned { checkpoint, lines }) => {
            let text = format!(
                "{} {}\n",
                checkpoint.size,
                oxy_core::hash::ObjectRef(checkpoint.root).to_hex()
            );
            if let Err(e) = std::fs::create_dir_all(&w.cfg.state).and_then(|_| {
                oxy_core::durable::write_atomic(&state_file(&w, &origin), text.as_bytes())
            }) {
                return plain(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    &format!("keeping the witness's state: {e}"),
                );
            }
            let body = Note {
                text: String::new(),
                sigs: lines,
            }
            .encode();
            (
                [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
                body.trim_start_matches('\n').to_string(),
            )
                .into_response()
        }
        Ok(Witnessed::Conflict(n)) => conflict(n),
        Err(WitnessError::Unsigned(e)) => plain(StatusCode::FORBIDDEN, &e),
        Err(WitnessError::Inconsistent(e)) => plain(StatusCode::UNPROCESSABLE_ENTITY, &e),
        Err(WitnessError::Bad(e)) => plain(StatusCode::BAD_REQUEST, &e),
    }
}

/// The witness's routes: `POST /add-checkpoint`.
pub fn router(cfg: WitnessConfig) -> Router {
    Router::new()
        .route("/add-checkpoint", post(add_checkpoint))
        .with_state(Arc::new(Witness {
            cfg,
            lock: Mutex::new(()),
        }))
}

/// Serve a witness on `listener` until stopped.
pub fn run_blocking(
    listener: std::net::TcpListener,
    mut cfg: WitnessConfig,
) -> std::io::Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;
    let tls = cfg.tls.take();
    rt.block_on(async move {
        listener.set_nonblocking(true)?;
        match tls {
            Some(tls) => {
                let _ = rustls::crypto::ring::default_provider().install_default();
                let rustls_cfg =
                    axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
                        .await
                        .map_err(|e| std::io::Error::other(format!("loading TLS files: {e}")))?;
                axum_server::from_tcp_rustls(listener, rustls_cfg)?
                    .serve(router(cfg).into_make_service())
                    .await
            }
            None => axum::serve(tokio::net::TcpListener::from_std(listener)?, router(cfg)).await,
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request;
    use tower::ServiceExt;

    #[tokio::test]
    async fn the_witness_cosigns_and_keeps_what_it_cosigned() {
        let dir = tempfile::tempdir().unwrap();
        let log_sk = SecretKey::generate();
        let log = log_sk.identity("log", "[email protected]", 1);
        let w_sk = SecretKey::generate();
        let w = w_sk.identity("witness", "[email protected]", 1);
        let cfg = || WitnessConfig {
            name: "witness.example".into(),
            identity: w.clone(),
            sk: w_sk.clone(),
            logs: vec![("hub.example/log".into(), log.clone())],
            state: dir.path().join("state"),
            tls: None,
        };
        let mut stored: Vec<tlog::Hash> = Vec::new();
        for n in 0..9u64 {
            stored.extend(
                tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&[n as u8]), &stored).unwrap(),
            );
        }
        let note = |n: u64| {
            Checkpoint {
                origin: "hub.example/log".into(),
                size: n,
                root: tlog::tree_hash(n, &stored).unwrap(),
                extensions: vec![],
            }
            .sign(&log, &log_sk)
            .unwrap()
            .encode()
        };
        let post = |body: String| async {
            let res = router(cfg())
                .oneshot(
                    Request::post("/add-checkpoint")
                        .body(Body::from(body))
                        .unwrap(),
                )
                .await
                .unwrap();
            let st = res.status();
            let b = axum::body::to_bytes(res.into_body(), 1 << 20)
                .await
                .unwrap();
            (st, String::from_utf8(b.to_vec()).unwrap())
        };
        let (st, lines) = post(
            AddCheckpoint {
                old_size: 0,
                proof: vec![],
                note: note(4),
            }
            .encode(),
        )
        .await;
        assert_eq!(st, StatusCode::OK, "{lines}");
        let cosigned =
            Note::parse(&format!("{}\n{lines}", Note::parse(&note(4)).unwrap().text)).unwrap();
        assert!(cosigned.cosigned_by("witness.example", &w).is_ok());
        // Kept across a restart (a new router): a wrong old size is told.
        let (st, size) = post(
            AddCheckpoint {
                old_size: 0,
                proof: vec![],
                note: note(9),
            }
            .encode(),
        )
        .await;
        assert_eq!((st, size.trim()), (StatusCode::CONFLICT, "4"));
        let proof = tlog::prove_consistency(9, 4, &stored).unwrap();
        let (st, _) = post(
            AddCheckpoint {
                old_size: 4,
                proof,
                note: note(9),
            }
            .encode(),
        )
        .await;
        assert_eq!(st, StatusCode::OK);
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! `oxy witness serve` (Oxygen spec §14.8, oxygenhub docs/transparency.md):
2		//! a small transparency-log witness anyone can run. It cosigns the
3		//! checkpoints of the logs it is told about (C2SP tlog-witness), each only
4		//! when it extends the last one this witness cosigned, with its own key
5		//! and clock, so a log cannot show two histories without the witness
6		//! seeing it.
7		//!
8		//! State is one file per log in the state directory: the size and root of
9		//! the last checkpoint cosigned, replaced atomically.
10
11		use axum::extract::State;
12		use axum::http::{header, StatusCode};
13		use axum::response::{IntoResponse, Response};
14		use axum::routing::post;
15		use axum::Router;
16		use oxy_core::identity::{Identity, SecretKey};
17		use oxy_core::tlog::{self, AddCheckpoint, Checkpoint, Note, WitnessError, WitnessKey, Witnessed};
18		use std::path::PathBuf;
19		use std::sync::{Arc, Mutex};
20
21		/// What a witness serves.
22		pub struct WitnessConfig {
23		/// Its name in cosignatures (a host name, no spaces).
24		pub name: String,
25		pub identity: Identity,
26		pub sk: SecretKey,
27		/// The logs it witnesses: their origins and keys.
28		pub logs: Vec<(String, Identity)>,
29		/// Where it keeps what it cosigned.
30		pub state: PathBuf,
31		pub tls: Option<crate::TlsFiles>,
32		}
33
34		struct Witness {
35		cfg: WitnessConfig,
36		/// One add-checkpoint at a time: each reads and moves its log's state.
37		lock: Mutex<()>,
38		}
39
40	5×	fn state_file(w: &Witness, origin: &str) -> PathBuf {
41	5×	let safe: String = origin
42	5×	.chars()
43	75×	.map(\|c\| {
44	75×	if c.is_ascii_alphanumeric() \|\| c == '.' \|\| c == '-' {
45	70×	c
46		} else {
47	5×	'_'
48		}
49	75×	})
50	5×	.collect();
51	5×	w.cfg.state.join(safe)
52	5×	}
53
54	3×	fn read_state(w: &Witness, origin: &str) -> Option<(u64, tlog::Hash)> {
55	3×	let text = std::fs::read_to_string(state_file(w, origin)).ok()?;
56	2×	let (size, root) = text.trim().split_once(' ')?;
57	2×	let root = oxy_core::hash::ObjectRef::from_hex(root)?;
58	2×	Some((size.parse().ok()?, root.0))
59	3×	}
60
61	0×	fn plain(status: StatusCode, why: &str) -> Response {
62	0×	(
63	0×	status,
64	0×	[(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
65	0×	format!("{why}\n"),
66	0×	)
67	0×	.into_response()
68	0×	}
69
70	3×	async fn add_checkpoint(State(w): State<Arc<Witness>>, body: String) -> Response {
71	3×	let Ok(req) = AddCheckpoint::parse(&body) else {
72	0×	return plain(StatusCode::BAD_REQUEST, "not an add-checkpoint request");
73		};
74	3×	let Some(origin) = Note::parse(&req.note)
75	3×	.ok()
76	3×	.and_then(\|n\| Checkpoint::parse(&n.text).ok())
77	3×	.map(\|c\| c.origin)
78		else {
79	0×	return plain(StatusCode::BAD_REQUEST, "the checkpoint does not parse");
80		};
81	3×	let Some(key) = w
82	3×	.cfg
83	3×	.logs
84	3×	.iter()
85	3×	.find(\|(o, _)\| *o == origin)
86	3×	.map(\|(_, k)\| k.clone())
87		else {
88	0×	return plain(
89		StatusCode::NOT_FOUND,
90	0×	&format!("this witness does not witness {origin}"),
91		);
92		};
93	3×	let _held = w.lock.lock().unwrap_or_else(\|p\| p.into_inner());
94	3×	let latest = read_state(&w, &origin);
95	3×	let me = WitnessKey {
96	3×	name: &w.cfg.name,
97	3×	identity: &w.cfg.identity,
98	3×	sk: &w.cfg.sk,
99	3×	};
100	3×	let now = std::time::SystemTime::now()
101	3×	.duration_since(std::time::UNIX_EPOCH)
102	3×	.map(\|d\| d.as_secs())
103	3×	.unwrap_or(0);
104	3×	let conflict = \|n: u64\| {
105	1×	(
106	1×	StatusCode::CONFLICT,
107	1×	[(header::CONTENT_TYPE, "text/x.tlog.size")],
108	1×	format!("{n}\n"),
109	1×	)
110	1×	.into_response()
111	1×	};
112	3×	match tlog::witness(&req, &origin, &key, latest, &me, now) {
113	2×	Ok(Witnessed::Cosigned { checkpoint, lines }) => {
114	2×	let text = format!(
115		"{} {}\n",
116		checkpoint.size,
117	2×	oxy_core::hash::ObjectRef(checkpoint.root).to_hex()
118		);
119	2×	if let Err(e) = std::fs::create_dir_all(&w.cfg.state).and_then(\|_\| {
120	2×	oxy_core::durable::write_atomic(&state_file(&w, &origin), text.as_bytes())
121	2×	}) {
122	0×	return plain(
123		StatusCode::INTERNAL_SERVER_ERROR,
124	0×	&format!("keeping the witness's state: {e}"),
125		);
126	2×	}
127	2×	let body = Note {
128	2×	text: String::new(),
129	2×	sigs: lines,
130	2×	}
131	2×	.encode();
132	2×	(
133	2×	[(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
134	2×	body.trim_start_matches('\n').to_string(),
135	2×	)
136	2×	.into_response()
137		}
138	1×	Ok(Witnessed::Conflict(n)) => conflict(n),
139	0×	Err(WitnessError::Unsigned(e)) => plain(StatusCode::FORBIDDEN, &e),
140	0×	Err(WitnessError::Inconsistent(e)) => plain(StatusCode::UNPROCESSABLE_ENTITY, &e),
141	0×	Err(WitnessError::Bad(e)) => plain(StatusCode::BAD_REQUEST, &e),
142		}
143	3×	}
144
145		/// The witness's routes: `POST /add-checkpoint`.
146	3×	pub fn router(cfg: WitnessConfig) -> Router {
147	3×	Router::new()
148	3×	.route("/add-checkpoint", post(add_checkpoint))
149	3×	.with_state(Arc::new(Witness {
150	3×	cfg,
151	3×	lock: Mutex::new(()),
152	3×	}))
153	3×	}
154
155		/// Serve a witness on `listener` until stopped.
156	0×	pub fn run_blocking(
157	0×	listener: std::net::TcpListener,
158	0×	mut cfg: WitnessConfig,
159	0×	) -> std::io::Result<()> {
160	0×	let rt = tokio::runtime::Builder::new_multi_thread()
161	0×	.enable_all()
162	0×	.build()?;
163	0×	let tls = cfg.tls.take();
164	0×	rt.block_on(async move {
165	0×	listener.set_nonblocking(true)?;
166	0×	match tls {
167	0×	Some(tls) => {
168	0×	let _ = rustls::crypto::ring::default_provider().install_default();
169	0×	let rustls_cfg =
170	0×	axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
171	0×	.await
172	0×	.map_err(\|e\| std::io::Error::other(format!("loading TLS files: {e}")))?;
173	0×	axum_server::from_tcp_rustls(listener, rustls_cfg)?
174	0×	.serve(router(cfg).into_make_service())
175	0×	.await
176		}
177	0×	None => axum::serve(tokio::net::TcpListener::from_std(listener)?, router(cfg)).await,
178		}
179	0×	})
180	0×	}
181
182		#[cfg(test)]
183		mod tests {
184		use super::*;
185		use axum::body::Body;
186		use axum::http::Request;
187		use tower::ServiceExt;
188
189		#[tokio::test]
190	1×	async fn the_witness_cosigns_and_keeps_what_it_cosigned() {
191	1×	let dir = tempfile::tempdir().unwrap();
192	1×	let log_sk = SecretKey::generate();
193	1×	let log = log_sk.identity("log", "[email protected]", 1);
194	1×	let w_sk = SecretKey::generate();
195	1×	let w = w_sk.identity("witness", "[email protected]", 1);
196	1×	let cfg = \|\| WitnessConfig {
197	3×	name: "witness.example".into(),
198	3×	identity: w.clone(),
199	3×	sk: w_sk.clone(),
200	3×	logs: vec![("hub.example/log".into(), log.clone())],
201	3×	state: dir.path().join("state"),
202	3×	tls: None,
203	3×	};
204	1×	let mut stored: Vec<tlog::Hash> = Vec::new();
205	9×	for n in 0..9u64 {
206	9×	stored.extend(
207	9×	tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&[n as u8]), &stored).unwrap(),
208	9×	);
209	9×	}
210	4×	let note = \|n: u64\| {
211	4×	Checkpoint {
212	4×	origin: "hub.example/log".into(),
213	4×	size: n,
214	4×	root: tlog::tree_hash(n, &stored).unwrap(),
215	4×	extensions: vec![],
216	4×	}
217	4×	.sign(&log, &log_sk)
218	4×	.unwrap()
219	4×	.encode()
220	4×	};
221	3×	let post = \|body: String\| async {
222	3×	let res = router(cfg())
223	3×	.oneshot(
224	3×	Request::post("/add-checkpoint")
225	3×	.body(Body::from(body))
226	3×	.unwrap(),
227	3×	)
228	3×	.await
229	3×	.unwrap();
230	3×	let st = res.status();
231	3×	let b = axum::body::to_bytes(res.into_body(), 1 << 20)
232	3×	.await
233	3×	.unwrap();
234	3×	(st, String::from_utf8(b.to_vec()).unwrap())
235	6×	};
236	1×	let (st, lines) = post(
237	1×	AddCheckpoint {
238	1×	old_size: 0,
239	1×	proof: vec![],
240	1×	note: note(4),
241	1×	}
242	1×	.encode(),
243	1×	)
244	1×	.await;
245	1×	assert_eq!(st, StatusCode::OK, "{lines}");
246	1×	let cosigned =
247	1×	Note::parse(&format!("{}\n{lines}", Note::parse(&note(4)).unwrap().text)).unwrap();
248	1×	assert!(cosigned.cosigned_by("witness.example", &w).is_ok());
249		// Kept across a restart (a new router): a wrong old size is told.
250	1×	let (st, size) = post(
251	1×	AddCheckpoint {
252	1×	old_size: 0,
253	1×	proof: vec![],
254	1×	note: note(9),
255	1×	}
256	1×	.encode(),
257	1×	)
258	1×	.await;
259	1×	assert_eq!((st, size.trim()), (StatusCode::CONFLICT, "4"));
260	1×	let proof = tlog::prove_consistency(9, 4, &stored).unwrap();
261	1×	let (st, _) = post(
262	1×	AddCheckpoint {
263	1×	old_size: 4,
264	1×	proof,
265	1×	note: note(9),
266	1×	}
267	1×	.encode(),
268	1×	)
269	1×	.await;
270	1×	assert_eq!(st, StatusCode::OK);
271	1×	}
272		}

Coverage / crates/oxy-server/src/witness.rs 79.8%

Keyboard